Adds sync message support to Channel-associated interfaces.

ipc/ipc_sync_channel.cc

 // Copyright (c) 2012 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_sync_channel.h"
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <utility>
 
 #include "base/bind.h"
 #include "base/lazy_instance.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
 #include "base/run_loop.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/threading/thread_local.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "base/trace_event/trace_event.h"
 #include "ipc/ipc_channel_factory.h"
 #include "ipc/ipc_logging.h"
 #include "ipc/ipc_message_macros.h"
 #include "ipc/ipc_sync_message.h"
 #include "ipc/mojo_event.h"
 #include "mojo/public/cpp/bindings/sync_handle_registry.h"
+#include "mojo/public/cpp/bindings/sync_handle_watcher.h"
 
 using base::WaitableEvent;
 
 namespace IPC {
 
 namespace {
 
 // A generic callback used when watching handles synchronously. Sets |*signal|
 // to true. Also sets |*error| to true in case of an error.
 void OnSyncHandleReady(bool* signal, bool* error, MojoResult result) {
@@ skipping 55 matching lines @@
     // SyncChannel objects can block the same thread).
     ReceivedSyncMsgQueue* rv = lazy_tls_ptr_.Pointer()->Get();
     if (!rv) {
       rv = new ReceivedSyncMsgQueue();
       ReceivedSyncMsgQueue::lazy_tls_ptr_.Pointer()->Set(rv);
     }
     rv->listener_count_++;
     return rv;
   }
 
+  // Prevents messages from being dispatched immediately when the dispatch event
+  // is signaled. Instead, |*dispatch_flag| will be set.
+  void BlockDispatch(bool *dispatch_flag) { dispatch_flag_ = dispatch_flag; }

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

style nit: "*" should be next to "bool" instead of the variable name.

[Ken Rockot(use gerrit already), 2016/08/02 19:05:05]

Done

+
+  // Allows messages to be dispatched immediately when the dispatch event is
+  // signaled.
+  void UnblockDispatch() { dispatch_flag_ = nullptr; }
+
   // Called on IPC thread when a synchronous message or reply arrives.
   void QueueMessage(const Message& msg, SyncChannel::SyncContext* context) {
     bool was_task_pending;
     {
       base::AutoLock auto_lock(message_lock_);
 
       was_task_pending = task_pending_;
       task_pending_ = true;
 
       // We set the event in case the listener thread is blocked (or is about
@@ skipping 17 matching lines @@
   // Called on the listener's thread to process any queues synchronous
   // messages.
   void DispatchMessagesTask(SyncContext* context) {
     {
       base::AutoLock auto_lock(message_lock_);
       task_pending_ = false;
     }
     context->DispatchMessages();
   }
 
+  // Dispatches any queued incoming sync messages. If |dispatching_context| is
+  // not null, messages which target a restricted dispatch channel will only be
+  // dispatched if |dispatching_context| belongs to the same restricted dispatch
+  // group as that channel. If |dispatching_context| is null, all queued
+  // messages are dispatched.
   void DispatchMessages(SyncContext* dispatching_context) {
     bool first_time = true;
     uint32_t expected_version = 0;
     SyncMessageQueue::iterator it;
     while (true) {
-      Message* message = NULL;
+      Message* message = nullptr;
       scoped_refptr<SyncChannel::SyncContext> context;
       {
         base::AutoLock auto_lock(message_lock_);
         if (first_time || message_queue_version_ != expected_version) {
           it = message_queue_.begin();
           first_time = false;
         }
         for (; it != message_queue_.end(); it++) {
           int message_group = it->context->restrict_dispatch_group();
-          if (message_group == kRestrictDispatchGroup_None ||
+          if (!dispatching_context ||
+              message_group == kRestrictDispatchGroup_None ||
               message_group == dispatching_context->restrict_dispatch_group()) {
             message = it->message;
             context = it->context;
             it = message_queue_.erase(it);
             message_queue_version_++;
             expected_version = message_queue_version_;
             break;
           }
         }
       }
 
-      if (message == NULL)
+      if (message == nullptr)
         break;
       context->OnDispatchMessage(*message);
       delete message;
     }
   }
 
   // SyncChannel calls this in its destructor.
   void RemoveContext(SyncContext* context) {
     base::AutoLock auto_lock(message_lock_);
 
     SyncMessageQueue::iterator iter = message_queue_.begin();
     while (iter != message_queue_.end()) {
       if (iter->context.get() == context) {
         delete iter->message;
         iter = message_queue_.erase(iter);
         message_queue_version_++;
       } else {
         iter++;
       }
     }
 
     if (--listener_count_ == 0) {
       DCHECK(lazy_tls_ptr_.Pointer()->Get());
-      lazy_tls_ptr_.Pointer()->Set(NULL);
+      lazy_tls_ptr_.Pointer()->Set(nullptr);
+      sync_dispatch_watcher_.reset();
     }
   }
 
   MojoEvent* dispatch_event() { return &dispatch_event_; }
   base::SingleThreadTaskRunner* listener_task_runner() {
     return listener_task_runner_.get();
   }
 
   // Holds a pointer to the per-thread ReceivedSyncMsgQueue object.
   static base::LazyInstance<base::ThreadLocalPointer<ReceivedSyncMsgQueue> >
@@ skipping 23 matching lines @@
  private:
   friend class base::RefCountedThreadSafe<ReceivedSyncMsgQueue>;
 
   // See the comment in SyncChannel::SyncChannel for why this event is created
   // as manual reset.
   ReceivedSyncMsgQueue()
       : message_queue_version_(0),
         listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
         task_pending_(false),
         listener_count_(0),
-        top_send_done_watcher_(NULL) {}
+        top_send_done_watcher_(nullptr) {
+    sync_dispatch_watcher_.reset(new mojo::SyncHandleWatcher(
+        dispatch_event_.GetHandle(), MOJO_HANDLE_SIGNAL_READABLE,
+        base::Bind(&ReceivedSyncMsgQueue::OnDispatchHandleReady,
+                   base::Unretained(this))));
+    sync_dispatch_watcher_->AllowWokenUpBySyncWatchOnSameThread();
+  }
 
   ~ReceivedSyncMsgQueue() {}
 
+  void OnDispatchHandleReady(MojoResult result) {
+    if (result != MOJO_RESULT_OK)
+      return;
+
+    if (dispatch_flag_) {
+      *dispatch_flag_ = true;
+      return;
+    }
+
+    // We were woken up during a sync wait, but no specific SyncChannel is
+    // currently waiting. i.e., some other Mojo interface on this thread is
+    // waiting for a response. Since we don't support anything analogous to
+    // restricted dispatch on Mojo interfaces, in this case it's safe to
+    // dispatch sync messages for any context.
+    DispatchMessages(nullptr);
+  }
+
   // Holds information about a queued synchronous message or reply.
   struct QueuedMessage {
     QueuedMessage(Message* m, SyncContext* c) : message(m), context(c) { }
     Message* message;
     scoped_refptr<SyncChannel::SyncContext> context;
   };
 
   typedef std::list<QueuedMessage> SyncMessageQueue;
   SyncMessageQueue message_queue_;
   uint32_t message_queue_version_;  // Used to signal DispatchMessages to rescan
 
   std::vector<QueuedMessage> received_replies_;
 
   // Signaled when we get a synchronous message that we must respond to, as the
   // sender needs its reply before it can reply to our original synchronous
   // message.
   MojoEvent dispatch_event_;
   scoped_refptr<base::SingleThreadTaskRunner> listener_task_runner_;
   base::Lock message_lock_;
   bool task_pending_;
   int listener_count_;
 
   // The current send done handle watcher for this thread. Used to maintain
   // a thread-local stack of send done watchers to ensure that nested sync
   // message loops complete correctly.
   mojo::Watcher* top_send_done_watcher_;
+
+  // If not null, the address of a flag to set when the dispatch event signals,
+  // in lieu of actually dispatching messages. This is used by
+  // SyncChannel::WaitForReply to restrict the scope of queued messages we're
+  // allowed to process while it's waiting.
+  bool* dispatch_flag_ = nullptr;
+
+  // Watches |dispatch_event_| during all sync handle watches on this thread.
+  std::unique_ptr<mojo::SyncHandleWatcher> sync_dispatch_watcher_;
 };
 
 base::LazyInstance<base::ThreadLocalPointer<SyncChannel::ReceivedSyncMsgQueue> >
     SyncChannel::ReceivedSyncMsgQueue::lazy_tls_ptr_ =
         LAZY_INSTANCE_INITIALIZER;
 
 SyncChannel::SyncContext::SyncContext(
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
     WaitableEvent* shutdown_event)
@@ skipping 222 matching lines @@
       sync_context()->IsChannelSendThreadSafe());
   AddFilter(filter.get());
   if (!did_init())
     pre_init_sync_message_filters_.push_back(filter);
   return filter;
 }
 
 bool SyncChannel::Send(Message* message) {
 #ifdef IPC_MESSAGE_LOG_ENABLED
   std::string name;
-  Logging::GetInstance()->GetMessageText(message->type(), &name, message, NULL);
+  Logging::GetInstance()->GetMessageText(
+      message->type(), &name, message, nullptr);
   TRACE_EVENT1("ipc", "SyncChannel::Send", "name", name);
 #else
   TRACE_EVENT2("ipc", "SyncChannel::Send",
                "class", IPC_MESSAGE_ID_CLASS(message->type()),
                "line", IPC_MESSAGE_ID_LINE(message->type()));
 #endif
   if (!message->is_sync()) {
     ChannelProxy::Send(message);
     return true;
   }
 
   SyncMessage* sync_msg = static_cast<SyncMessage*>(message);
   bool pump_messages = sync_msg->ShouldPumpMessages();
 
   // *this* might get deleted in WaitForReply.
   scoped_refptr<SyncContext> context(sync_context());
   if (!context->Push(sync_msg)) {
     DVLOG(1) << "Channel is shutting down. Dropping sync message.";
     delete message;
     return false;
   }
 
   ChannelProxy::Send(message);
 
   // Wait for reply, or for any other incoming synchronous messages.
-  // *this* might get deleted, so only call static functions at this point.
+  // |this| might get deleted, so only call static functions at this point.
   scoped_refptr<mojo::SyncHandleRegistry> registry = sync_handle_registry_;
   WaitForReply(registry.get(), context.get(), pump_messages);
 
   TRACE_EVENT_FLOW_END0(TRACE_DISABLED_BY_DEFAULT("ipc.flow"),
                         "SyncChannel::Send", context->GetSendDoneEvent());
 
   return context->Pop();
 }
 
 void SyncChannel::WaitForReply(mojo::SyncHandleRegistry* registry,
                                SyncContext* context,
                                bool pump_messages) {
   context->DispatchMessages();
 
   const MojoEvent* pump_messages_event = nullptr;
   if (pump_messages)
     pump_messages_event = g_pump_messages_event.Get().event();
 
   while (true) {
     bool dispatch = false;
     bool send_done = false;
     bool should_pump_messages = false;
     bool error = false;
-    registry->RegisterHandle(context->GetDispatchEvent()->GetHandle(),
-                             MOJO_HANDLE_SIGNAL_READABLE,
-                             base::Bind(&OnSyncHandleReady, &dispatch, &error));
-    registry->RegisterHandle(
+    bool registered = registry->RegisterHandle(
         context->GetSendDoneEvent()->GetHandle(),
         MOJO_HANDLE_SIGNAL_READABLE,
         base::Bind(&OnSyncHandleReady, &send_done, &error));
+    DCHECK(registered);
     if (pump_messages_event) {
-      registry->RegisterHandle(
+      registered = registry->RegisterHandle(
           pump_messages_event->GetHandle(), MOJO_HANDLE_SIGNAL_READABLE,
           base::Bind(&OnSyncHandleReady, &should_pump_messages, &error));
+      DCHECK(registered);
     }
 
     const bool* stop_flags[] = { &dispatch, &send_done, &should_pump_messages };
+    context->received_sync_msgs()->BlockDispatch(&dispatch);
     registry->WatchAllHandles(stop_flags, 3);
+    context->received_sync_msgs()->UnblockDispatch();
     DCHECK(!error);
 
-    registry->UnregisterHandle(context->GetDispatchEvent()->GetHandle());
+
     registry->UnregisterHandle(context->GetSendDoneEvent()->GetHandle());
     if (pump_messages_event)
       registry->UnregisterHandle(pump_messages_event->GetHandle());
 
     if (dispatch) {
-      // We're waiting for a reply, but we received a blocking synchronous
-      // call.  We must process it or otherwise a deadlock might occur.
+      // We're waiting for a reply, but we received a blocking synchronous call.
+      // We must process it to avoid potential deadlocks.
       context->GetDispatchEvent()->Reset();
       context->DispatchMessages();
       continue;
     }
 
     if (should_pump_messages)
       WaitForReplyWithNestedMessageLoop(context);  // Run a nested message loop.
 
     break;
   }
@@ skipping 38 matching lines @@
 
 void SyncChannel::OnDispatchHandleReady(MojoResult result) {
   DCHECK(result == MOJO_RESULT_OK || result == MOJO_RESULT_ABORTED);
   if (result == MOJO_RESULT_OK) {
     sync_context()->GetDispatchEvent()->Reset();
     sync_context()->DispatchMessages();
   }
 }
 
 void SyncChannel::StartWatching() {
-  // Ideally we only want to watch this object when running a nested message
-  // loop.  However, we don't know when it exits if there's another nested
-  // message loop running under it or not, so we wouldn't know whether to
-  // stop or keep watching.  So we always watch it.
+  // |dispatch_watcher_| watches the event asynchronously, only dispatching
+  // messages once the listener thread is unblocked and pumping its task queue.
+  // The ReceivedSyncMsgQueue also watches this event and may dispatch
+  // immediately if woken up by the same

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

the comment doesn't seem to be complete.

[Ken Rockot(use gerrit already), 2016/08/02 19:05:05]

Yikes - Done

   dispatch_watcher_.Start(sync_context()->GetDispatchEvent()->GetHandle(),
                           MOJO_HANDLE_SIGNAL_READABLE,
                           base::Bind(&SyncChannel::OnDispatchHandleReady,
                                      base::Unretained(this)));
 }
 
 void SyncChannel::OnChannelInit() {
   for (const auto& filter : pre_init_sync_message_filters_) {
     filter->set_is_channel_send_thread_safe(
         context()->IsChannelSendThreadSafe());
   }
   pre_init_sync_message_filters_.clear();
 }
 
 }  // namespace IPC