Revert of Mojo: Armed Watchers

mojo/edk/system/message_pipe_dispatcher.cc

Index: mojo/edk/system/message_pipe_dispatcher.cc
diff --git a/mojo/edk/system/message_pipe_dispatcher.cc b/mojo/edk/system/message_pipe_dispatcher.cc
index 76f412306d042ce7d072b4bcf8c2eb7e4cb43de7..f27336b56f4d074a0e5e32ec2a592fcccff363a3 100644
--- a/mojo/edk/system/message_pipe_dispatcher.cc
+++ b/mojo/edk/system/message_pipe_dispatcher.cc
@@ -164,8 +164,7 @@
     : node_controller_(node_controller),
       port_(port),
       pipe_id_(pipe_id),
-      endpoint_(endpoint),
-      watchers_(this) {
+      endpoint_(endpoint) {
   DVLOG(2) << "Creating new MessagePipeDispatcher for port " << port.name()
            << " [pipe_id=" << pipe_id << "; endpoint=" << endpoint << "]";
 
@@ -184,7 +183,6 @@
     port0 = port_;
     port_closed_.Set(true);
     awakables_.CancelAll();
-    watchers_.NotifyClosed();
   }
 
   ports::PortRef port1;
@@ -193,7 +191,6 @@
     port1 = other->port_;
     other->port_closed_.Set(true);
     other->awakables_.CancelAll();
-    other->watchers_.NotifyClosed();
   }
 
   // Both ports are always closed by this call.
@@ -210,6 +207,27 @@
   DVLOG(2) << "Closing message pipe " << pipe_id_ << " endpoint " << endpoint_
            << " [port=" << port_.name() << "]";
   return CloseNoLock();
+}
+
+MojoResult MessagePipeDispatcher::Watch(MojoHandleSignals signals,
+                                        const Watcher::WatchCallback& callback,
+                                        uintptr_t context) {
+  base::AutoLock lock(signal_lock_);
+
+  if (port_closed_ || in_transit_)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  return awakables_.AddWatcher(
+      signals, callback, context, GetHandleSignalsStateNoLock());
+}
+
+MojoResult MessagePipeDispatcher::CancelWatch(uintptr_t context) {
+  base::AutoLock lock(signal_lock_);
+
+  if (port_closed_ || in_transit_)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  return awakables_.RemoveWatcher(context);
 }
 
 MojoResult MessagePipeDispatcher::WriteMessage(
@@ -281,12 +299,6 @@
   }
 
   if (no_space) {
-    if (may_discard) {
-      // May have been the last message on the pipe. Need to update signals just
-      // in case.
-      base::AutoLock lock(signal_lock_);
-      watchers_.NotifyState(GetHandleSignalsStateNoLock());
-    }
     // |*num_handles| (and/or |*num_bytes| if |read_any_size| is false) wasn't
     // sufficient to hold this message's data. The message will still be in
     // queue unless MOJO_READ_MESSAGE_FLAG_MAY_DISCARD was set.
@@ -306,13 +318,6 @@
 
   // Alright! We have a message and the caller has provided sufficient storage
   // in which to receive it.
-
-  {
-    // We need to update anyone watching our signals in case that was the last
-    // available message.
-    base::AutoLock lock(signal_lock_);
-    watchers_.NotifyState(GetHandleSignalsStateNoLock());
-  }
 
   std::unique_ptr<PortsMessage> msg(
       static_cast<PortsMessage*>(ports_message.release()));
@@ -391,23 +396,6 @@
   return GetHandleSignalsStateNoLock();
 }
 
-MojoResult MessagePipeDispatcher::AddWatcherRef(
-    const scoped_refptr<WatcherDispatcher>& watcher,
-    uintptr_t context) {
-  base::AutoLock lock(signal_lock_);
-  if (port_closed_ || in_transit_)
-    return MOJO_RESULT_INVALID_ARGUMENT;
-  return watchers_.Add(watcher, context, GetHandleSignalsStateNoLock());
-}
-
-MojoResult MessagePipeDispatcher::RemoveWatcherRef(WatcherDispatcher* watcher,
-                                                   uintptr_t context) {
-  base::AutoLock lock(signal_lock_);
-  if (port_closed_ || in_transit_)
-    return MOJO_RESULT_INVALID_ARGUMENT;
-  return watchers_.Remove(watcher, context);
-}
-
 MojoResult MessagePipeDispatcher::AddAwakable(
     Awakable* awakable,
     MojoHandleSignals signals,
@@ -508,9 +496,7 @@
   in_transit_.Set(false);
 
   // Something may have happened while we were waiting for potential transit.
-  HandleSignalsState state = GetHandleSignalsStateNoLock();
-  awakables_.AwakeForStateChange(state);
-  watchers_.NotifyState(state);
+  awakables_.AwakeForStateChange(GetHandleSignalsStateNoLock());
 }
 
 // static
@@ -546,7 +532,6 @@
 
   port_closed_.Set(true);
   awakables_.CancelAll();
-  watchers_.NotifyClosed();
 
   if (!port_transferred_) {
     base::AutoUnlock unlock(signal_lock_);
@@ -611,9 +596,7 @@
   }
 #endif
 
-  HandleSignalsState state = GetHandleSignalsStateNoLock();
-  awakables_.AwakeForStateChange(state);
-  watchers_.NotifyState(state);
+  awakables_.AwakeForStateChange(GetHandleSignalsStateNoLock());
 }
 
 }  // namespace edk