Mojo C++ bindings: introduce an optional array to store transferred interface IDs in messages.

ipc/ipc_mojo_bootstrap.cc

 // Copyright 2014 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "ipc/ipc_mojo_bootstrap.h"
 
 #include <stdint.h>
 
 #include <map>
 #include <memory>
@@ skipping 138 matching lines @@
   }
 
   mojo::ScopedInterfaceEndpointHandle CreateLocalEndpointHandle(
       mojo::InterfaceId id) override {
     if (!mojo::IsValidInterfaceId(id))
       return mojo::ScopedInterfaceEndpointHandle();
 
     base::AutoLock locker(lock_);
     bool inserted = false;
     Endpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
-    if (inserted && encountered_error_)
-      endpoint->set_peer_closed();
+    if (inserted) {
+      DCHECK(!endpoint->handle_created());
+      if (encountered_error_)
+        endpoint->set_peer_closed();
+    } else {
+      if (endpoint->handle_created())
+        return mojo::ScopedInterfaceEndpointHandle();
+    }
 
+    endpoint->set_handle_created();
     return CreateScopedInterfaceEndpointHandle(id, true);
   }
 
   void CloseEndpointHandle(
       mojo::InterfaceId id,
       bool is_local,
       const base::Optional<mojo::DisconnectReason>& reason) override {
     if (!mojo::IsValidInterfaceId(id))
       return;
 
@@ skipping 61 matching lines @@
           base::Bind(&ChannelAssociatedGroupController::RaiseError, this));
     }
   }
 
  private:
   class Endpoint;
   class ControlMessageProxyThunk;
   friend class Endpoint;
   friend class ControlMessageProxyThunk;
 
+  // Message objects cannot be destroyed under the controller's lock, if they
+  // contain ScopedInterfaceEndpointHandle objects.
+  // IncomingMessageWrapper is used to wrap messages which haven't got the
+  // payload interface IDs deserialized into ScopedInterfaceEndpointHandles.
+  // Wrapper objects are always destroyed under the controller's lock. When a
+  // wrapper is destroyed and the message hasn't been consumed, the wrapper is
+  // responsible to send endpoint closed notifications.
+  class IncomingMessageWrapper {
+   public:
+    IncomingMessageWrapper() = default;
+
+    IncomingMessageWrapper(ChannelAssociatedGroupController* controller,
+                           mojo::Message* message)
+        : controller_(controller), value_(std::move(*message)) {
+      DCHECK(value_.associated_endpoint_handles()->empty());
+    }
+
+    IncomingMessageWrapper(IncomingMessageWrapper&& other)
+        : controller_(other.controller_), value_(std::move(other.value_)) {}
+
+    ~IncomingMessageWrapper() {
+      if (value_.IsNull())
+        return;
+
+      controller_->lock_.AssertAcquired();
+
+      uint32_t num_ids = value_.payload_num_interface_ids();
+      const uint32_t* ids = value_.payload_interface_ids();
+      for (uint32_t i = 0; i < num_ids; ++i) {
+        base::AutoUnlock unlocker(controller_->lock_);
+        controller_->control_message_proxy_.NotifyPeerEndpointClosed(
+            ids[i], base::nullopt);
+      }
+    }
+
+    IncomingMessageWrapper& operator=(IncomingMessageWrapper&& other) {
+      controller_ = other.controller_;
+      value_ = std::move(other.value_);
+      return *this;
+    }
+
+    // Must be called outside of the controller's lock.
+    bool TakeMessage(mojo::Message* output) {
+      DCHECK(!value_.IsNull());
+
+      *output = std::move(value_);
+      return output->DeserializeAssociatedEndpointHandles(controller_);
+    }
+
+    const mojo::Message& value() const { return value_; }
+
+   private:
+    ChannelAssociatedGroupController* controller_ = nullptr;
+    // It must not hold any ScopedInterfaceEndpointHandle objects.
+    mojo::Message value_;
+
+    DISALLOW_COPY_AND_ASSIGN(IncomingMessageWrapper);
+  };
+
   class Endpoint : public base::RefCountedThreadSafe<Endpoint>,
                    public mojo::InterfaceEndpointController {
    public:
     Endpoint(ChannelAssociatedGroupController* controller, mojo::InterfaceId id)
         : controller_(controller), id_(id) {}
 
     mojo::InterfaceId id() const { return id_; }
 
     bool closed() const {
       controller_->lock_.AssertAcquired();
       return closed_;
     }
 
     void set_closed() {
       controller_->lock_.AssertAcquired();
       closed_ = true;
     }
 
     bool peer_closed() const {
       controller_->lock_.AssertAcquired();
       return peer_closed_;
     }
 
     void set_peer_closed() {
       controller_->lock_.AssertAcquired();
       peer_closed_ = true;
     }
 
+    bool handle_created() const {
+      controller_->lock_.AssertAcquired();
+      return handle_created_;
+    }
+
+    void set_handle_created() {
+      controller_->lock_.AssertAcquired();
+      handle_created_ = true;
+    }
+
     const base::Optional<mojo::DisconnectReason>& disconnect_reason() const {
       return disconnect_reason_;
     }
 
     void set_disconnect_reason(
         const base::Optional<mojo::DisconnectReason>& disconnect_reason) {
       disconnect_reason_ = disconnect_reason;
     }
 
     base::SingleThreadTaskRunner* task_runner() const {
@@ skipping 20 matching lines @@
       controller_->lock_.AssertAcquired();
       DCHECK(client_);
       DCHECK(task_runner_->BelongsToCurrentThread());
       DCHECK(!closed_);
 
       task_runner_ = nullptr;
       client_ = nullptr;
       sync_watcher_.reset();
     }
 
-    uint32_t EnqueueSyncMessage(mojo::Message message) {
+    uint32_t EnqueueSyncMessage(IncomingMessageWrapper message) {
       controller_->lock_.AssertAcquired();
       uint32_t id = GenerateSyncMessageId();
       sync_messages_.emplace(id, std::move(message));
       SignalSyncMessageEvent();
       return id;
     }
 
     void SignalSyncMessageEvent() {
       controller_->lock_.AssertAcquired();
       EnsureSyncMessageEventExists();
       sync_message_event_->Signal();
     }
 
-    mojo::Message PopSyncMessage(uint32_t id) {
+    IncomingMessageWrapper PopSyncMessage(uint32_t id) {
       controller_->lock_.AssertAcquired();
       if (sync_messages_.empty() || sync_messages_.front().first != id)
-        return mojo::Message();
-      mojo::Message message = std::move(sync_messages_.front().second);
+        return IncomingMessageWrapper();
+      IncomingMessageWrapper message = std::move(sync_messages_.front().second);
       sync_messages_.pop();
       return message;
     }
 
     // mojo::InterfaceEndpointController:
     bool SendMessage(mojo::Message* message) override {
       DCHECK(task_runner_->BelongsToCurrentThread());
       message->set_interface_id(id_);
+      message->SerializeAssociatedEndpointHandles(controller_);
       return controller_->SendMessage(message);
     }
 
     void AllowWokenUpBySyncWatchOnSameThread() override {
       DCHECK(task_runner_->BelongsToCurrentThread());
 
       EnsureSyncWatcherExists();
       sync_watcher_->AllowWokenUpBySyncWatchOnSameThread();
     }
 
@@ skipping 25 matching lines @@
 
       scoped_refptr<Endpoint> keepalive(this);
       scoped_refptr<AssociatedGroupController> controller_keepalive(
           controller_);
 
       bool reset_sync_watcher = false;
       {
         base::AutoLock locker(controller_->lock_);
         bool more_to_process = false;
         if (!sync_messages_.empty()) {
-          mojo::Message message = std::move(sync_messages_.front().second);
+          IncomingMessageWrapper message_wrapper =
+              std::move(sync_messages_.front().second);
           sync_messages_.pop();
 
           bool dispatch_succeeded;
           mojo::InterfaceEndpointClient* client = client_;
           {
             base::AutoUnlock unlocker(controller_->lock_);
-            dispatch_succeeded = client->HandleIncomingMessage(&message);
+            mojo::Message message;
+            dispatch_succeeded = message_wrapper.TakeMessage(&message) &&
+                                 client->HandleIncomingMessage(&message);
           }
 
           if (!sync_messages_.empty())
             more_to_process = true;
 
           if (!dispatch_succeeded)
             controller_->RaiseError();
         }
 
         if (!more_to_process)
@@ skipping 41 matching lines @@
       uint32_t id = next_sync_message_id_++;
       DCHECK(sync_messages_.empty() || sync_messages_.front().first != id);
       return id;
     }
 
     ChannelAssociatedGroupController* const controller_;
     const mojo::InterfaceId id_;
 
     bool closed_ = false;
     bool peer_closed_ = false;
+    bool handle_created_ = false;
     base::Optional<mojo::DisconnectReason> disconnect_reason_;
     mojo::InterfaceEndpointClient* client_ = nullptr;
     scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
     std::unique_ptr<mojo::SyncHandleWatcher> sync_watcher_;
     std::unique_ptr<MojoEvent> sync_message_event_;
-    std::queue<std::pair<uint32_t, mojo::Message>> sync_messages_;
+    std::queue<std::pair<uint32_t, IncomingMessageWrapper>> sync_messages_;
     uint32_t next_sync_message_id_ = 0;
 
     DISALLOW_COPY_AND_ASSIGN(Endpoint);
   };
 
   class ControlMessageProxyThunk : public MessageReceiver {
    public:
     explicit ControlMessageProxyThunk(
         ChannelAssociatedGroupController* controller)
         : controller_(controller) {}
 
    private:
     // MessageReceiver:
     bool Accept(mojo::Message* message) override {
+      message->SerializeAssociatedEndpointHandles(controller_);
       return controller_->SendMessage(message);
     }
 
     ChannelAssociatedGroupController* controller_;
 
     DISALLOW_COPY_AND_ASSIGN(ControlMessageProxyThunk);
   };
 
   ~ChannelAssociatedGroupController() override {
     DCHECK(!connector_);
@@ skipping 133 matching lines @@
     endpoints_.insert({ id, endpoint });
     if (inserted)
       *inserted = true;
     return endpoint;
   }
 
   // mojo::MessageReceiver:
   bool Accept(mojo::Message* message) override {
     DCHECK(thread_checker_.CalledOnValidThread());
 
-    if (mojo::PipeControlMessageHandler::IsPipeControlMessage(message))
-      return control_message_handler_.Accept(message);
+    if (mojo::PipeControlMessageHandler::IsPipeControlMessage(message)) {
+      return message->DeserializeAssociatedEndpointHandles(this) &&
+             control_message_handler_.Accept(message);
+    }
 
     mojo::InterfaceId id = message->interface_id();
     DCHECK(mojo::IsValidInterfaceId(id));
 
     base::AutoLock locker(lock_);
     Endpoint* endpoint = GetEndpointForDispatch(id, true /* create */);
     mojo::InterfaceEndpointClient* client =
         endpoint ? endpoint->client() : nullptr;
     if (!client || !endpoint->task_runner()->BelongsToCurrentThread()) {
       // No client has been bound yet or the client runs tasks on another
       // thread. We assume the other thread must always be the one on which
       // |proxy_task_runner_| runs tasks, since that's the only valid scenario.
       //
       // If the client is not yet bound, it must be bound by the time this task
       // runs or else it's programmer error.
       DCHECK(proxy_task_runner_);
 
       if (message->has_flag(mojo::Message::kFlagIsSync)) {
+        IncomingMessageWrapper message_wrapper(this, message);
         // Sync messages may need to be handled by the endpoint if it's blocking
         // on a sync reply. We pass ownership of the message to the endpoint's
         // sync message queue. If the endpoint was blocking, it will dequeue the
         // message and dispatch it. Otherwise the posted |AcceptSyncMessage()|
         // call will dequeue the message and dispatch it.
-        uint32_t message_id = endpoint->EnqueueSyncMessage(std::move(*message));
+        uint32_t message_id =
+            endpoint->EnqueueSyncMessage(std::move(message_wrapper));
         proxy_task_runner_->PostTask(
             FROM_HERE,
             base::Bind(&ChannelAssociatedGroupController::AcceptSyncMessage,
                        this, id, message_id));
         return true;
       }
 
       proxy_task_runner_->PostTask(
           FROM_HERE,
           base::Bind(&ChannelAssociatedGroupController::AcceptOnProxyThread,
                      this, base::Passed(message)));
       return true;
     }
 
     // We do not expect to receive sync responses on the master endpoint thread.
     // If it's happening, it's a bug.
     DCHECK(!message->has_flag(mojo::Message::kFlagIsSync) ||
            !message->has_flag(mojo::Message::kFlagIsResponse));
 
     base::AutoUnlock unlocker(lock_);
-    return client->HandleIncomingMessage(message);
+    return message->DeserializeAssociatedEndpointHandles(this) &&
+           client->HandleIncomingMessage(message);
   }
 
   void AcceptOnProxyThread(mojo::Message message) {
     DCHECK(proxy_task_runner_->BelongsToCurrentThread());
 
     mojo::InterfaceId id = message.interface_id();
     DCHECK(mojo::IsValidInterfaceId(id) && !mojo::IsMasterInterfaceId(id));
 
     base::AutoLock locker(lock_);
+    IncomingMessageWrapper message_wrapper(this, &message);
+
     Endpoint* endpoint = GetEndpointForDispatch(id, false /* create */);
     if (!endpoint)
       return;
 
     mojo::InterfaceEndpointClient* client = endpoint->client();
     if (!client)
       return;
 
     DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
 
     // Sync messages should never make their way to this method.
-    DCHECK(!message.has_flag(mojo::Message::kFlagIsSync));
+    DCHECK(!message_wrapper.value().has_flag(mojo::Message::kFlagIsSync));
 
     bool result = false;
     {
       base::AutoUnlock unlocker(lock_);
-      result = client->HandleIncomingMessage(&message);
+      mojo::Message message;
+      result = message_wrapper.TakeMessage(&message) &&
+               client->HandleIncomingMessage(&message);
     }
 
     if (!result)
       RaiseError();
   }
 
   void AcceptSyncMessage(mojo::InterfaceId interface_id, uint32_t message_id) {
     DCHECK(proxy_task_runner_->BelongsToCurrentThread());
 
     base::AutoLock locker(lock_);
     Endpoint* endpoint =
         GetEndpointForDispatch(interface_id, false /* create */);
     if (!endpoint)
       return;
 
     DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
-    mojo::Message message = endpoint->PopSyncMessage(message_id);
+    IncomingMessageWrapper message_wrapper =
+        endpoint->PopSyncMessage(message_id);
 
     // The message must have already been dequeued by the endpoint waking up
     // from a sync wait. Nothing to do.
-    if (message.IsNull())
+    if (message_wrapper.value().IsNull())
       return;
 
     mojo::InterfaceEndpointClient* client = endpoint->client();
     if (!client)
       return;
 
     bool result = false;
     {
       base::AutoUnlock unlocker(lock_);
-      result = client->HandleIncomingMessage(&message);
+      mojo::Message message;
+      result = message_wrapper.TakeMessage(&message) &&
+               client->HandleIncomingMessage(&message);
     }
 
     if (!result)
       RaiseError();
   }
 
   Endpoint* GetEndpointForDispatch(mojo::InterfaceId id, bool create) {
     lock_.AssertAcquired();
     auto iter = endpoints_.find(id);
     if (iter != endpoints_.end())
@@ skipping 139 matching lines @@
     Channel::Mode mode,
     Delegate* delegate,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner) {
   return base::MakeUnique<MojoBootstrapImpl>(
       std::move(handle), delegate,
       new ChannelAssociatedGroupController(mode == Channel::MODE_SERVER,
                                            ipc_task_runner));
 }
 
 }  // namespace IPC