Adds sync message support to Channel-associated interfaces.

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>
+#include <queue>
 #include <utility>
 #include <vector>
 
 #include "base/callback.h"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
 #include "base/single_thread_task_runner.h"
 #include "base/synchronization/lock.h"
 #include "base/threading/thread_task_runner_handle.h"
+#include "ipc/mojo_event.h"
 #include "mojo/public/cpp/bindings/associated_group.h"
 #include "mojo/public/cpp/bindings/associated_group_controller.h"
 #include "mojo/public/cpp/bindings/connector.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_client.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_controller.h"
 #include "mojo/public/cpp/bindings/interface_id.h"
 #include "mojo/public/cpp/bindings/message.h"
 #include "mojo/public/cpp/bindings/message_header_validator.h"
 #include "mojo/public/cpp/bindings/pipe_control_message_handler.h"
 #include "mojo/public/cpp/bindings/pipe_control_message_handler_delegate.h"
 #include "mojo/public/cpp/bindings/pipe_control_message_proxy.h"
+#include "mojo/public/cpp/bindings/sync_handle_watcher.h"
 
 namespace IPC {
 
 namespace {
 
 class ChannelAssociatedGroupController
     : public mojo::AssociatedGroupController,
       public mojo::MessageReceiver,
       public mojo::PipeControlMessageHandlerDelegate {
  public:
@@ skipping 224 matching lines @@
     }
 
     void DetachClient() {
       controller_->lock_.AssertAcquired();
       DCHECK(client_);
       DCHECK(task_runner_->BelongsToCurrentThread());
       DCHECK(!closed_);
 
       task_runner_ = nullptr;
       client_ = nullptr;
+      sync_watcher_.reset();
+    }
+
+    void EnqueueSyncMessage(std::unique_ptr<mojo::Message> message) {
+      controller_->lock_.AssertAcquired();
+      sync_messages_.emplace(std::move(message));
+      SignalSyncMessageEvent();
+    }
+
+    void SignalSyncMessageEvent() {
+      controller_->lock_.AssertAcquired();
+      EnsureSyncMessageEventExists();
+      sync_message_event_->Signal();
+    }
+
+    std::unique_ptr<mojo::Message> PopSyncMessage(mojo::Message* raw_message) {
+      controller_->lock_.AssertAcquired();
+      if (sync_messages_.empty() || sync_messages_.front().get() != raw_message)
+        return nullptr;
+      std::unique_ptr<mojo::Message> message =
+          std::move(sync_messages_.front());
+      sync_messages_.pop();
+      return message;
     }
 
     // mojo::InterfaceEndpointController:
     bool SendMessage(mojo::Message* message) override {
       DCHECK(task_runner_->BelongsToCurrentThread());
       message->set_interface_id(id_);
       return controller_->SendMessage(message);
     }
 
     void AllowWokenUpBySyncWatchOnSameThread() override {
       DCHECK(task_runner_->BelongsToCurrentThread());
 
-      // TODO(rockot): Implement sync waiting.
-      NOTREACHED();
+      EnsureSyncWatcherExists();
+      sync_watcher_->AllowWokenUpBySyncWatchOnSameThread();
     }
 
     bool SyncWatch(const bool* should_stop) override {
       DCHECK(task_runner_->BelongsToCurrentThread());
 
       // It's not legal to make sync calls from the master endpoint's thread,
       // and in fact they must only happen from the proxy task runner.
       DCHECK(!controller_->task_runner_->BelongsToCurrentThread());
       DCHECK(controller_->proxy_task_runner_->BelongsToCurrentThread());
 
-      // TODO(rockot): Implement sync waiting.
-      NOTREACHED();
-      return false;
+      EnsureSyncWatcherExists();
+      return sync_watcher_->SyncWatch(should_stop);
     }
 
    private:
     friend class base::RefCountedThreadSafe<Endpoint>;
 
-    ~Endpoint() override {}
+    ~Endpoint() override {
+      controller_->lock_.AssertAcquired();
+      DCHECK(!client_);
+      DCHECK(closed_);
+      DCHECK(peer_closed_);
+      DCHECK(!sync_watcher_);
+    }
+
+    void OnSyncMessageEventHandleReady(MojoResult result) {
+      DCHECK(task_runner_->BelongsToCurrentThread());
+
+      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()) {
+          std::unique_ptr<mojo::Message> message(
+              std::move(sync_messages_.front()));
+          sync_messages_.pop();
+
+          bool dispatch_succeeded;
+          mojo::InterfaceEndpointClient* client = client_;
+          {
+            base::AutoUnlock unlocker(controller_->lock_);
+            dispatch_succeeded = client->HandleIncomingMessage(message.get());
+          }
+
+          if (!sync_messages_.empty())
+            more_to_process = true;
+
+          if (!dispatch_succeeded)
+            controller_->RaiseError();
+        }
+
+        if (!more_to_process)
+          sync_message_event_->Reset();
+
+        // If there are no queued sync messages and the peer has closed, there
+        // there won't be incoming sync messages in the future.
+        reset_sync_watcher = !more_to_process && peer_closed_;
+      }
+
+      if (reset_sync_watcher) {
+        // If a SyncWatch() call (or multiple ones) of this interface endpoint
+        // is on the call stack, resetting the sync watcher will allow it to
+        // exit when the call stack unwinds to that frame.
+        sync_watcher_.reset();
+      }
+    }
+
+    void EnsureSyncWatcherExists() {
+      DCHECK(task_runner_->BelongsToCurrentThread());
+      if (sync_watcher_)
+        return;
+
+      {
+        base::AutoLock locker(controller_->lock_);
+        EnsureSyncMessageEventExists();
+        if (!sync_messages_.empty())
+          SignalSyncMessageEvent();
+      }
+
+      sync_watcher_.reset(new mojo::SyncHandleWatcher(
+          sync_message_event_->GetHandle(), MOJO_HANDLE_SIGNAL_READABLE,
+          base::Bind(&Endpoint::OnSyncMessageEventHandleReady,
+                     base::Unretained(this))));
+    }
+
+    void EnsureSyncMessageEventExists() {
+      controller_->lock_.AssertAcquired();
+      if (!sync_message_event_)
+        sync_message_event_.reset(new MojoEvent);
+    }
 
     ChannelAssociatedGroupController* const controller_;
     const mojo::InterfaceId id_;
 
     bool closed_ = false;
     bool peer_closed_ = false;
     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::unique_ptr<mojo::Message>> sync_messages_;
 
     DISALLOW_COPY_AND_ASSIGN(Endpoint);
   };
 
   class ControlMessageProxyThunk : public MessageReceiver {
    public:
     explicit ControlMessageProxyThunk(
         ChannelAssociatedGroupController* controller)
         : controller_(controller) {}
 
@@ skipping 89 matching lines @@
     DCHECK(endpoint->task_runner() && endpoint->client());
     if (endpoint->task_runner()->BelongsToCurrentThread() && !force_async) {
       mojo::InterfaceEndpointClient* client = endpoint->client();
 
       base::AutoUnlock unlocker(lock_);
       client->NotifyError();
     } else {
       endpoint->task_runner()->PostTask(
           FROM_HERE,
           base::Bind(&ChannelAssociatedGroupController
-                ::NotifyEndpointOfErrorOnEndpointThread, this,
-                make_scoped_refptr(endpoint)));
+                ::NotifyEndpointOfErrorOnEndpointThread, this, endpoint->id(),
+                endpoint));
     }
   }
 
-  void NotifyEndpointOfErrorOnEndpointThread(scoped_refptr<Endpoint> endpoint) {
+  void NotifyEndpointOfErrorOnEndpointThread(mojo::InterfaceId id,
+                                             Endpoint* endpoint) {
     base::AutoLock locker(lock_);
+    auto iter = endpoints_.find(id);
+    if (iter == endpoints_.end() || iter->second.get() != endpoint)
+      return;
     if (!endpoint->client())
       return;
+
     DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
-    NotifyEndpointOfError(endpoint.get(), false /* force_async */);
+    NotifyEndpointOfError(endpoint, false /* force_async */);
   }
 
   void MarkClosedAndMaybeRemove(Endpoint* endpoint) {
     lock_.AssertAcquired();
     endpoint->set_closed();
     if (endpoint->closed() && endpoint->peer_closed())
       endpoints_.erase(endpoint->id());
   }
 
   void MarkPeerClosedAndMaybeRemove(Endpoint* endpoint) {
     lock_.AssertAcquired();
     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);
 
     auto iter = endpoints_.find(id);
     if (iter != endpoints_.end())
@@ skipping 23 matching lines @@
 
     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_);
+
       std::unique_ptr<mojo::Message> passed_message(new mojo::Message);
       message->MoveTo(passed_message.get());
+
+      if (passed_message->has_flag(mojo::Message::kFlagIsSync)) {
+        // 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.
+        mojo::Message* raw_message = passed_message.get();
+        endpoint->EnqueueSyncMessage(std::move(passed_message));
+        proxy_task_runner_->PostTask(
+            FROM_HERE,
+            base::Bind(&ChannelAssociatedGroupController::AcceptSyncMessage,
+                       this, id, raw_message));

[yzshen1, 2016/08/02 17:38:37]

Raw message pointer may not be an appropriate ID: it is possible to create a
message at the same address as a message that is already gone.

[Ken Rockot(use gerrit already), 2016/08/02 18:21:24]

I've switched to using sequential integers. Overflow shouldn't be a problem
since it's only possible if we accumulate very long message/task queues (and
then we have other problems!) WDYT?

[yzshen1, 2016/08/02 18:34:11]

SGTM

+        return true;
+      }
+
       proxy_task_runner_->PostTask(
           FROM_HERE,
           base::Bind(&ChannelAssociatedGroupController::AcceptOnProxyThread,
                      this, base::Passed(&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));
+    DCHECK(!message->has_flag(mojo::Message::kFlagIsSync) ||
+           !message->has_flag(mojo::Message::kFlagIsResponse));
 
     base::AutoUnlock unlocker(lock_);
     return client->HandleIncomingMessage(message);
   }
 
   void AcceptOnProxyThread(std::unique_ptr<mojo::Message> message) {
     DCHECK(proxy_task_runner_->BelongsToCurrentThread());
 
     mojo::InterfaceId id = message->interface_id();
     DCHECK(mojo::IsValidInterfaceId(id) && !mojo::IsMasterInterfaceId(id));
 
+    scoped_refptr<ChannelAssociatedGroupController> keepalive(this);

[yzshen1, 2016/08/02 17:38:37]

I think this line is not needed because the base::Callback holds on to a ref of
|this|? (and also line 672)

[Ken Rockot(use gerrit already), 2016/08/02 18:21:24]

Done

     base::AutoLock locker(lock_);
     Endpoint* endpoint = GetEndpointForDispatch(id);
     if (!endpoint)
       return;
 
     mojo::InterfaceEndpointClient* client = endpoint->client();
     if (!client)
       return;
 
     DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
 
-    // TODO(rockot): Implement sync dispatch. For now, sync messages are
-    // unsupported here.
+    // Sync messages should never make their way to this method.
     DCHECK(!message->has_flag(mojo::Message::kFlagIsSync));
 
     bool result = false;
     {
       base::AutoUnlock unlocker(lock_);
       result = client->HandleIncomingMessage(message.get());
     }
 
     if (!result)
       RaiseError();
   }
 
+  void AcceptSyncMessage(mojo::InterfaceId interface_id,
+                         mojo::Message* raw_message) {
+    DCHECK(proxy_task_runner_->BelongsToCurrentThread());
+
+    scoped_refptr<ChannelAssociatedGroupController> keepalive(this);
+    base::AutoLock locker(lock_);
+    Endpoint* endpoint = GetEndpointForDispatch(interface_id);
+    if (!endpoint)
+      return;
+
+    mojo::InterfaceEndpointClient* client = endpoint->client();
+    if (!client)

[yzshen1, 2016/08/02 17:38:37]

If there is not client, what should we do here? Should we drop the message?
Returning directly will leave the sync message in the queue.

Do we allow this endpoint to re-attach to a client again? (I guess we don't) If
we don't, then this shouldn't be an issue except that it takes up some space for
longer. Otherwise, we may not be able to process the messages in the sync
message queue.

[Ken Rockot(use gerrit already), 2016/08/02 18:21:24]

Fixed to drop the message in this case

[yzshen1, 2016/08/02 18:34:11]

Btw, I think we don't have code to disallow re-attaching an endpoint client
currently. I think we should add such code, because we want to guarantee
FIFO-ness and we also drop messages in AcceptOnProxyThread/AcceptSyncMessage.
(This doesn't block landing this CL.)

+      return;
+
+    DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
+    std::unique_ptr<mojo::Message> message =
+        endpoint->PopSyncMessage(raw_message);
+
+    // The message must have already been dequeued by the endpoint waking up
+    // from a sync wait. Nothing to do.
+    if (!message)
+      return;
+
+    bool result = false;
+    {
+      base::AutoUnlock unlocker(lock_);
+      result = client->HandleIncomingMessage(message.get());
+    }
+
+    if (!result)
+      RaiseError();
+  }
+
   Endpoint* GetEndpointForDispatch(mojo::InterfaceId id) {
     lock_.AssertAcquired();
     bool inserted = false;
     Endpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
     if (inserted) {
       MarkClosedAndMaybeRemove(endpoint);
       if (!mojo::IsMasterInterfaceId(id))
         control_message_proxy_.NotifyPeerEndpointClosed(id);
       return nullptr;
     }
@@ skipping 116 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