Revert of Mojo C++ bindings: support sync methods - part 2

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"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/synchronization/lock.h"
-#include "mojo/public/cpp/bindings/lib/sync_handle_watcher.h"
 
 namespace mojo {
 namespace internal {
 
 namespace {
 
 // Similar to base::AutoLock, except that it does nothing if |lock| passed into
 // the constructor is null.
 class MayAutoLock {
  public:
@@ skipping 22 matching lines @@
                      ConnectorConfig config,
                      const MojoAsyncWaiter* waiter)
     : waiter_(waiter),
       message_pipe_(std::move(message_pipe)),
       incoming_receiver_(nullptr),
       async_wait_id_(0),
       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),
-      sync_handle_watcher_callback_count_(0),
-      weak_factory_(this) {
+      destroyed_flag_(nullptr),
+      lock_(config == MULTI_THREADED_SEND ? new base::Lock : nullptr) {
   // 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());
 
+  if (destroyed_flag_)
+    *destroyed_flag_ = true;
+
   CancelWait();
 }
 
 void Connector::CloseMessagePipe() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   CancelWait();
   MayAutoLock locker(lock_.get());
   Close(std::move(message_pipe_));
 }
@@ skipping 106 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() {
-  DCHECK(thread_checker_.CalledOnValidThread());
-
-  if (error_)
-    return false;
-
-  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;
-}
-
-void Connector::UnregisterSyncHandleWatch() {
-  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();
-
-  return SyncHandleWatcher::current()->WatchAllHandles(message_pipe_.get(),
-                                                       should_stop);
-}
-
 // static
 void Connector::CallOnHandleReady(void* closure, MojoResult result) {
   Connector* self = static_cast<Connector*>(closure);
-  CHECK(self->async_wait_id_ != 0);
-  self->async_wait_id_ = 0;
-  self->OnHandleReadyInternal(result);
+  self->OnHandleReady(result);
 }
 
-void Connector::OnSyncHandleWatcherHandleReady(MojoResult result) {
-  base::WeakPtr<Connector> weak_self(weak_factory_.GetWeakPtr());
-
-  sync_handle_watcher_callback_count_++;
-  OnHandleReadyInternal(result);
-  // At this point, this object might have been deleted.
-  if (weak_self)
-    sync_handle_watcher_callback_count_--;
-}
-
-void Connector::OnHandleReadyInternal(MojoResult result) {
+void Connector::OnHandleReady(MojoResult result) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
+  CHECK(async_wait_id_ != 0);
+  async_wait_id_ = 0;
   if (result != MOJO_RESULT_OK) {
     HandleError(result != MOJO_RESULT_FAILED_PRECONDITION, false);
     return;
   }
   ReadAllAvailableMessages();
   // At this point, this object might have been deleted. Return.
 }
 
 void Connector::WaitToReadMore() {
   CHECK(!async_wait_id_);
   CHECK(!paused_);
   async_wait_id_ = waiter_->AsyncWait(message_pipe_.get().value(),
                                       MOJO_HANDLE_SIGNAL_READABLE,
                                       MOJO_DEADLINE_INDEFINITE,
                                       &Connector::CallOnHandleReady,
                                       this);
-
-  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)));
-  }
 }
 
 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.
-  base::WeakPtr<Connector> weak_self = weak_factory_.GetWeakPtr();
+  bool was_destroyed_during_dispatch = false;
+  bool* previous_destroyed_flag = destroyed_flag_;
+  destroyed_flag_ = &was_destroyed_during_dispatch;
 
   Message message;
   const MojoResult rv = ReadMessage(message_pipe_.get(), &message);
   *read_result = rv;
 
   if (rv == MOJO_RESULT_OK) {
     // Dispatching the message may spin in a nested message loop. To ensure we
     // continue dispatching messages when this happens start listening for
     // messagse now.
     if (!async_wait_id_) {
       // TODO: Need to evaluate the perf impact of this.
       WaitToReadMore();
     }
     receiver_result =
         incoming_receiver_ && incoming_receiver_->Accept(&message);
   }
 
-  if (!weak_self)
+  if (was_destroyed_during_dispatch) {
+    if (previous_destroyed_flag)
+      *previous_destroyed_flag = true;  // Propagate flag.
     return false;
+  }
+
+  destroyed_flag_ = previous_destroyed_flag;
 
   if (rv == MOJO_RESULT_SHOULD_WAIT)
     return true;
 
   if (rv != MOJO_RESULT_OK) {
     HandleError(rv != MOJO_RESULT_FAILED_PRECONDITION, false);
     return false;
   }
 
   if (enforce_errors_from_incoming_receiver_ && !receiver_result) {
@@ skipping 21 matching lines @@
       // because of thread hops. In that case, there'll still be an async
       // waiter.
       if (!async_wait_id_)
         WaitToReadMore();
       break;
     }
   }
 }
 
 void Connector::CancelWait() {
-  if (async_wait_id_) {
-    waiter_->CancelWait(async_wait_id_);
-    async_wait_id_ = 0;
-  }
+  if (!async_wait_id_)
+    return;
 
-  if (registered_with_sync_handle_watcher_) {
-    SyncHandleWatcher::current()->UnregisterHandle(message_pipe_.get());
-    registered_with_sync_handle_watcher_ = false;
-  }
+  waiter_->CancelWait(async_wait_id_);
+  async_wait_id_ = 0;
 }
 
 void Connector::HandleError(bool force_pipe_reset, bool force_async_handler) {
   if (error_ || !message_pipe_.is_valid())
     return;
 
   if (!force_pipe_reset && force_async_handler)
     force_pipe_reset = true;
 
   if (paused_) {
@@ skipping 19 matching lines @@
     if (!paused_)
       WaitToReadMore();
   } else {
     error_ = true;
     connection_error_handler_.Run();
   }
 }
 
 }  // namespace internal
 }  // namespace mojo