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

mojo/public/cpp/bindings/lib/multiplex_router.cc

 // Copyright 2015 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 "mojo/public/cpp/bindings/lib/multiplex_router.h"
 
 #include <stdint.h>
 
 #include <utility>
 
@@ skipping 19 matching lines @@
 // this object.
 class MultiplexRouter::InterfaceEndpoint
     : public base::RefCounted<InterfaceEndpoint>,
       public InterfaceEndpointController {
  public:
   InterfaceEndpoint(MultiplexRouter* router, InterfaceId id)
       : router_(router),
         id_(id),
         closed_(false),
         peer_closed_(false),
+        handle_created_(false),
         client_(nullptr),
         event_signalled_(false) {}
 
   // ---------------------------------------------------------------------------
   // The following public methods are safe to call from any threads without
   // locking.
 
   InterfaceId id() const { return id_; }
 
   // ---------------------------------------------------------------------------
   // The following public methods are called under the router's lock.
 
   bool closed() const { return closed_; }
   void set_closed() {
     router_->AssertLockAcquired();
     closed_ = true;
   }
 
   bool peer_closed() const { return peer_closed_; }
   void set_peer_closed() {
     router_->AssertLockAcquired();
     peer_closed_ = true;
   }
 
+  bool handle_created() const { return handle_created_; }
+  void set_handle_created() {
+    router_->AssertLockAcquired();
+    handle_created_ = true;
+  }
+
   const base::Optional<DisconnectReason>& disconnect_reason() const {
     return disconnect_reason_;
   }
   void set_disconnect_reason(
       const base::Optional<DisconnectReason>& disconnect_reason) {
     router_->AssertLockAcquired();
     disconnect_reason_ = disconnect_reason;
   }
 
   base::SingleThreadTaskRunner* task_runner() const {
@@ skipping 53 matching lines @@
     event_signalled_ = false;
   }
 
   // ---------------------------------------------------------------------------
   // The following public methods (i.e., InterfaceEndpointController
   // implementation) are called by the client on the same thread as the
   // AttachClient() call. They are called outside of the router's lock.
 
   bool SendMessage(Message* message) override {
     DCHECK(task_runner_->BelongsToCurrentThread());
+    message->SerializeAssociatedEndpointHandles(router_);
     message->set_interface_id(id_);
     return router_->connector_.Accept(message);
   }
 
   void AllowWokenUpBySyncWatchOnSameThread() override {
     DCHECK(task_runner_->BelongsToCurrentThread());
 
     EnsureSyncWatcherExists();
     sync_watcher_->AllowWokenUpBySyncWatchOnSameThread();
   }
@@ skipping 83 matching lines @@
   const InterfaceId id_;
 
   // ---------------------------------------------------------------------------
   // The following members are accessed under the router's lock.
 
   // Whether the endpoint has been closed.
   bool closed_;
   // Whether the peer endpoint has been closed.
   bool peer_closed_;
 
+  // Whether there is already a ScopedInterfaceEndpointHandle created for this
+  // endpoint.
+  bool handle_created_;
+
   base::Optional<DisconnectReason> disconnect_reason_;
 
   // The task runner on which |client_|'s methods can be called.
   scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
   // Not owned. It is null if no client is attached to this endpoint.
   InterfaceEndpointClient* client_;
 
   // A message pipe used as an event to signal that sync messages are available.
   // The message pipe handles are initialized under the router's lock and remain
   // unchanged afterwards. They may be accessed outside of the router's lock
   // later.
   ScopedMessagePipeHandle sync_message_event_sender_;
   ScopedMessagePipeHandle sync_message_event_receiver_;
   bool event_signalled_;
 
   // ---------------------------------------------------------------------------
   // The following members are only valid while a client is attached. They are
   // used exclusively on the client's thread. They may be accessed outside of
   // the router's lock.
 
   std::unique_ptr<SyncHandleWatcher> sync_watcher_;
 
   DISALLOW_COPY_AND_ASSIGN(InterfaceEndpoint);
 };
 
+// Message objects cannot be destroyed under the router'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 router's lock. When a wrapper is
+// destroyed and the message hasn't been consumed, the wrapper is responsible
+// to send endpoint closed notifications.
+class MultiplexRouter::IncomingMessageWrapper {
+ public:
+  IncomingMessageWrapper() = default;
+
+  IncomingMessageWrapper(MultiplexRouter* router, Message* message)
+      : router_(router), value_(std::move(*message)) {
+    DCHECK(value_.associated_endpoint_handles()->empty());
+  }
+
+  IncomingMessageWrapper(IncomingMessageWrapper&& other)
+      : router_(other.router_), value_(std::move(other.value_)) {}
+
+  ~IncomingMessageWrapper() {
+    if (value_.IsNull())
+      return;
+
+    router_->AssertLockAcquired();
+
+    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) {
+      MayAutoUnlock unlocker(router_->lock_.get());
+      router_->control_message_proxy_.NotifyPeerEndpointClosed(ids[i],
+                                                               base::nullopt);
+    }
+  }
+
+  IncomingMessageWrapper& operator=(IncomingMessageWrapper&& other) {
+    router_ = other.router_;
+    value_ = std::move(other.value_);
+    return *this;
+  }
+
+  // Must be called outside of the router's lock.
+  bool TakeMessage(Message* output) {
+    DCHECK(!value_.IsNull());
+
+    *output = std::move(value_);
+    return output->DeserializeAssociatedEndpointHandles(router_);
+  }
+
+  const Message& value() const { return value_; }
+
+ private:
+  MultiplexRouter* router_ = nullptr;
+  // It must not hold any ScopedInterfaceEndpointHandle objects.
+  Message value_;
+
+  DISALLOW_COPY_AND_ASSIGN(IncomingMessageWrapper);
+};
+
 struct MultiplexRouter::Task {
  public:
   // Doesn't take ownership of |message| but takes its contents.
-  static std::unique_ptr<Task> CreateMessageTask(Message* message) {
+  static std::unique_ptr<Task> CreateMessageTask(
+      IncomingMessageWrapper message_wrapper) {
     Task* task = new Task(MESSAGE);
-    task->message = std::move(*message);
+    task->message_wrapper = std::move(message_wrapper);
     return base::WrapUnique(task);
   }
   static std::unique_ptr<Task> CreateNotifyErrorTask(
       InterfaceEndpoint* endpoint) {
     Task* task = new Task(NOTIFY_ERROR);
     task->endpoint_to_notify = endpoint;
     return base::WrapUnique(task);
   }
 
   ~Task() {}
 
   bool IsMessageTask() const { return type == MESSAGE; }
   bool IsNotifyErrorTask() const { return type == NOTIFY_ERROR; }
 
-  Message message;
+  IncomingMessageWrapper message_wrapper;
   scoped_refptr<InterfaceEndpoint> endpoint_to_notify;
 
   enum Type { MESSAGE, NOTIFY_ERROR };
   Type type;
 
  private:
   explicit Task(Type in_type) : type(in_type) {}
+
+  DISALLOW_COPY_AND_ASSIGN(Task);
 };
 
 MultiplexRouter::MultiplexRouter(
     ScopedMessagePipeHandle message_pipe,
     Config config,
     bool set_interface_id_namesapce_bit,
     scoped_refptr<base::SingleThreadTaskRunner> runner)
     : set_interface_id_namespace_bit_(set_interface_id_namesapce_bit),
       task_runner_(runner),
       header_validator_(nullptr),
@@ skipping 91 matching lines @@
 
 ScopedInterfaceEndpointHandle MultiplexRouter::CreateLocalEndpointHandle(
     InterfaceId id) {
   if (!IsValidInterfaceId(id))
     return ScopedInterfaceEndpointHandle();
 
   MayAutoLock locker(lock_.get());
   bool inserted = false;
   InterfaceEndpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
   if (inserted) {
+    DCHECK(!endpoint->handle_created());
+
     if (encountered_error_)
       UpdateEndpointStateMayRemove(endpoint, PEER_ENDPOINT_CLOSED);
   } else {
     // If the endpoint already exist, it is because we have received a
     // notification that the peer endpoint has closed.
     CHECK(!endpoint->closed());
     CHECK(endpoint->peer_closed());
+
+    if (endpoint->handle_created())
+      return ScopedInterfaceEndpointHandle();
   }
+
+  endpoint->set_handle_created();
   return CreateScopedInterfaceEndpointHandle(id, true);
 }
 
 void MultiplexRouter::CloseEndpointHandle(
     InterfaceId id,
     bool is_local,
     const base::Optional<DisconnectReason>& reason) {
   if (!IsValidInterfaceId(id))
     return;
 
@@ skipping 125 matching lines @@
 }
 
 bool MultiplexRouter::Accept(Message* message) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   scoped_refptr<MultiplexRouter> protector(this);
   MayAutoLock locker(lock_.get());
 
   DCHECK(!paused_);
 
+  IncomingMessageWrapper message_wrapper(this, message);
+
   ClientCallBehavior client_call_behavior =
       connector_.during_sync_handle_watcher_callback()
           ? ALLOW_DIRECT_CLIENT_CALLS_FOR_SYNC_MESSAGES
           : ALLOW_DIRECT_CLIENT_CALLS;
 
-  bool processed =
-      tasks_.empty() && ProcessIncomingMessage(message, client_call_behavior,
-                                               connector_.task_runner());
+  bool processed = tasks_.empty() && ProcessIncomingMessage(
+                                         &message_wrapper, client_call_behavior,
+                                         connector_.task_runner());
 
   if (!processed) {
     // Either the task queue is not empty or we cannot process the message
     // directly. In both cases, there is no need to call ProcessTasks().
-    tasks_.push_back(Task::CreateMessageTask(message));
+    tasks_.push_back(Task::CreateMessageTask(std::move(message_wrapper)));
     Task* task = tasks_.back().get();
 
-    if (task->message.has_flag(Message::kFlagIsSync)) {
-      InterfaceId id = task->message.interface_id();
+    if (task->message_wrapper.value().has_flag(Message::kFlagIsSync)) {
+      InterfaceId id = task->message_wrapper.value().interface_id();
       sync_message_tasks_[id].push_back(task);
       auto iter = endpoints_.find(id);
       if (iter != endpoints_.end())
         iter->second->SignalSyncMessageEvent();
     }
   } else if (!tasks_.empty()) {
     // Processing the message may result in new tasks (for error notification)
     // being added to the queue. In this case, we have to attempt to process the
     // tasks.
     ProcessTasks(client_call_behavior, connector_.task_runner());
   }
 
   // Always return true. If we see errors during message processing, we will
   // explicitly call Connector::RaiseError() to disconnect the message pipe.
   return true;
 }
 
 bool MultiplexRouter::OnPeerAssociatedEndpointClosed(
     InterfaceId id,
     const base::Optional<DisconnectReason>& reason) {
-  AssertLockAcquired();
-
   DCHECK(!IsMasterInterfaceId(id) || reason);
 
+  MayAutoLock locker(lock_.get());
   InterfaceEndpoint* endpoint = FindOrInsertEndpoint(id, nullptr);
 
   if (reason)
     endpoint->set_disconnect_reason(reason);
 
   // It is possible that this endpoint has been set as peer closed. That is
   // because when the message pipe is closed, all the endpoints are updated with
   // PEER_ENDPOINT_CLOSED. We continue to process remaining tasks in the queue,
   // as long as there are refs keeping the router alive. If there is a
   // PeerAssociatedEndpointClosedEvent control message in the queue, we will get
   // here and see that the endpoint has been marked as peer closed.
   if (!endpoint->peer_closed()) {
     if (endpoint->client())
       tasks_.push_back(Task::CreateNotifyErrorTask(endpoint));
     UpdateEndpointStateMayRemove(endpoint, PEER_ENDPOINT_CLOSED);
   }
 
   // No need to trigger a ProcessTasks() because it is already on the stack.
 
   return true;
 }
 
 bool MultiplexRouter::OnAssociatedEndpointClosedBeforeSent(InterfaceId id) {
-  AssertLockAcquired();
-
   if (IsMasterInterfaceId(id))
     return false;
 
-  InterfaceEndpoint* endpoint = FindOrInsertEndpoint(id, nullptr);
-  DCHECK(!endpoint->closed());
-  UpdateEndpointStateMayRemove(endpoint, ENDPOINT_CLOSED);
+  {
+    MayAutoLock locker(lock_.get());
 
-  MayAutoUnlock unlocker(lock_.get());
+    InterfaceEndpoint* endpoint = FindOrInsertEndpoint(id, nullptr);
+    DCHECK(!endpoint->closed());
+    UpdateEndpointStateMayRemove(endpoint, ENDPOINT_CLOSED);
+  }
+
   control_message_proxy_.NotifyPeerEndpointClosed(id, base::nullopt);
-
   return true;
 }
 
 void MultiplexRouter::OnPipeConnectionError() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   scoped_refptr<MultiplexRouter> protector(this);
   MayAutoLock locker(lock_.get());
 
   encountered_error_ = true;
@@ skipping 22 matching lines @@
   AssertLockAcquired();
 
   if (posted_to_process_tasks_)
     return;
 
   while (!tasks_.empty() && !paused_) {
     std::unique_ptr<Task> task(std::move(tasks_.front()));
     tasks_.pop_front();
 
     InterfaceId id = kInvalidInterfaceId;
-    bool sync_message = task->IsMessageTask() && !task->message.IsNull() &&
-                        task->message.has_flag(Message::kFlagIsSync);
+    bool sync_message =
+        task->IsMessageTask() && !task->message_wrapper.value().IsNull() &&
+        task->message_wrapper.value().has_flag(Message::kFlagIsSync);
     if (sync_message) {
-      id = task->message.interface_id();
+      id = task->message_wrapper.value().interface_id();
       auto& sync_message_queue = sync_message_tasks_[id];
       DCHECK_EQ(task.get(), sync_message_queue.front());
       sync_message_queue.pop_front();
     }
 
     bool processed =
         task->IsNotifyErrorTask()
             ? ProcessNotifyErrorTask(task.get(), client_call_behavior,
                                      current_task_runner)
-            : ProcessIncomingMessage(&task->message, client_call_behavior,
-                                     current_task_runner);
+            : ProcessIncomingMessage(&task->message_wrapper,
+                                     client_call_behavior, current_task_runner);
 
     if (!processed) {
       if (sync_message) {
         auto& sync_message_queue = sync_message_tasks_[id];
         sync_message_queue.push_front(task.get());
       }
       tasks_.push_front(std::move(task));
       break;
     } else {
       if (sync_message) {
@@ skipping 12 matching lines @@
   if (iter == sync_message_tasks_.end())
     return false;
 
   if (paused_)
     return true;
 
   MultiplexRouter::Task* task = iter->second.front();
   iter->second.pop_front();
 
   DCHECK(task->IsMessageTask());
-  Message message = std::move(task->message);
+  IncomingMessageWrapper message_wrapper = std::move(task->message_wrapper);
 
   // Note: after this call, |task| and |iter| may be invalidated.
   bool processed = ProcessIncomingMessage(
-      &message, ALLOW_DIRECT_CLIENT_CALLS_FOR_SYNC_MESSAGES, nullptr);
+      &message_wrapper, ALLOW_DIRECT_CLIENT_CALLS_FOR_SYNC_MESSAGES, nullptr);
   DCHECK(processed);
 
   iter = sync_message_tasks_.find(id);
   if (iter == sync_message_tasks_.end())
     return false;
 
   if (iter->second.empty()) {
     sync_message_tasks_.erase(iter);
     return false;
   }
@@ skipping 31 matching lines @@
     //
     // It is safe to call into |client| without the lock. Because |client| is
     // always accessed on the same thread, including DetachEndpointClient().
     MayAutoUnlock unlocker(lock_.get());
     client->NotifyError(disconnect_reason);
   }
   return true;
 }
 
 bool MultiplexRouter::ProcessIncomingMessage(
-    Message* message,
+    IncomingMessageWrapper* message_wrapper,
     ClientCallBehavior client_call_behavior,
     base::SingleThreadTaskRunner* current_task_runner) {
   DCHECK(!current_task_runner || current_task_runner->BelongsToCurrentThread());
   DCHECK(!paused_);
-  DCHECK(message);
+  DCHECK(message_wrapper);
   AssertLockAcquired();
 
-  if (message->IsNull()) {
+  if (message_wrapper->value().IsNull()) {
     // This is a sync message and has been processed during sync handle
     // watching.
     return true;
   }
 
-  if (PipeControlMessageHandler::IsPipeControlMessage(message)) {
-    if (!control_message_handler_.Accept(message))
+  if (PipeControlMessageHandler::IsPipeControlMessage(
+          &message_wrapper->value())) {
+    bool result = false;
+
+    {
+      MayAutoUnlock unlocker(lock_.get());
+      Message message;
+      result = message_wrapper->TakeMessage(&message) &&
+               control_message_handler_.Accept(&message);
+    }
+
+    if (!result)
       RaiseErrorInNonTestingMode();
+
     return true;
   }
 
-  InterfaceId id = message->interface_id();
+  InterfaceId id = message_wrapper->value().interface_id();
   DCHECK(IsValidInterfaceId(id));
 
   bool inserted = false;
   InterfaceEndpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
   if (inserted) {
     // Currently, it is legitimate to receive messages for an endpoint
     // that is not registered. For example, the endpoint is transferred in
     // a message that is discarded. Once we add support to specify all
     // enclosing endpoints in message header, we should be able to remove
     // this.
@@ skipping 16 matching lines @@
   if (endpoint->closed())
     return true;
 
   if (!endpoint->client()) {
     // We need to wait until a client is attached in order to dispatch further
     // messages.
     return false;
   }
 
   bool can_direct_call;
-  if (message->has_flag(Message::kFlagIsSync)) {
+  if (message_wrapper->value().has_flag(Message::kFlagIsSync)) {
     can_direct_call = client_call_behavior != NO_DIRECT_CLIENT_CALLS &&
                       endpoint->task_runner()->BelongsToCurrentThread();
   } else {
     can_direct_call = client_call_behavior == ALLOW_DIRECT_CLIENT_CALLS &&
                       endpoint->task_runner() == current_task_runner;
   }
 
   if (!can_direct_call) {
     MaybePostToProcessTasks(endpoint->task_runner());
     return false;
   }
 
   DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
 
   InterfaceEndpointClient* client = endpoint->client();
   bool result = false;
   {
     // We must unlock before calling into |client| because it may call this
     // object within HandleIncomingMessage(). Holding the lock will lead to
     // deadlock.
     //
     // It is safe to call into |client| without the lock. Because |client| is
     // always accessed on the same thread, including DetachEndpointClient().
     MayAutoUnlock unlocker(lock_.get());
-    result = client->HandleIncomingMessage(message);
+    Message message;
+    result = message_wrapper->TakeMessage(&message) &&
+             client->HandleIncomingMessage(&message);
   }
   if (!result)
     RaiseErrorInNonTestingMode();
 
   return true;
 }
 
 void MultiplexRouter::MaybePostToProcessTasks(
     base::SingleThreadTaskRunner* task_runner) {
   AssertLockAcquired();
@@ skipping 64 matching lines @@
 
 void MultiplexRouter::AssertLockAcquired() {
 #if DCHECK_IS_ON()
   if (lock_)
     lock_->AssertAcquired();
 #endif
 }
 
 }  // namespace internal
 }  // namespace mojo