Mojo C++ bindings: refactor SyncHandleWatcher.

mojo/public/cpp/bindings/lib/connector.cc

 // Copyright 2013 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/connector.h"
 
 #include <stdint.h>
 #include <utility>
 
 #include "base/bind.h"
@@ skipping 35 matching lines @@
 
 Connector::Connector(ScopedMessagePipeHandle message_pipe,
                      ConnectorConfig config)
     : message_pipe_(std::move(message_pipe)),
       incoming_receiver_(nullptr),
       error_(false),
       drop_writes_(false),
       enforce_errors_from_incoming_receiver_(true),
       paused_(false),
       lock_(config == MULTI_THREADED_SEND ? new base::Lock : nullptr),
-      register_sync_handle_watch_count_(0),
-      registered_with_sync_handle_watcher_(false),
+      allow_woken_up_by_others_(false),
       sync_handle_watcher_callback_count_(0),
       weak_factory_(this) {
   weak_self_ = weak_factory_.GetWeakPtr();
   // Even though we don't have an incoming receiver, we still want to monitor
   // the message pipe to know if is closed or encounters an error.
   WaitToReadMore();
 }
 
 Connector::~Connector() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   CancelWait();
 }
 
 void Connector::CloseMessagePipe() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   CancelWait();
   MayAutoLock locker(lock_.get());
-  Close(std::move(message_pipe_));
+  message_pipe_.reset();
 }
 
 ScopedMessagePipeHandle Connector::PassMessagePipe() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   CancelWait();
   MayAutoLock locker(lock_.get());
   return std::move(message_pipe_);
 }
 
@@ skipping 97 matching lines @@
       CHECK(false) << "Race condition or other bug detected";
       return false;
     default:
       // This particular write was rejected, presumably because of bad input.
       // The pipe is not necessarily in a bad state.
       return false;
   }
   return true;
 }
 
-bool Connector::RegisterSyncHandleWatch() {
+void Connector::AllowWokenUpBySyncWatchOnSameThread() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  if (error_)
-    return false;
+  allow_woken_up_by_others_ = true;
 
-  register_sync_handle_watch_count_++;
-
-  if (!registered_with_sync_handle_watcher_ && !paused_) {
-    registered_with_sync_handle_watcher_ =
-        SyncHandleWatcher::current()->RegisterHandle(
-            message_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
-            base::Bind(&Connector::OnSyncHandleWatcherHandleReady,
-                       base::Unretained(this)));
-  }
-  return true;
+  EnsureSyncWatcherExists();
+  sync_watcher_->AllowWokenUpBySyncWatchOnSameThread();
 }
 
-void Connector::UnregisterSyncHandleWatch() {
+bool Connector::SyncWatch(const bool* should_stop) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  if (register_sync_handle_watch_count_ == 0) {
-    NOTREACHED();
-    return;
-  }
-
-  register_sync_handle_watch_count_--;
-  if (register_sync_handle_watch_count_ > 0)
-    return;
-
-  if (registered_with_sync_handle_watcher_) {
-    SyncHandleWatcher::current()->UnregisterHandle(message_pipe_.get());
-    registered_with_sync_handle_watcher_ = false;
-  }
-}
-
-bool Connector::RunSyncHandleWatch(const bool* should_stop) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  DCHECK_GT(register_sync_handle_watch_count_, 0u);
-
   if (error_)
     return false;
 
   ResumeIncomingMethodCallProcessing();
 
-  if (!should_stop_sync_handle_watch_)
-    should_stop_sync_handle_watch_ = new base::RefCountedData<bool>(false);
-
-  // This object may be destroyed during the WatchAllHandles() call. So we have
-  // to preserve the boolean that WatchAllHandles uses.
-  scoped_refptr<base::RefCountedData<bool>> preserver =
-      should_stop_sync_handle_watch_;
-  const bool* should_stop_array[] = {should_stop,
-                                     &should_stop_sync_handle_watch_->data};
-  return SyncHandleWatcher::current()->WatchAllHandles(should_stop_array, 2);
+  EnsureSyncWatcherExists();
+  return sync_watcher_->SyncWatch(should_stop);
 }
 
 void Connector::OnWatcherHandleReady(MojoResult result) {
   OnHandleReadyInternal(result);
 }
 
 void Connector::OnSyncHandleWatcherHandleReady(MojoResult result) {
   base::WeakPtr<Connector> weak_self(weak_self_);
 
   sync_handle_watcher_callback_count_++;
@@ skipping 24 matching lines @@
                  base::Unretained(this)));
 
   if (rv != MOJO_RESULT_OK) {
     // If the watch failed because the handle is invalid or its conditions can
     // no longer be met, we signal the error asynchronously to avoid reentry.
     base::ThreadTaskRunnerHandle::Get()->PostTask(
         FROM_HERE,
         base::Bind(&Connector::OnWatcherHandleReady, weak_self_, rv));
   }
 
-  if (register_sync_handle_watch_count_ > 0 &&
-      !registered_with_sync_handle_watcher_) {
-    registered_with_sync_handle_watcher_ =
-        SyncHandleWatcher::current()->RegisterHandle(
-            message_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
-            base::Bind(&Connector::OnSyncHandleWatcherHandleReady,
-                       base::Unretained(this)));
+  if (allow_woken_up_by_others_) {
+    EnsureSyncWatcherExists();
+    sync_watcher_->AllowWokenUpBySyncWatchOnSameThread();
   }
 }
 
 bool Connector::ReadSingleMessage(MojoResult* read_result) {
   CHECK(!paused_);
 
   bool receiver_result = false;
 
   // Detect if |this| was destroyed during message dispatch. Allow for the
   // possibility of re-entering ReadMore() through message dispatch.
@@ skipping 37 matching lines @@
     if (paused_)
       return;
 
     if (rv == MOJO_RESULT_SHOULD_WAIT)
       break;
   }
 }
 
 void Connector::CancelWait() {
   handle_watcher_.Cancel();
-
-  if (registered_with_sync_handle_watcher_) {
-    SyncHandleWatcher::current()->UnregisterHandle(message_pipe_.get());
-    registered_with_sync_handle_watcher_ = false;
-  }
-
-  if (should_stop_sync_handle_watch_)
-    should_stop_sync_handle_watch_->data = true;
+  sync_watcher_.reset();
 }
 
 void Connector::HandleError(bool force_pipe_reset, bool force_async_handler) {
   if (error_ || !message_pipe_.is_valid())
     return;
 
-  if (during_sync_handle_watcher_callback() || paused_) {
-    // Enforce calling the error handler asynchronously if:
-    // - currently we are in a sync handle watcher callback. We don't want the
-    // error handler to reenter an ongoing sync call.
-    // - the user has paused receiving messages. We need to wait until the user
-    // starts receiving messages again.
+  if (paused_) {
+    // Enforce calling the error handler asynchronously if the user has paused
+    // receiving messages. We need to wait until the user starts receiving
+    // messages again.
     force_async_handler = true;
   }
 
   if (!force_pipe_reset && force_async_handler)
     force_pipe_reset = true;
 
   if (force_pipe_reset) {
     CancelWait();
     MayAutoLock locker(lock_.get());
-    Close(std::move(message_pipe_));
+    message_pipe_.reset();
     MessagePipe dummy_pipe;
     message_pipe_ = std::move(dummy_pipe.handle0);
   } else {
     CancelWait();
   }
 
   if (force_async_handler) {
     if (!paused_)
       WaitToReadMore();
   } else {
     error_ = true;
     connection_error_handler_.Run();
   }
 }
 
+void Connector::EnsureSyncWatcherExists() {
+  if (sync_watcher_)
+    return;
+  sync_watcher_.reset(new SyncHandleWatcher(
+      message_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
+      base::Bind(&Connector::OnSyncHandleWatcherHandleReady,
+                 base::Unretained(this))));
+}
+
 }  // namespace internal
 }  // namespace mojo