Reland Mojo-based waiting for IPC::SyncChannel

ipc/ipc_sync_message_filter.cc

Index: ipc/ipc_sync_message_filter.cc
diff --git a/ipc/ipc_sync_message_filter.cc b/ipc/ipc_sync_message_filter.cc
index 1d17432a319ff6f12d614b037edefde701d109eb..53ead2c0b8d17d52a25894839ced76f02479c8b0 100644
--- a/ipc/ipc_sync_message_filter.cc
+++ b/ipc/ipc_sync_message_filter.cc
@@ -7,14 +7,94 @@
 #include "base/bind.h"
 #include "base/location.h"
 #include "base/logging.h"
+#include "base/macros.h"
+#include "base/memory/ptr_util.h"
+#include "base/memory/ref_counted.h"
+#include "base/message_loop/message_loop.h"
 #include "base/single_thread_task_runner.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "ipc/ipc_channel.h"
 #include "ipc/ipc_sync_message.h"
+#include "ipc/mojo_event.h"
+#include "mojo/public/cpp/bindings/sync_handle_registry.h"
 
 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) {
+  *signal = true;
+  *error = result != MOJO_RESULT_OK;
+}
+
+}  // namespace
+
+// A helper class created by SyncMessageFilter to watch the lifetime of the IO
+// MessageLoop. This holds a weak ref to the SyncMessageFilter and notifies it
+// on its own thread if the SyncMessageFilter is still alive at the time of
+// IO MessageLoop destruction.
+class SyncMessageFilter::IOMessageLoopObserver
+    : public base::MessageLoop::DestructionObserver,
+      public base::RefCountedThreadSafe<IOMessageLoopObserver> {
+ public:
+  IOMessageLoopObserver(
+      base::WeakPtr<SyncMessageFilter> weak_filter,
+      scoped_refptr<base::SingleThreadTaskRunner> filter_task_runner)
+      : weak_filter_(weak_filter), filter_task_runner_(filter_task_runner) {}
+
+  void StartOnIOThread() {
+    DCHECK(!watching_);
+    watching_ = true;
+    io_task_runner_ = base::ThreadTaskRunnerHandle::Get();
+    base::MessageLoop::current()->AddDestructionObserver(this);
+  }
+
+  void Stop() {
+    if (!io_task_runner_)
+      return;
+
+    if (io_task_runner_->BelongsToCurrentThread()) {
+      StopOnIOThread();
+    } else {
+      io_task_runner_->PostTask(
+          FROM_HERE, base::Bind(&IOMessageLoopObserver::StopOnIOThread, this));
+    }
+  }
+
+ private:
+  void StopOnIOThread() {
+    DCHECK(io_task_runner_->BelongsToCurrentThread());
+    if (!watching_)
+      return;
+    watching_ = false;
+    base::MessageLoop::current()->RemoveDestructionObserver(this);
+  }
+
+  // base::MessageLoop::DestructionObserver:
+  void WillDestroyCurrentMessageLoop() override {
+    DCHECK(io_task_runner_ && io_task_runner_->BelongsToCurrentThread());
+    DCHECK(watching_);
+    StopOnIOThread();
+    filter_task_runner_->PostTask(
+        FROM_HERE,
+        base::Bind(&SyncMessageFilter::OnIOMessageLoopDestroyed, weak_filter_));
+  }
+
+  friend class base::RefCountedThreadSafe<IOMessageLoopObserver>;
+
+  ~IOMessageLoopObserver() override {}
+
+  bool watching_ = false;
+  base::WeakPtr<SyncMessageFilter> weak_filter_;
+  scoped_refptr<base::SingleThreadTaskRunner> filter_task_runner_;
+  scoped_refptr<base::SingleThreadTaskRunner> io_task_runner_;
+
+  DISALLOW_COPY_AND_ASSIGN(IOMessageLoopObserver);
+};
+
 bool SyncMessageFilter::Send(Message* message) {
   if (!message->is_sync()) {
     {
@@ -23,7 +103,7 @@ bool SyncMessageFilter::Send(Message* message) {
         sender_->Send(message);
         return true;
       } else if (!io_task_runner_.get()) {
-        pending_messages_.push_back(message);
+        pending_messages_.emplace_back(base::WrapUnique(message));
         return true;
       }
     }
@@ -33,9 +113,7 @@ bool SyncMessageFilter::Send(Message* message) {
     return true;
   }
 
-  base::WaitableEvent done_event(
-      base::WaitableEvent::ResetPolicy::MANUAL,
-      base::WaitableEvent::InitialState::NOT_SIGNALED);
+  MojoEvent done_event;
   PendingSyncMsg pending_message(
       SyncMessage::GetMessageId(*message),
       static_cast<SyncMessage*>(message)->GetReplyDeserializer(),
@@ -56,15 +134,32 @@ bool SyncMessageFilter::Send(Message* message) {
           FROM_HERE,
           base::Bind(&SyncMessageFilter::SendOnIOThread, this, message));
     } else {
-      pending_messages_.push_back(message);
+      pending_messages_.emplace_back(base::WrapUnique(message));
     }
   }
 
-  base::WaitableEvent* events[2] = { shutdown_event_, &done_event };
-  if (base::WaitableEvent::WaitMany(events, 2) == 1) {
+  bool done = false;
+  bool shutdown = false;
+  bool error = false;
+  scoped_refptr<mojo::SyncHandleRegistry> registry =
+      mojo::SyncHandleRegistry::current();
+  registry->RegisterHandle(shutdown_mojo_event_.GetHandle(),
+                           MOJO_HANDLE_SIGNAL_READABLE,
+                           base::Bind(&OnSyncHandleReady, &shutdown, &error));
+  registry->RegisterHandle(done_event.GetHandle(),
+                           MOJO_HANDLE_SIGNAL_READABLE,
+                           base::Bind(&OnSyncHandleReady, &done, &error));
+
+  const bool* stop_flags[] = { &done, &shutdown };
+  registry->WatchAllHandles(stop_flags, 2);
+  DCHECK(!error);
+
+  if (done) {
     TRACE_EVENT_FLOW_END0(TRACE_DISABLED_BY_DEFAULT("ipc.flow"),
                           "SyncMessageFilter::Send", &done_event);
   }
+  registry->UnregisterHandle(shutdown_mojo_event_.GetHandle());
+  registry->UnregisterHandle(done_event.GetHandle());
 
   {
     base::AutoLock auto_lock(lock_);
@@ -76,26 +171,32 @@ bool SyncMessageFilter::Send(Message* message) {
 }
 
 void SyncMessageFilter::OnFilterAdded(Sender* sender) {
-  std::vector<Message*> pending_messages;
+  std::vector<std::unique_ptr<Message>> pending_messages;
   {
     base::AutoLock auto_lock(lock_);
     sender_ = sender;
     io_task_runner_ = base::ThreadTaskRunnerHandle::Get();
-    pending_messages_.release(&pending_messages);
+    shutdown_watcher_.StartWatching(
+        shutdown_event_,
+        base::Bind(&SyncMessageFilter::OnShutdownEventSignaled, this));
+    io_message_loop_observer_->StartOnIOThread();
+    std::swap(pending_messages_, pending_messages);
   }
-  for (auto* msg : pending_messages)
-    SendOnIOThread(msg);
+  for (auto& msg : pending_messages)
+    SendOnIOThread(msg.release());
 }
 
 void SyncMessageFilter::OnChannelError() {
   base::AutoLock auto_lock(lock_);
   sender_ = NULL;
+  shutdown_watcher_.StopWatching();
   SignalAllEvents();
 }
 
 void SyncMessageFilter::OnChannelClosing() {
   base::AutoLock auto_lock(lock_);
   sender_ = NULL;
+  shutdown_watcher_.StopWatching();
   SignalAllEvents();
 }
 
@@ -124,10 +225,14 @@ SyncMessageFilter::SyncMessageFilter(base::WaitableEvent* shutdown_event,
     : sender_(NULL),
       is_channel_send_thread_safe_(is_channel_send_thread_safe),
       listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
-      shutdown_event_(shutdown_event) {
+      shutdown_event_(shutdown_event),
+      weak_factory_(this) {
+  io_message_loop_observer_ = new IOMessageLoopObserver(
+      weak_factory_.GetWeakPtr(), listener_task_runner_);
 }
 
 SyncMessageFilter::~SyncMessageFilter() {
+  io_message_loop_observer_->Stop();
 }
 
 void SyncMessageFilter::SendOnIOThread(Message* message) {
@@ -157,4 +262,17 @@ void SyncMessageFilter::SignalAllEvents() {
   }
 }
 
+void SyncMessageFilter::OnShutdownEventSignaled(base::WaitableEvent* event) {
+  DCHECK_EQ(event, shutdown_event_);
+  shutdown_mojo_event_.Signal();
+}
+
+void SyncMessageFilter::OnIOMessageLoopDestroyed() {
+  // Since we use an async WaitableEventWatcher to watch the shutdown event
+  // from the IO thread, we can't forward the shutdown signal after the IO
+  // message loop is destroyed. Since that destruction indicates shutdown
+  // anyway, we manually signal the shutdown event in this case.
+  shutdown_mojo_event_.Signal();
+}
+
 }  // namespace IPC