Mojo C++ bindings: fix MultiplexRouter and ChannelAssociatedGroupController.

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 89 matching lines @@
       sender_id = 1;
       receiver_id = 1 | mojo::kInterfaceIdNamespaceMask;
     }
 
     {
       base::AutoLock locker(lock_);
       Endpoint* sender_endpoint = new Endpoint(this, sender_id);
       Endpoint* receiver_endpoint = new Endpoint(this, receiver_id);
       endpoints_.insert({ sender_id, sender_endpoint });
       endpoints_.insert({ receiver_id, receiver_endpoint });
+      sender_endpoint->set_handle_created();
+      receiver_endpoint->set_handle_created();
     }
 
     mojo::ScopedInterfaceEndpointHandle sender_handle =
         CreateScopedInterfaceEndpointHandle(sender_id, true);
     mojo::ScopedInterfaceEndpointHandle receiver_handle =
         CreateScopedInterfaceEndpointHandle(receiver_id, true);
 
     sender->Bind(mojom::ChannelAssociatedPtrInfo(std::move(sender_handle), 0));
     receiver->Bind(std::move(receiver_handle));
   }
@@ skipping 17 matching lines @@
       id = next_interface_id_++;
       if (set_interface_id_namespace_bit_)
         id |= mojo::kInterfaceIdNamespaceMask;
     } while (ContainsKey(endpoints_, id));
 
     Endpoint* endpoint = new Endpoint(this, id);
     if (encountered_error_)
       endpoint->set_peer_closed();
     endpoints_.insert({ id, endpoint });
 
+    endpoint->set_handle_created();
     *local_endpoint = CreateScopedInterfaceEndpointHandle(id, true);
     *remote_endpoint = CreateScopedInterfaceEndpointHandle(id, false);
   }
 
   mojo::ScopedInterfaceEndpointHandle CreateLocalEndpointHandle(
       mojo::InterfaceId id) override {
     if (!mojo::IsValidInterfaceId(id))
       return mojo::ScopedInterfaceEndpointHandle();
 
     base::AutoLock locker(lock_);
@@ skipping 83 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 {
+  // MessageWrapper objects are always destroyed under the controller's lock. On
+  // destruction, if the message it wrappers contains
+  // ScopedInterfaceEndpointHandles (which cannot be destructed under the
+  // controller's lock), the wrapper unlocks to clean them up.
+  class MessageWrapper {
    public:
-    IncomingMessageWrapper() = default;
+    MessageWrapper() = default;
 
-    IncomingMessageWrapper(ChannelAssociatedGroupController* controller,
-                           mojo::Message* message)
-        : controller_(controller), value_(std::move(*message)) {
-      DCHECK(value_.associated_endpoint_handles()->empty());
-    }
+    MessageWrapper(ChannelAssociatedGroupController* controller,
+                   mojo::Message message)
+        : controller_(controller), value_(std::move(message)) {}
 
-    IncomingMessageWrapper(IncomingMessageWrapper&& other)
+    MessageWrapper(MessageWrapper&& other)
         : controller_(other.controller_), value_(std::move(other.value_)) {}
 
-    ~IncomingMessageWrapper() {
-      if (value_.IsNull())
+    ~MessageWrapper() {
+      if (value_.associated_endpoint_handles()->empty())
         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);
+        value_.mutable_associated_endpoint_handles()->clear();
       }
     }
 
-    IncomingMessageWrapper& operator=(IncomingMessageWrapper&& other) {
+    MessageWrapper& operator=(MessageWrapper&& 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_; }
+    mojo::Message& value() { return value_; }
 
    private:
     ChannelAssociatedGroupController* controller_ = nullptr;
-    // It must not hold any ScopedInterfaceEndpointHandle objects.
     mojo::Message value_;
 
-    DISALLOW_COPY_AND_ASSIGN(IncomingMessageWrapper);
+    DISALLOW_COPY_AND_ASSIGN(MessageWrapper);
   };
 
   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_; }
 
@@ skipping 60 matching lines @@
       controller_->lock_.AssertAcquired();
       DCHECK(client_);
       DCHECK(task_runner_->BelongsToCurrentThread());
       DCHECK(!closed_);
 
       task_runner_ = nullptr;
       client_ = nullptr;
       sync_watcher_.reset();
     }
 
-    uint32_t EnqueueSyncMessage(IncomingMessageWrapper message) {
+    uint32_t EnqueueSyncMessage(MessageWrapper 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();
     }
 
-    IncomingMessageWrapper PopSyncMessage(uint32_t id) {
+    MessageWrapper PopSyncMessage(uint32_t id) {
       controller_->lock_.AssertAcquired();
       if (sync_messages_.empty() || sync_messages_.front().first != id)
-        return IncomingMessageWrapper();
-      IncomingMessageWrapper message = std::move(sync_messages_.front().second);
+        return MessageWrapper();
+      MessageWrapper 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);
@@ skipping 34 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()) {
-          IncomingMessageWrapper message_wrapper =
+          MessageWrapper message_wrapper =
               std::move(sync_messages_.front().second);
           sync_messages_.pop();
 
           bool dispatch_succeeded;
           mojo::InterfaceEndpointClient* client = client_;
           {
             base::AutoUnlock unlocker(controller_->lock_);
-            mojo::Message message;
-            dispatch_succeeded = message_wrapper.TakeMessage(&message) &&
-                                 client->HandleIncomingMessage(&message);
+            dispatch_succeeded =
+                client->HandleIncomingMessage(&message_wrapper.value());
           }
 
           if (!sync_messages_.empty())
             more_to_process = true;
 
           if (!dispatch_succeeded)
             controller_->RaiseError();
         }
 
         if (!more_to_process)
@@ skipping 47 matching lines @@
     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, IncomingMessageWrapper>> sync_messages_;
+    std::queue<std::pair<uint32_t, MessageWrapper>> 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) {}
@@ skipping 133 matching lines @@
     endpoint->set_peer_closed();
     endpoint->SignalSyncMessageEvent();
     if (endpoint->closed() && endpoint->peer_closed())
       endpoints_.erase(endpoint->id());
   }
 
   Endpoint* FindOrInsertEndpoint(mojo::InterfaceId id, bool* inserted) {
     lock_.AssertAcquired();
     DCHECK(!inserted || !*inserted);
 
+    Endpoint* endpoint = FindEndpoint(id);
+    if (!endpoint) {
+      endpoint = new Endpoint(this, id);
+      endpoints_.insert({id, endpoint});
+      if (inserted)
+        *inserted = true;
+    }
+    return endpoint;
+  }
+
+  Endpoint* FindEndpoint(mojo::InterfaceId id) {
+    lock_.AssertAcquired();
     auto iter = endpoints_.find(id);
-    if (iter != endpoints_.end())
-      return iter->second.get();
-
-    Endpoint* endpoint = new Endpoint(this, id);
-    endpoints_.insert({ id, endpoint });
-    if (inserted)
-      *inserted = true;
-    return endpoint;
+    return iter != endpoints_.end() ? iter->second.get() : nullptr;
   }
 
   // mojo::MessageReceiver:
   bool Accept(mojo::Message* message) override {
     DCHECK(thread_checker_.CalledOnValidThread());
 
-    if (mojo::PipeControlMessageHandler::IsPipeControlMessage(message)) {
-      return message->DeserializeAssociatedEndpointHandles(this) &&
-             control_message_handler_.Accept(message);
-    }
+    if (!message->DeserializeAssociatedEndpointHandles(this))
+      return false;
+
+    if (mojo::PipeControlMessageHandler::IsPipeControlMessage(message))
+      return 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;
+    Endpoint* endpoint = FindEndpoint(id);
+    if (!endpoint)
+      return true;
+
+    mojo::InterfaceEndpointClient* client = endpoint->client();
     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);
+        MessageWrapper message_wrapper(this, std::move(*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_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 message->DeserializeAssociatedEndpointHandles(this) &&
-           client->HandleIncomingMessage(message);
+    return 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 */);
+    Endpoint* endpoint = FindEndpoint(id);
     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_wrapper.value().has_flag(mojo::Message::kFlagIsSync));
+    DCHECK(!message.has_flag(mojo::Message::kFlagIsSync));
 
     bool result = false;
     {
       base::AutoUnlock unlocker(lock_);
-      mojo::Message message;
-      result = message_wrapper.TakeMessage(&message) &&
-               client->HandleIncomingMessage(&message);
+      result = 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 */);
+    Endpoint* endpoint = FindEndpoint(interface_id);
     if (!endpoint)
       return;
 
     DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
-    IncomingMessageWrapper message_wrapper =
-        endpoint->PopSyncMessage(message_id);
+    MessageWrapper 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_wrapper.value().IsNull())
       return;
 
     mojo::InterfaceEndpointClient* client = endpoint->client();
     if (!client)
       return;
 
     bool result = false;
     {
       base::AutoUnlock unlocker(lock_);
-      mojo::Message message;
-      result = message_wrapper.TakeMessage(&message) &&
-               client->HandleIncomingMessage(&message);
+      result = client->HandleIncomingMessage(&message_wrapper.value());
     }
 
     if (!result)
       RaiseError();
   }
 
-  Endpoint* GetEndpointForDispatch(mojo::InterfaceId id, bool create) {
-    lock_.AssertAcquired();
-    auto iter = endpoints_.find(id);
-    if (iter != endpoints_.end())
-      return iter->second.get();
-    if (!create)
-      return nullptr;
-    bool inserted = false;
-    Endpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
-    DCHECK(inserted);
-    return endpoint;
-  }
-
   // mojo::PipeControlMessageHandlerDelegate:
   bool OnPeerAssociatedEndpointClosed(
       mojo::InterfaceId id,
       const base::Optional<mojo::DisconnectReason>& reason) override {
     DCHECK(thread_checker_.CalledOnValidThread());
 
     DCHECK(!mojo::IsMasterInterfaceId(id) || reason);
 
     scoped_refptr<ChannelAssociatedGroupController> keepalive(this);
     base::AutoLock locker(lock_);
@@ skipping 120 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