Mojo: Armed Watchers

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>
@@ skipping 23 matching lines @@
 
 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;
 }
 
-// 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|.
+// A ReadyCallback for use with mojo::SimpleWatcher. 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();
 }
 
 class PumpMessagesEvent {
  public:
   PumpMessagesEvent() { event_.Signal(); }
   ~PumpMessagesEvent() {}
 
@@ skipping 166 matching lines @@
     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;
       }
     }
   }
 
-  mojo::Watcher* top_send_done_watcher() {
+  mojo::SimpleWatcher* top_send_done_watcher() {
     return top_send_done_watcher_;
   }
 
-  void set_top_send_done_watcher(mojo::Watcher* watcher) {
+  void set_top_send_done_watcher(mojo::SimpleWatcher* 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),
@@ skipping 45 matching lines @@
   // 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_;
+  mojo::SimpleWatcher* 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_;
 };
@@ skipping 206 matching lines @@
   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)),
       sync_handle_registry_(mojo::SyncHandleRegistry::current()),
-      dispatch_watcher_(FROM_HERE) {
+      dispatch_watcher_(FROM_HERE,
+                        mojo::SimpleWatcher::ArmingPolicy::AUTOMATIC) {
   // 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() {
 }
 
 void SyncChannel::SetRestrictDispatchChannelGroup(int group) {
@@ skipping 74 matching lines @@
           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->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 to avoid potential deadlocks.
       context->GetDispatchEvent()->Reset();
       context->DispatchMessages();
       continue;
     }
 
     if (should_pump_messages)
       WaitForReplyWithNestedMessageLoop(context);  // Run a nested message loop.
 
     break;
   }
 }
 
 void SyncChannel::WaitForReplyWithNestedMessageLoop(SyncContext* context) {
-  mojo::Watcher send_done_watcher(FROM_HERE);
+  mojo::SimpleWatcher send_done_watcher(
+      FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::AUTOMATIC);
 
   ReceivedSyncMsgQueue* sync_msg_queue = context->received_sync_msgs();
   DCHECK_NE(sync_msg_queue, nullptr);
 
-  mojo::Watcher* old_watcher = sync_msg_queue->top_send_done_watcher();
+  mojo::SimpleWatcher* old_watcher = sync_msg_queue->top_send_done_watcher();
   mojo::Handle old_handle(mojo::kInvalidHandleValue);
-  mojo::Watcher::ReadyCallback old_callback;
+  mojo::SimpleWatcher::ReadyCallback old_callback;
 
   // 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);
 
   {
     base::RunLoop nested_loop;
-    send_done_watcher.Start(
+    send_done_watcher.Watch(
         context->GetSendDoneEvent()->GetHandle(), MOJO_HANDLE_SIGNAL_READABLE,
         base::Bind(&RunOnHandleReady, nested_loop.QuitClosure()));
 
     base::MessageLoop::ScopedNestableTaskAllower allow(
         base::MessageLoop::current());
     nested_loop.Run();
     send_done_watcher.Cancel();
   }
 
   sync_msg_queue->set_top_send_done_watcher(old_watcher);
   if (old_watcher)
-    old_watcher->Start(old_handle, MOJO_HANDLE_SIGNAL_READABLE, old_callback);
+    old_watcher->Watch(old_handle, MOJO_HANDLE_SIGNAL_READABLE, old_callback);
 }
 
 void SyncChannel::OnDispatchHandleReady(MojoResult result) {
   DCHECK_EQ(result, MOJO_RESULT_OK);
   sync_context()->GetDispatchEvent()->Reset();
   sync_context()->DispatchMessages();
 }
 
 void SyncChannel::StartWatching() {
   // |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 a message which it's allowed to dispatch.
-  dispatch_watcher_.Start(sync_context()->GetDispatchEvent()->GetHandle(),
-                          MOJO_HANDLE_SIGNAL_READABLE,
-                          base::Bind(&SyncChannel::OnDispatchHandleReady,
-                                     base::Unretained(this)));
+  dispatch_watcher_.Watch(
+      sync_context()->GetDispatchEvent()->GetHandle(),
+      MOJO_HANDLE_SIGNAL_READABLE,
+      base::Bind(&SyncChannel::OnDispatchHandleReady, base::Unretained(this)));
 }
 
 void SyncChannel::OnChannelInit() {
   pre_init_sync_message_filters_.clear();
 }
 
 }  // namespace IPC