[mojo-edk] Add MojoWatch and MojoCancelWatch APIs

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 "base/macros.h"
 #include "base/memory/ref_counted.h"
@@ skipping 72 matching lines @@
 
   node_controller_->SetPortObserver(
       port_,
       make_scoped_refptr(new PortObserverThunk(this)));
 }
 
 Dispatcher::Type MessagePipeDispatcher::GetType() const {
   return Type::MESSAGE_PIPE;
 }
 
-MojoResult MessagePipeDispatcher::Close() {
+MojoResult MessagePipeDispatcher::Close(RequestContext* request_context) {
   base::AutoLock lock(signal_lock_);
   DVLOG(1) << "Closing message pipe " << pipe_id_ << " endpoint " << endpoint_
            << " [port=" << port_.name() << "]";
-  return CloseNoLock();
+  return CloseNoLock(request_context);
+}
+
+MojoResult MessagePipeDispatcher::Watch(MojoHandleSignals signals,
+                                        const WatchCallback& callback,
+                                        uintptr_t context,
+                                        RequestContext* request_context) {
+  base::AutoLock lock(signal_lock_);
+
+  if (port_closed_ || in_transit_)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  HandleSignalsState state = GetHandleSignalsStateNoLock();
+  return watchers_.Add(signals, callback, context, state, request_context);
+}
+
+MojoResult MessagePipeDispatcher::CancelWatch(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(
     const void* bytes,
     uint32_t num_bytes,
     const DispatcherInTransit* dispatchers,
     uint32_t num_dispatchers,
-    MojoWriteMessageFlags flags) {
+    MojoWriteMessageFlags flags,
+    RequestContext* request_context) {
 
   {
     base::AutoLock lock(signal_lock_);
     if (port_closed_ || in_transit_)
       return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
   // A structure for retaining information about every Dispatcher we're about
   // to send. This information is collected by calling StartSerialize() on
   // each dispatcher in sequence.
@@ skipping 103 matching lines @@
              << " [port=" << port_.name() << "; rv=" << rv
              << "; num_bytes=" << num_bytes << "]";
 
     if (rv != ports::OK) {
       if (rv == ports::ERROR_PORT_UNKNOWN ||
           rv == ports::ERROR_PORT_STATE_UNEXPECTED ||
           rv == ports::ERROR_PORT_CANNOT_SEND_PEER) {
         result = MOJO_RESULT_INVALID_ARGUMENT;
       } else if (rv == ports::ERROR_PORT_PEER_CLOSED) {
         base::AutoLock lock(signal_lock_);
-        awakables_.AwakeForStateChange(GetHandleSignalsStateNoLock());
+        NotifyObserversForStateChangeNoLock(request_context);
         result = MOJO_RESULT_FAILED_PRECONDITION;
       } else {
         NOTREACHED();
         result = MOJO_RESULT_UNKNOWN;
       }
       cancel_transit = true;
     } else {
       DCHECK(!message);
     }
   }
 
   if (cancel_transit) {
     // We ended up not sending the message. Release all the platform handles.
     // Their dipatchers retain ownership when transit is canceled, so these are
     // not actually leaking.
     DCHECK(message);
     Channel::MessagePtr m = message->TakeChannelMessage();
     ScopedPlatformHandleVectorPtr handles = m->TakeHandles();
     if (handles)
       handles->clear();
   }
 
   return result;
 }
 
 MojoResult MessagePipeDispatcher::ReadMessage(void* bytes,
                                               uint32_t* num_bytes,
                                               MojoHandle* handles,
                                               uint32_t* num_handles,
-                                              MojoReadMessageFlags flags) {
+                                              MojoReadMessageFlags flags,
+                                              RequestContext* request_context) {
   {
     base::AutoLock lock(signal_lock_);
     // We can't read from a port that's closed or in transit!
     if (port_closed_ || in_transit_)
       return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
   bool no_space = false;
   bool may_discard = flags & MOJO_READ_MESSAGE_FLAG_MAY_DISCARD;
 
@@ skipping 59 matching lines @@
 
   if (!ports_message) {
     // No message was available in queue.
 
     if (rv == ports::OK)
       return MOJO_RESULT_SHOULD_WAIT;
 
     // Peer is closed and there are no more messages to read.
     DCHECK_EQ(rv, ports::ERROR_PORT_PEER_CLOSED);
     base::AutoLock lock(signal_lock_);
-    awakables_.AwakeForStateChange(GetHandleSignalsStateNoLock());
+    NotifyObserversForStateChangeNoLock(request_context);
     return MOJO_RESULT_FAILED_PRECONDITION;
   }
 
   // Alright! We have a message and the caller has provided sufficient storage
   // in which to receive it.
 
   scoped_ptr<PortsMessage> message(
       static_cast<PortsMessage*>(ports_message.release()));
 
   const MessageHeader* header =
@@ skipping 128 matching lines @@
 bool MessagePipeDispatcher::EndSerialize(void* destination,
                                          ports::PortName* ports,
                                          PlatformHandle* handles) {
   SerializedState* state = static_cast<SerializedState*>(destination);
   state->pipe_id = pipe_id_;
   state->endpoint = static_cast<int8_t>(endpoint_);
   ports[0] = port_.name();
   return true;
 }
 
-bool MessagePipeDispatcher::BeginTransit() {
+bool MessagePipeDispatcher::BeginTransit(RequestContext* request_context) {
   base::AutoLock lock(signal_lock_);
   if (in_transit_ || port_closed_)
     return false;
   in_transit_ = true;
   return in_transit_;
 }
 
-void MessagePipeDispatcher::CompleteTransitAndClose() {
+void MessagePipeDispatcher::CompleteTransitAndClose(
+    RequestContext* request_context) {
   node_controller_->SetPortObserver(port_, nullptr);
 
   base::AutoLock lock(signal_lock_);
   in_transit_ = false;
   port_transferred_ = true;
-  CloseNoLock();
+  CloseNoLock(request_context);
 }
 
-void MessagePipeDispatcher::CancelTransit() {
+void MessagePipeDispatcher::CancelTransit(RequestContext* request_context) {
   base::AutoLock lock(signal_lock_);
   in_transit_ = false;
 
   // Something may have happened while we were waiting for potential transit.
-  awakables_.AwakeForStateChange(GetHandleSignalsStateNoLock());
+  NotifyObserversForStateChangeNoLock(request_context);
 }
 
 // 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) {
   if (num_ports != 1 || num_handles || num_bytes != sizeof(SerializedState))
     return nullptr;
 
   const SerializedState* state = static_cast<const SerializedState*>(data);
 
   ports::PortRef port;
   CHECK_EQ(
       ports::OK,
       internal::g_core->GetNodeController()->node()->GetPort(ports[0], &port));
 
   return new MessagePipeDispatcher(internal::g_core->GetNodeController(), port,
                                    state->pipe_id, state->endpoint);
 }
 
 MessagePipeDispatcher::~MessagePipeDispatcher() {
   DCHECK(port_closed_ && !in_transit_);
 }
 
-MojoResult MessagePipeDispatcher::CloseNoLock() {
+MojoResult MessagePipeDispatcher::CloseNoLock(RequestContext* request_context) {
   signal_lock_.AssertAcquired();
   if (port_closed_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   port_closed_ = true;
+  watchers_.CancelAll(request_context);
   awakables_.CancelAll();
 
   if (!port_transferred_) {
     base::AutoUnlock unlock(signal_lock_);
     node_controller_->ClosePort(port_);
   }
 
   return MOJO_RESULT_OK;
 }
 
@@ skipping 17 matching lines @@
     rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
     rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
   } else {
     rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
   }
   rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
   return rv;
 }
 
 void MessagePipeDispatcher::OnPortStatusChanged() {
+  // Create a new RequestContext that will run its finalizers after
+  // |signal_lock_| is released.
+  RequestContext request_context;
+
   base::AutoLock lock(signal_lock_);
 
   // We stop observing our port as soon as it's transferred, but this can race
   // with events which are raised right before that happens. This is fine to
   // ignore.
   if (port_transferred_)
     return;
 
 #if !defined(NDEBUG)
   ports::PortStatus port_status;
   node_controller_->node()->GetStatus(port_, &port_status);
   if (port_status.has_messages) {
     ports::ScopedMessage unused;
     size_t message_size = 0;
     node_controller_->node()->GetMessageIf(
         port_, [&message_size](const ports::Message& message) {
           message_size = message.num_payload_bytes();
           return false;
         }, &unused);
     DVLOG(1) << "New message detected on message pipe " << pipe_id_
              << " endpoint " << endpoint_ << " [port=" << port_.name()
              << "; size=" << message_size << "]";
   }
   if (port_status.peer_closed) {
     DVLOG(1) << "Peer closure detected on message pipe " << pipe_id_
              << " endpoint " << endpoint_ << " [port=" << port_.name() << "]";
   }
 #endif
 
-  awakables_.AwakeForStateChange(GetHandleSignalsStateNoLock());
+  NotifyObserversForStateChangeNoLock(&request_context);
+}
+
+void MessagePipeDispatcher::NotifyObserversForStateChangeNoLock(
+    RequestContext* request_context) {
+  HandleSignalsState state = GetHandleSignalsStateNoLock();
+  awakables_.AwakeForStateChange(state);
+  watchers_.NotifyOfStateChange(state, request_context);
 }
 
 }  // namespace edk
 }  // namespace mojo