Use Mojo pipes to signal sync IPC events

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/synchronization/waitable_event_watcher.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"
 
 using base::TimeDelta;
 using base::TimeTicks;
 using base::WaitableEvent;
 
 namespace IPC {
+
+namespace {
+
+// A lazy thread-local Mojo Event which is always signaled. Used to wake up the
+// sync waiter when a SyncMessage requires the MessageLoop to be pumped while
+// waiting for a reply. This object is created lazily and ref-counted so it can
+// be cleaned up when no longer in use.
+class PumpMessagesEvent {
+ public:
+  // Acquires the event for this thread. Creates a new instance if necessary.
+  static PumpMessagesEvent* Acquire() {
+    PumpMessagesEvent* pump_messages_event = g_event_.Pointer()->Get();
+    if (!pump_messages_event) {
+      pump_messages_event = new PumpMessagesEvent;
+      pump_messages_event->event_.Signal();
+      g_event_.Pointer()->Set(pump_messages_event);
+    }
+    pump_messages_event->ref_count_++;
+    return pump_messages_event;
+  }
+
+  // Releases a handle to this event. There must be a 1:1 correspondence between
+  // calls to Acquire() and calls to Release().
+  static void Release() {
+    PumpMessagesEvent* pump_messages_event = g_event_.Pointer()->Get();
+    DCHECK(pump_messages_event);
+    DCHECK_GT(pump_messages_event->ref_count_, 0);
+    pump_messages_event->ref_count_--;
+    if (!pump_messages_event->ref_count_) {
+      g_event_.Pointer()->Set(nullptr);
+      delete pump_messages_event;
+    }
+  }
+
+  const mojo::Handle& GetHandle() const { return event_.GetHandle(); }
+
+ private:
+  PumpMessagesEvent() {}
+  ~PumpMessagesEvent() {}
+
+  int ref_count_ = 0;
+  MojoEvent event_;
+
+  static base::LazyInstance<base::ThreadLocalPointer<PumpMessagesEvent>>
+      g_event_;
+
+  DISALLOW_COPY_AND_ASSIGN(PumpMessagesEvent);
+};
+
+// 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) {
+  *signal = true;
+  *error = result != MOJO_RESULT_OK;
+}
+
+// A ReadyCallback for use with mojo::Watcher. Ignores the result (DCHECKs, but
+// is only used in cases where failure should be impossible) and runs
+// |callback|.
+void RunOnHandleReady(const base::Closure& callback, MojoResult result) {
+  DCHECK_EQ(result, MOJO_RESULT_OK);
+  callback.Run();
+}
+
+}  // namespace
+
+base::LazyInstance<base::ThreadLocalPointer<PumpMessagesEvent>>
+PumpMessagesEvent::g_event_ = LAZY_INSTANCE_INITIALIZER;
+
 // When we're blocked in a Send(), we need to process incoming synchronous
 // messages right away because it could be blocking our reply (either
 // directly from the same object we're calling, or indirectly through one or
 // more other channels).  That means that in SyncContext's OnMessageReceived,
 // we need to process sync message right away if we're blocked.  However a
 // simple check isn't sufficient, because the listener thread can be in the
 // process of calling Send.
 // To work around this, when SyncChannel filters a sync message, it sets
 // an event that the listener thread waits on during its Send() call.  This
 // allows us to dispatch incoming sync messages when blocked.  The race
@@ skipping 106 matching lines @@
         iter++;
       }
     }
 
     if (--listener_count_ == 0) {
       DCHECK(lazy_tls_ptr_.Pointer()->Get());
       lazy_tls_ptr_.Pointer()->Set(NULL);
     }
   }
 
-  WaitableEvent* dispatch_event() { return &dispatch_event_; }
+  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> >
       lazy_tls_ptr_;
 
   // Called on the ipc thread to check if we can unblock any current Send()
   // calls based on a queued reply.
   void DispatchReplies() {
     for (size_t i = 0; i < received_replies_.size(); ++i) {
       Message* message = received_replies_[i].message;
       if (received_replies_[i].context->TryToUnblockListener(message)) {
         delete message;
         received_replies_.erase(received_replies_.begin() + i);
         return;
       }
     }
   }
 
-  base::WaitableEventWatcher* top_send_done_watcher() {
+  mojo::Watcher* top_send_done_watcher() {
     return top_send_done_watcher_;
   }
 
-  void set_top_send_done_watcher(base::WaitableEventWatcher* watcher) {
+  void set_top_send_done_watcher(mojo::Watcher* watcher) {
     top_send_done_watcher_ = watcher;
   }
 
  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),
-        dispatch_event_(base::WaitableEvent::ResetPolicy::MANUAL,
-                        base::WaitableEvent::InitialState::NOT_SIGNALED),
         listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
         task_pending_(false),
         listener_count_(0),
         top_send_done_watcher_(NULL) {}
 
   ~ReceivedSyncMsgQueue() {}
 
   // 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_;
 
-  // Set when we got a synchronous message that we must respond to as the
+  // 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.
-  WaitableEvent dispatch_event_;
+  MojoEvent dispatch_event_;
   scoped_refptr<base::SingleThreadTaskRunner> listener_task_runner_;
   base::Lock message_lock_;
   bool task_pending_;
   int listener_count_;
 
-  // The current send done event watcher for this thread. Used to maintain
-  // a local global stack of send done watchers to ensure that nested sync
+  // 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.
-  base::WaitableEventWatcher* top_send_done_watcher_;
+  mojo::Watcher* top_send_done_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 15 matching lines @@
   // Create the tracking information for this message. This object is stored
   // by value since all members are pointers that are cheap to copy. These
   // pointers are cleaned up in the Pop() function.
   //
   // The event is created as manual reset because in between Signal and
   // OnObjectSignalled, another Send can happen which would stop the watcher
   // from being called.  The event would get watched later, when the nested
   // Send completes, so the event will need to remain set.
   PendingSyncMsg pending(
       SyncMessage::GetMessageId(*sync_msg), sync_msg->GetReplyDeserializer(),
-      new WaitableEvent(base::WaitableEvent::ResetPolicy::MANUAL,
-                        base::WaitableEvent::InitialState::NOT_SIGNALED));
+      new MojoEvent);
   base::AutoLock auto_lock(deserializers_lock_);
   deserializers_.push_back(pending);
 }
 
 bool SyncChannel::SyncContext::Pop() {
   bool result;
   {
     base::AutoLock auto_lock(deserializers_lock_);
     PendingSyncMsg msg = deserializers_.back();
     delete msg.deserializer;
     delete msg.done_event;
-    msg.done_event = NULL;
+    msg.done_event = nullptr;
     deserializers_.pop_back();
     result = msg.send_result;
   }
 
   // We got a reply to a synchronous Send() call that's blocking the listener
   // thread.  However, further down the call stack there could be another
   // blocking Send() call, whose reply we received after we made this last
   // Send() call.  So check if we have any queued replies available that
   // can now unblock the listener thread.
   ipc_task_runner()->PostTask(
       FROM_HERE, base::Bind(&ReceivedSyncMsgQueue::DispatchReplies,
                             received_sync_msgs_.get()));
 
   return result;
 }
 
-WaitableEvent* SyncChannel::SyncContext::GetSendDoneEvent() {
+MojoEvent* SyncChannel::SyncContext::GetSendDoneEvent() {
   base::AutoLock auto_lock(deserializers_lock_);
   return deserializers_.back().done_event;
 }
 
-WaitableEvent* SyncChannel::SyncContext::GetDispatchEvent() {
+MojoEvent* SyncChannel::SyncContext::GetDispatchEvent() {
   return received_sync_msgs_->dispatch_event();
 }
 
 void SyncChannel::SyncContext::DispatchMessages() {
   received_sync_msgs_->DispatchMessages(this);
 }
 
 bool SyncChannel::SyncContext::TryToUnblockListener(const Message* msg) {
   base::AutoLock auto_lock(deserializers_lock_);
   if (deserializers_.empty() ||
       !SyncMessage::IsMessageReplyTo(*msg, deserializers_.back().id)) {
     return false;
   }
 
   if (!msg->is_reply_error()) {
     bool send_result = deserializers_.back().deserializer->
         SerializeOutputParameters(*msg);
     deserializers_.back().send_result = send_result;
     DVLOG_IF(1, !send_result) << "Couldn't deserialize reply message";
   } else {
     DVLOG(1) << "Received error reply";
   }
 
-  base::WaitableEvent* done_event = deserializers_.back().done_event;
+  MojoEvent* done_event = deserializers_.back().done_event;
   TRACE_EVENT_FLOW_BEGIN0(
       TRACE_DISABLED_BY_DEFAULT("ipc.flow"),
       "SyncChannel::SyncContext::TryToUnblockListener", done_event);
 
   done_event->Signal();
 
   return true;
 }
 
 void SyncChannel::SyncContext::Clear() {
@@ skipping 25 matching lines @@
 
 void SyncChannel::SyncContext::OnChannelError() {
   CancelPendingSends();
   shutdown_watcher_.StopWatching();
   Context::OnChannelError();
 }
 
 void SyncChannel::SyncContext::OnChannelOpened() {
   shutdown_watcher_.StartWatching(
       shutdown_event_,
-      base::Bind(&SyncChannel::SyncContext::OnWaitableEventSignaled,
+      base::Bind(&SyncChannel::SyncContext::OnShutdownEventSignaled,
                  base::Unretained(this)));
   Context::OnChannelOpened();
 }
 
 void SyncChannel::SyncContext::OnChannelClosed() {
   CancelPendingSends();
   shutdown_watcher_.StopWatching();
   Context::OnChannelClosed();
 }
 
 void SyncChannel::SyncContext::CancelPendingSends() {
   base::AutoLock auto_lock(deserializers_lock_);
   PendingSyncMessageQueue::iterator iter;
   DVLOG(1) << "Canceling pending sends";
   for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++) {
     TRACE_EVENT_FLOW_BEGIN0(TRACE_DISABLED_BY_DEFAULT("ipc.flow"),
                             "SyncChannel::SyncContext::CancelPendingSends",
                             iter->done_event);
     iter->done_event->Signal();
   }
 }
 
-void SyncChannel::SyncContext::OnWaitableEventSignaled(WaitableEvent* event) {
-  if (event == shutdown_event_) {
-    // Process shut down before we can get a reply to a synchronous message.
-    // Cancel pending Send calls, which will end up setting the send done event.
-    CancelPendingSends();
-  } else {
-    // We got the reply, timed out or the process shutdown.
-    DCHECK_EQ(GetSendDoneEvent(), event);
-    base::MessageLoop::current()->QuitNow();
-  }
-}
+void SyncChannel::SyncContext::OnShutdownEventSignaled(WaitableEvent* event) {
+  DCHECK_EQ(event, shutdown_event_);
 
-base::WaitableEventWatcher::EventCallback
-    SyncChannel::SyncContext::MakeWaitableEventCallback() {
-  return base::Bind(&SyncChannel::SyncContext::OnWaitableEventSignaled, this);
+  // Process shut down before we can get a reply to a synchronous message.
+  // Cancel pending Send calls, which will end up setting the send done event.
+  CancelPendingSends();
 }
 
 // static
 std::unique_ptr<SyncChannel> SyncChannel::Create(
     const IPC::ChannelHandle& channel_handle,
     Channel::Mode mode,
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
     bool create_pipe_now,
     base::WaitableEvent* shutdown_event) {
@@ skipping 23 matching lines @@
     WaitableEvent* shutdown_event) {
   return base::WrapUnique(
       new SyncChannel(listener, ipc_task_runner, shutdown_event));
 }
 
 SyncChannel::SyncChannel(
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
     WaitableEvent* shutdown_event)
     : ChannelProxy(new SyncContext(listener, ipc_task_runner, shutdown_event)) {
+  // Keep a thread-local PumpMessagesEvent alive at least as long as any
+  // SyncChannel exists. This is balanced in the SyncChannel destructor below.
+  PumpMessagesEvent::Acquire();
+
   // The current (listener) thread must be distinct from the IPC thread, or else
   // sending synchronous messages will deadlock.
   DCHECK_NE(ipc_task_runner.get(), base::ThreadTaskRunnerHandle::Get().get());
   StartWatching();
 }
 
 SyncChannel::~SyncChannel() {
+  PumpMessagesEvent::Release();
 }
 
 void SyncChannel::SetRestrictDispatchChannelGroup(int group) {
   sync_context()->set_restrict_dispatch_group(group);
 }
 
 scoped_refptr<SyncMessageFilter> SyncChannel::CreateSyncMessageFilter() {
   scoped_refptr<SyncMessageFilter> filter = new SyncMessageFilter(
       sync_context()->shutdown_event(),
       sync_context()->IsChannelSendThreadSafe());
@@ skipping 20 matching lines @@
 
   // *this* might get deleted in WaitForReply.
   scoped_refptr<SyncContext> context(sync_context());
   if (context->shutdown_event()->IsSignaled()) {
     DVLOG(1) << "shutdown event is signaled";
     delete message;
     return false;
   }
 
   SyncMessage* sync_msg = static_cast<SyncMessage*>(message);
+  bool pump_messages = sync_msg->ShouldPumpMessages();
   context->Push(sync_msg);
-  WaitableEvent* pump_messages_event = sync_msg->pump_messages_event();
 
   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.
-  WaitForReply(context.get(), pump_messages_event);
+  WaitForReply(context.get(), pump_messages);
 
   TRACE_EVENT_FLOW_END0(TRACE_DISABLED_BY_DEFAULT("ipc.flow"),
                         "SyncChannel::Send", context->GetSendDoneEvent());
 
   return context->Pop();
 }
 
-void SyncChannel::WaitForReply(
-    SyncContext* context, WaitableEvent* pump_messages_event) {
+void SyncChannel::WaitForReply(SyncContext* context, bool pump_messages) {
   context->DispatchMessages();
+
+  PumpMessagesEvent* pump_messages_event = nullptr;
+  if (pump_messages)
+    pump_messages_event = PumpMessagesEvent::Acquire();
+
+  scoped_refptr<mojo::SyncHandleRegistry> registry =
+      mojo::SyncHandleRegistry::current();
+
   while (true) {
-    WaitableEvent* objects[] = {
-      context->GetDispatchEvent(),
-      context->GetSendDoneEvent(),
-      pump_messages_event
-    };
+    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(
+        context->GetSendDoneEvent()->GetHandle(),
+        MOJO_HANDLE_SIGNAL_READABLE,
+        base::Bind(&OnSyncHandleReady, &send_done, &error));
+    if (pump_messages_event) {
+      registry->RegisterHandle(
+          pump_messages_event->GetHandle(), MOJO_HANDLE_SIGNAL_READABLE,
+          base::Bind(&OnSyncHandleReady, &should_pump_messages, &error));
+    }
 
-    unsigned count = pump_messages_event ? 3: 2;
-    size_t result = WaitableEvent::WaitMany(objects, count);
-    if (result == 0 /* dispatch event */) {
+    const bool* stop_flags[] = { &dispatch, &send_done, &should_pump_messages };
+    bool result = registry->WatchAllHandles(stop_flags, 3);
+    DCHECK(result);
+    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.
       context->GetDispatchEvent()->Reset();
       context->DispatchMessages();
       continue;
     }
 
-    if (result == 2 /* pump_messages_event */)
+    DCHECK(send_done || should_pump_messages);
+
+    if (should_pump_messages)
       WaitForReplyWithNestedMessageLoop(context);  // Run a nested message loop.
 
     break;
   }
+
+  if (pump_messages_event)
+    PumpMessagesEvent::Release();
 }
 
 void SyncChannel::WaitForReplyWithNestedMessageLoop(SyncContext* context) {
-  base::WaitableEventWatcher send_done_watcher;
+  mojo::Watcher send_done_watcher;
 
   ReceivedSyncMsgQueue* sync_msg_queue = context->received_sync_msgs();
-  DCHECK(sync_msg_queue != NULL);
+  DCHECK_NE(sync_msg_queue, nullptr);
 
-  base::WaitableEventWatcher* old_send_done_event_watcher =
-      sync_msg_queue->top_send_done_watcher();
+  mojo::Watcher* old_watcher = sync_msg_queue->top_send_done_watcher();
+  mojo::Handle old_handle(mojo::kInvalidHandleValue);
+  mojo::Watcher::ReadyCallback old_callback;
 
-  base::WaitableEventWatcher::EventCallback old_callback;
-  base::WaitableEvent* old_event = NULL;
-
-  // Maintain a local global stack of send done delegates to ensure that
-  // nested sync calls complete in the correct sequence, i.e. the
-  // outermost call completes first, etc.
-  if (old_send_done_event_watcher) {
-    old_callback = old_send_done_event_watcher->callback();
-    old_event = old_send_done_event_watcher->GetWatchedEvent();
-    old_send_done_event_watcher->StopWatching();
+  // Maintain a thread-local stack of watchers to ensure nested calls complete
+  // in the correct sequence, i.e. the outermost call completes first, etc.
+  if (old_watcher) {
+    old_callback = old_watcher->ready_callback();
+    old_handle = old_watcher->handle();
+    old_watcher->Cancel();
   }
 
   sync_msg_queue->set_top_send_done_watcher(&send_done_watcher);
 
-  send_done_watcher.StartWatching(context->GetSendDoneEvent(),
-                                  context->MakeWaitableEventCallback());
+  {
+    base::RunLoop nested_loop;
+    send_done_watcher.Start(
+        context->GetSendDoneEvent()->GetHandle(), MOJO_HANDLE_SIGNAL_READABLE,
+        base::Bind(&RunOnHandleReady, nested_loop.QuitClosure()));
 
-  {
     base::MessageLoop::ScopedNestableTaskAllower allow(
         base::MessageLoop::current());
-    base::MessageLoop::current()->Run();
+    nested_loop.Run();
+    send_done_watcher.Cancel();
   }
 
-  sync_msg_queue->set_top_send_done_watcher(old_send_done_event_watcher);
-  if (old_send_done_event_watcher && old_event) {
-    old_send_done_event_watcher->StartWatching(old_event, old_callback);
-  }
+  sync_msg_queue->set_top_send_done_watcher(old_watcher);
+  if (old_watcher)
+    old_watcher->Start(old_handle, MOJO_HANDLE_SIGNAL_READABLE, old_callback);
 }
 
-void SyncChannel::OnWaitableEventSignaled(WaitableEvent* event) {
-  DCHECK(event == sync_context()->GetDispatchEvent());
-  // The call to DispatchMessages might delete this object, so reregister
-  // the object watcher first.
-  event->Reset();
-  dispatch_watcher_.StartWatching(event, dispatch_watcher_callback_);
+void SyncChannel::OnDispatchHandleReady(MojoResult result) {
+  DCHECK_EQ(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, and create the event as
-  // manual reset since the object watcher might otherwise reset the event
-  // when we're doing a WaitMany.
-  dispatch_watcher_callback_ =
-      base::Bind(&SyncChannel::OnWaitableEventSignaled,
-                  base::Unretained(this));
-  dispatch_watcher_.StartWatching(sync_context()->GetDispatchEvent(),
-                                  dispatch_watcher_callback_);
+  // stop or keep watching.  So we always watch it.
+  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();
 }
 
@@ skipping 23 matching lines @@
   TRACE_EVENT2("ipc", "SyncChannel::SendOnIPCThread",
                "class", IPC_MESSAGE_ID_CLASS(message->type()),
                "line", IPC_MESSAGE_ID_LINE(message->type()));
 #endif
   if (!message->is_sync())
     return ChannelProxy::SendOnIPCThread(std::move(message));
   return Send(message.release());
 }
 
 }  // namespace IPC