Mojo: Move waiting APIs to public library

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;
@@ skipping 195 matching lines @@
 }
 
 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,
-    uintptr_t context,
-    HandleSignalsState* signals_state) {
-  base::AutoLock lock(signal_lock_);
-
-  if (port_closed_ || in_transit_) {
-    if (signals_state)
-      *signals_state = HandleSignalsState();
-    return MOJO_RESULT_INVALID_ARGUMENT;
-  }
-
-  HandleSignalsState state = GetHandleSignalsStateNoLock();
-
-  DVLOG(2) << "Getting signal state for pipe " << pipe_id_ << " endpoint "
-           << endpoint_ << " [awakable=" << awakable << "; port="
-           << port_.name() << "; signals=" << signals << "; satisfied="
-           << state.satisfied_signals << "; satisfiable="
-           << state.satisfiable_signals << "]";
-
-  if (state.satisfies(signals)) {
-    if (signals_state)
-      *signals_state = state;
-    DVLOG(2) << "Signals already set for " << port_.name();
-    return MOJO_RESULT_ALREADY_EXISTS;
-  }
-  if (!state.can_satisfy(signals)) {
-    if (signals_state)
-      *signals_state = state;
-    DVLOG(2) << "Signals impossible to satisfy for " << port_.name();
-    return MOJO_RESULT_FAILED_PRECONDITION;
-  }
-
-  DVLOG(2) << "Adding awakable to pipe " << pipe_id_ << " endpoint "
-           << endpoint_ << " [awakable=" << awakable << "; port="
-           << port_.name() << "; signals=" << signals << "]";
-
-  awakables_.Add(awakable, signals, context);
-  return MOJO_RESULT_OK;
-}
-
-void MessagePipeDispatcher::RemoveAwakable(Awakable* awakable,
-                                           HandleSignalsState* signals_state) {
-  base::AutoLock lock(signal_lock_);
-  if (port_closed_ || in_transit_) {
-    if (signals_state)
-      *signals_state = HandleSignalsState();
-  } else if (signals_state) {
-    *signals_state = GetHandleSignalsStateNoLock();
-  }
-
-  DVLOG(2) << "Removing awakable from pipe " << pipe_id_ << " endpoint "
-           << endpoint_ << " [awakable=" << awakable << "; port="
-           << port_.name() << "]";
-
-  awakables_.Remove(awakable);
-}
-
 void MessagePipeDispatcher::StartSerialize(uint32_t* num_bytes,
                                            uint32_t* num_ports,
                                            uint32_t* num_handles) {
   *num_bytes = static_cast<uint32_t>(sizeof(SerializedState));
   *num_ports = 1;
   *num_handles = 0;
 }
 
 bool MessagePipeDispatcher::EndSerialize(void* destination,
                                          ports::PortName* ports,
@@ skipping 21 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.
-  HandleSignalsState state = GetHandleSignalsStateNoLock();
-  awakables_.AwakeForStateChange(state);
-  watchers_.NotifyState(state);
+  watchers_.NotifyState(GetHandleSignalsStateNoLock());
 }
 
 // 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 14 matching lines @@
 MessagePipeDispatcher::~MessagePipeDispatcher() {
   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;
 }
 
@@ skipping 45 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
 
-  HandleSignalsState state = GetHandleSignalsStateNoLock();
-  awakables_.AwakeForStateChange(state);
-  watchers_.NotifyState(state);
+  watchers_.NotifyState(GetHandleSignalsStateNoLock());
 }
 
 }  // namespace edk
 }  // namespace mojo