Revert of [mojo-bindings] Use Watch API instead of MessagePumpMojo

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 "base/thread_task_runner_handle.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 {
@@ skipping 216 matching lines @@
 
   // 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);
 }
 
-void Connector::OnWatcherHandleReady(MojoResult result) {
+void Connector::OnHandleWatcherHandleReady(MojoResult result) {
   OnHandleReadyInternal(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) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   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(!handle_watcher_.is_watching());
   CHECK(!paused_);
-  DCHECK(!handle_watcher_.IsWatching());
-
-  MojoResult rv = handle_watcher_.Start(
-      message_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
-      base::Bind(&Connector::OnWatcherHandleReady,
-                 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_factory_.GetWeakPtr(), rv));
-  }
+  handle_watcher_.Start(message_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
+                        MOJO_DEADLINE_INDEFINITE,
+                        base::Bind(&Connector::OnHandleWatcherHandleReady,
+                                   base::Unretained(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();
 
   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 (!handle_watcher_.is_watching()) {
+      // TODO: Need to evaluate the perf impact of this.
+      WaitToReadMore();
+    }
     receiver_result =
         incoming_receiver_ && incoming_receiver_->Accept(&message);
   }
 
   if (!weak_self)
     return false;
 
   if (rv == MOJO_RESULT_SHOULD_WAIT)
     return true;
 
@@ skipping 13 matching lines @@
   while (!error_) {
     MojoResult rv;
 
     // Return immediately if |this| was destroyed. Do not touch any members!
     if (!ReadSingleMessage(&rv))
       return;
 
     if (paused_)
       return;
 
-    if (rv == MOJO_RESULT_SHOULD_WAIT)
+    if (rv == MOJO_RESULT_SHOULD_WAIT) {
+      // ReadSingleMessage could end up calling HandleError which resets
+      // message_pipe_ to a dummy one that is closed. The old EDK will see the
+      // that the peer is closed immediately, while the new one is asynchronous
+      // because of thread hops. In that case, there'll still be an async
+      // waiter.
+      if (!handle_watcher_.is_watching())
+        WaitToReadMore();
       break;
+    }
   }
 }
 
 void Connector::CancelWait() {
-  handle_watcher_.Cancel();
+  handle_watcher_.Stop();
 
   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;
 }
 
@@ skipping 15 matching lines @@
     CancelWait();
     MayAutoLock locker(lock_.get());
     Close(std::move(message_pipe_));
     MessagePipe dummy_pipe;
     message_pipe_ = std::move(dummy_pipe.handle0);
   } else {
     CancelWait();
   }
 
   if (force_async_handler) {
+    // |dummy_pipe.handle1| has been destructed. Reading the pipe will
+    // eventually cause a read error on |message_pipe_| and set error state.
     if (!paused_)
       WaitToReadMore();
   } else {
     error_ = true;
     connection_error_handler_.Run();
   }
 }
 
 }  // namespace internal
 }  // namespace mojo