Mojo: Armed Watchers

mojo/edk/system/message_pipe_dispatcher.cc

 // Copyright 2015 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/edk/system/message_pipe_dispatcher.h"
 
 #include <limits>
 #include <memory>
 
 #include "base/logging.h"
@@ skipping 165 matching lines @@
 bool MessagePipeDispatcher::Fuse(MessagePipeDispatcher* other) {
   node_controller_->SetPortObserver(port_, nullptr);
   node_controller_->SetPortObserver(other->port_, nullptr);
 
   ports::PortRef port0;
   {
     base::AutoLock lock(signal_lock_);
     port0 = port_;
     port_closed_.Set(true);
     awakables_.CancelAll();
+    watchers_.NotifyClosed();
   }
 
   ports::PortRef port1;
   {
     base::AutoLock lock(other->signal_lock_);
     port1 = other->port_;
     other->port_closed_.Set(true);
     other->awakables_.CancelAll();
+    other->watchers_.NotifyClosed();
   }
 
   // Both ports are always closed by this call.
   int rv = node_controller_->MergeLocalPorts(port0, port1);
   return rv == ports::OK;
 }
 
 Dispatcher::Type MessagePipeDispatcher::GetType() const {
   return Type::MESSAGE_PIPE;
 }
 
 MojoResult MessagePipeDispatcher::Close() {
   base::AutoLock lock(signal_lock_);
   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) {
+MojoResult MessagePipeDispatcher::RegisterWatcher(
+    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());
+  return watchers_.Add(signals, callback, context,
+                       GetHandleSignalsStateNoLock());
 }
 
-MojoResult MessagePipeDispatcher::CancelWatch(uintptr_t context) {
+MojoResult MessagePipeDispatcher::ArmWatcher(uintptr_t context) {
   base::AutoLock lock(signal_lock_);
 
   if (port_closed_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  return awakables_.RemoveWatcher(context);
+  return watchers_.Arm(context, GetHandleSignalsStateNoLock());
+}
+
+MojoResult MessagePipeDispatcher::UnregisterWatcher(uintptr_t context) {
+  base::AutoLock lock(signal_lock_);
+
+  if (port_closed_ || in_transit_)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  return watchers_.Remove(context);
 }
 
 MojoResult MessagePipeDispatcher::WriteMessage(
     std::unique_ptr<MessageForTransit> message,
     MojoWriteMessageFlags flags) {
   if (port_closed_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   size_t num_bytes = message->num_bytes();
   int rv = node_controller_->SendMessage(port_, message->TakePortsMessage());
@@ skipping 38 matching lines @@
   // that it specifies a size at least as large as the next available payload.
   //
   // If |read_any_size| is true, the input value of |*num_bytes| is ignored.
   // This flag exists to support both new and old API behavior.
 
   ports::ScopedMessage ports_message;
   ReadMessageFilter filter(read_any_size, may_discard, num_bytes, num_handles,
                            &no_space, &invalid_message);
   int rv = node_controller_->node()->GetMessage(port_, &ports_message, &filter);
 
+  {
+    base::AutoLock lock(signal_lock_);
+    watchers_.NotifyState(GetHandleSignalsStateNoLock());
+  }
+
   if (invalid_message)
     return MOJO_RESULT_UNKNOWN;
 
   if (rv != ports::OK && rv != ports::ERROR_PORT_PEER_CLOSED) {
     if (rv == ports::ERROR_PORT_UNKNOWN ||
         rv == ports::ERROR_PORT_STATE_UNEXPECTED)
       return MOJO_RESULT_INVALID_ARGUMENT;
 
     NOTREACHED();
     return MOJO_RESULT_UNKNOWN;  // TODO: Add a better error code here?
@@ skipping 190 matching lines @@
   port_transferred_ = true;
   in_transit_.Set(false);
   CloseNoLock();
 }
 
 void MessagePipeDispatcher::CancelTransit() {
   base::AutoLock lock(signal_lock_);
   in_transit_.Set(false);
 
   // Something may have happened while we were waiting for potential transit.
-  awakables_.AwakeForStateChange(GetHandleSignalsStateNoLock());
+  HandleSignalsState state = GetHandleSignalsStateNoLock();
+  awakables_.AwakeForStateChange(state);
+  watchers_.NotifyState(state);
 }
 
 // static
 scoped_refptr<Dispatcher> MessagePipeDispatcher::Deserialize(
     const void* data,
     size_t num_bytes,
     const ports::PortName* ports,
     size_t num_ports,
     PlatformHandle* handles,
     size_t num_handles) {
@@ skipping 15 matching lines @@
   DCHECK(port_closed_ && !in_transit_);
 }
 
 MojoResult MessagePipeDispatcher::CloseNoLock() {
   signal_lock_.AssertAcquired();
   if (port_closed_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   port_closed_.Set(true);
   awakables_.CancelAll();
+  watchers_.NotifyClosed();
 
   if (!port_transferred_) {
     base::AutoUnlock unlock(signal_lock_);
     node_controller_->ClosePort(port_);
   }
 
   return MOJO_RESULT_OK;
 }
 
 HandleSignalsState MessagePipeDispatcher::GetHandleSignalsStateNoLock() const {
@@ skipping 44 matching lines @@
                << " endpoint " << endpoint_ << " [port=" << port_.name()
                << "; size=" << filter.message_size() << "]";
     }
     if (port_status.peer_closed) {
       DVLOG(2) << "Peer closure detected on message pipe " << pipe_id_
                << " endpoint " << endpoint_ << " [port=" << port_.name() << "]";
     }
   }
 #endif
 
-  awakables_.AwakeForStateChange(GetHandleSignalsStateNoLock());
+  HandleSignalsState state = GetHandleSignalsStateNoLock();
+  awakables_.AwakeForStateChange(state);
+  watchers_.NotifyState(state);
 }
 
 }  // namespace edk
 }  // namespace mojo