Use WaitableEvents to wake up sync IPC waiting

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>
 
 #include "base/bind.h"
 #include "base/location.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
 #include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
+#include "base/synchronization/waitable_event.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_client.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_controller.h"
 #include "mojo/public/cpp/bindings/lib/may_auto_lock.h"
-#include "mojo/public/cpp/bindings/sync_handle_watcher.h"
+#include "mojo/public/cpp/bindings/sync_event_watcher.h"
 
 namespace mojo {
 namespace internal {
 
 // InterfaceEndpoint stores the information of an interface endpoint registered
 // with the router.
 // No one other than the router's |endpoints_| and |tasks_| should hold refs to
 // 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) {}
+        client_(nullptr) {}
 
   // ---------------------------------------------------------------------------
   // 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.
 
@@ skipping 49 matching lines @@
     DCHECK(task_runner_->BelongsToCurrentThread());
     DCHECK(!closed_);
 
     task_runner_ = nullptr;
     client_ = nullptr;
     sync_watcher_.reset();
   }
 
   void SignalSyncMessageEvent() {
     router_->AssertLockAcquired();
-    if (event_signalled_)
+    if (sync_message_event_signaled_)
       return;
-
-    event_signalled_ = true;
-    if (!sync_message_event_sender_.is_valid())
-      return;
-
-    MojoResult result =
-        WriteMessageRaw(sync_message_event_sender_.get(), nullptr, 0, nullptr,
-                        0, MOJO_WRITE_MESSAGE_FLAG_NONE);
-    DCHECK_EQ(MOJO_RESULT_OK, result);
+    sync_message_event_signaled_ = true;
+    if (sync_message_event_)
+      sync_message_event_->Signal();
   }
 
   void ResetSyncMessageSignal() {
     router_->AssertLockAcquired();
-
-    if (!event_signalled_)
+    if (!sync_message_event_signaled_)
       return;
-
-    event_signalled_ = false;
-    if (!sync_message_event_receiver_.is_valid())
-      return;
-
-    MojoResult result =
-        ReadMessageRaw(sync_message_event_receiver_.get(), nullptr, nullptr,
-                       nullptr, nullptr, MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
-    DCHECK_EQ(MOJO_RESULT_OK, result);
+    sync_message_event_signaled_ = false;
+    if (sync_message_event_)
+      sync_message_event_->Reset();
   }
 
   // ---------------------------------------------------------------------------
   // 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->set_interface_id(id_);
@@ skipping 19 matching lines @@
 
   ~InterfaceEndpoint() override {
     router_->AssertLockAcquired();
 
     DCHECK(!client_);
     DCHECK(closed_);
     DCHECK(peer_closed_);
     DCHECK(!sync_watcher_);
   }
 
-  void OnHandleReady(MojoResult result) {
+  void OnSyncEventSignaled() {
     DCHECK(task_runner_->BelongsToCurrentThread());
     scoped_refptr<MultiplexRouter> router_protector(router_);
 
-    // Because we never close |sync_message_event_{sender,receiver}_| before
-    // destruction or set a deadline, |result| should always be MOJO_RESULT_OK.
-    DCHECK_EQ(MOJO_RESULT_OK, result);
-
     MayAutoLock locker(&router_->lock_);
     scoped_refptr<InterfaceEndpoint> self_protector(this);
 
     bool more_to_process = router_->ProcessFirstSyncMessageForEndpoint(id_);
 
     if (!more_to_process)
       ResetSyncMessageSignal();
 
     // Currently there are no queued sync messages and the peer has closed so
     // there won't be incoming sync messages in the future.
     if (!more_to_process && peer_closed_) {
       // 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;
 
     {
       MayAutoLock locker(&router_->lock_);
-
-      if (!sync_message_event_sender_.is_valid()) {
-        MojoResult result =
-            CreateMessagePipe(nullptr, &sync_message_event_sender_,
-                              &sync_message_event_receiver_);
-        DCHECK_EQ(MOJO_RESULT_OK, result);
-
-        if (event_signalled_) {
-          // Reset the flag so that SignalSyncMessageEvent() will actually
-          // signal using the newly-created message pipe.
-          event_signalled_ = false;
-          SignalSyncMessageEvent();
-        }
+      if (!sync_message_event_) {
+        sync_message_event_.emplace(
+            base::WaitableEvent::ResetPolicy::MANUAL,
+            base::WaitableEvent::InitialState::NOT_SIGNALED);

[yzshen1, 2017/03/24 16:41:47]

If the flag |event_signaled_| is already set to true here, we need to signal the
event.

[Ken Rockot(use gerrit already), 2017/03/24 16:43:55]

Done

       }
     }
-
-    sync_watcher_.reset(new SyncHandleWatcher(
-        sync_message_event_receiver_.get(), MOJO_HANDLE_SIGNAL_READABLE,
-        base::Bind(&InterfaceEndpoint::OnHandleReady, base::Unretained(this))));
+    sync_watcher_.reset(
+        new SyncEventWatcher(&sync_message_event_.value(),
+                             base::Bind(&InterfaceEndpoint::OnSyncEventSignaled,
+                                        base::Unretained(this))));
   }
 
   // ---------------------------------------------------------------------------
   // The following members are safe to access from any threads.
 
   MultiplexRouter* const router_;
   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_;
+  // An event used to signal that sync messages are available. The event is
+  // initialized under the router's lock and remains unchanged afterwards. It
+  // may be accessed outside of the router's lock later.
+  base::Optional<base::WaitableEvent> sync_message_event_;
+  bool sync_message_event_signaled_ = false;
 
   // ---------------------------------------------------------------------------
   // 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_;
+  std::unique_ptr<SyncEventWatcher> sync_watcher_;
 
   DISALLOW_COPY_AND_ASSIGN(InterfaceEndpoint);
 };
 
 // MessageWrapper objects are always destroyed under the router's lock. On
 // destruction, if the message it wrappers contains
 // ScopedInterfaceEndpointHandles (which cannot be destructed under the
 // router's lock), the wrapper unlocks to clean them up.
 class MultiplexRouter::MessageWrapper {
  public:
@@ skipping 697 matching lines @@
 
 void MultiplexRouter::AssertLockAcquired() {
 #if DCHECK_IS_ON()
   if (lock_)
     lock_->AssertAcquired();
 #endif
 }
 
 }  // namespace internal
 }  // namespace mojo