Mojo: Move waiting APIs to public library

mojo/edk/system/data_pipe_producer_dispatcher.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/edk/system/data_pipe_producer_dispatcher.h"
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <utility>
@@ skipping 110 matching lines @@
     // Don't return "should wait" since you can't wait for a specified amount of
     // data.
     return MOJO_RESULT_OUT_OF_RANGE;
   }
 
   DCHECK_LE(available_capacity_, options_.capacity_num_bytes);
   uint32_t num_bytes_to_write = std::min(*num_bytes, available_capacity_);
   if (num_bytes_to_write == 0)
     return MOJO_RESULT_SHOULD_WAIT;
 
-  HandleSignalsState old_state = GetHandleSignalsStateNoLock();
-
   *num_bytes = num_bytes_to_write;
 
   CHECK(ring_buffer_mapping_);
   uint8_t* data = static_cast<uint8_t*>(ring_buffer_mapping_->GetBase());
   CHECK(data);
 
   const uint8_t* source = static_cast<const uint8_t*>(elements);
   CHECK(source);
 
   DCHECK_LE(write_offset_, options_.capacity_num_bytes);
   uint32_t tail_bytes_to_write =
       std::min(options_.capacity_num_bytes - write_offset_,
                num_bytes_to_write);
   uint32_t head_bytes_to_write = num_bytes_to_write - tail_bytes_to_write;
 
   DCHECK_GT(tail_bytes_to_write, 0u);
   memcpy(data + write_offset_, source, tail_bytes_to_write);
   if (head_bytes_to_write > 0)
     memcpy(data, source + tail_bytes_to_write, head_bytes_to_write);
 
   DCHECK_LE(num_bytes_to_write, available_capacity_);
   available_capacity_ -= num_bytes_to_write;
   write_offset_ = (write_offset_ + num_bytes_to_write) %
       options_.capacity_num_bytes;
 
-  HandleSignalsState new_state = GetHandleSignalsStateNoLock();
-  if (!new_state.equals(old_state))
-    awakable_list_.AwakeForStateChange(new_state);
-  watchers_.NotifyState(new_state);
+  watchers_.NotifyState(GetHandleSignalsStateNoLock());
 
   base::AutoUnlock unlock(lock_);
   NotifyWrite(num_bytes_to_write);
 
   return MOJO_RESULT_OK;
 }
 
 MojoResult DataPipeProducerDispatcher::BeginWriteData(
     void** buffer,
     uint32_t* buffer_num_bytes,
@@ skipping 53 matching lines @@
     write_offset_ = (write_offset_ + num_bytes_written) %
         options_.capacity_num_bytes;
 
     base::AutoUnlock unlock(lock_);
     NotifyWrite(num_bytes_written);
   }
 
   in_two_phase_write_ = false;
 
   // If we're now writable, we *became* writable (since we weren't writable
-  // during the two-phase write), so awake producer awakables.
-  HandleSignalsState new_state = GetHandleSignalsStateNoLock();
-  if (new_state.satisfies(MOJO_HANDLE_SIGNAL_WRITABLE))
-    awakable_list_.AwakeForStateChange(new_state);
-  watchers_.NotifyState(new_state);
+  // during the two-phase write), so notify watchers.
+  watchers_.NotifyState(GetHandleSignalsStateNoLock());
 
   return rv;
 }
 
 HandleSignalsState DataPipeProducerDispatcher::GetHandleSignalsState() const {
   base::AutoLock lock(lock_);
   return GetHandleSignalsStateNoLock();
 }
 
 MojoResult DataPipeProducerDispatcher::AddWatcherRef(
     const scoped_refptr<WatcherDispatcher>& watcher,
     uintptr_t context) {
   base::AutoLock lock(lock_);
   if (is_closed_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
   return watchers_.Add(watcher, context, GetHandleSignalsStateNoLock());
 }
 
 MojoResult DataPipeProducerDispatcher::RemoveWatcherRef(
     WatcherDispatcher* watcher,
     uintptr_t context) {
   base::AutoLock lock(lock_);
   if (is_closed_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
   return watchers_.Remove(watcher, context);
 }
 
-MojoResult DataPipeProducerDispatcher::AddAwakable(
-    Awakable* awakable,
-    MojoHandleSignals signals,
-    uintptr_t context,
-    HandleSignalsState* signals_state) {
-  base::AutoLock lock(lock_);
-  if (!shared_ring_buffer_ || in_transit_) {
-    if (signals_state)
-      *signals_state = HandleSignalsState();
-    return MOJO_RESULT_INVALID_ARGUMENT;
-  }
-  UpdateSignalsStateNoLock();
-  HandleSignalsState state = GetHandleSignalsStateNoLock();
-  if (state.satisfies(signals)) {
-    if (signals_state)
-      *signals_state = state;
-    return MOJO_RESULT_ALREADY_EXISTS;
-  }
-  if (!state.can_satisfy(signals)) {
-    if (signals_state)
-      *signals_state = state;
-    return MOJO_RESULT_FAILED_PRECONDITION;
-  }
-
-  awakable_list_.Add(awakable, signals, context);
-  return MOJO_RESULT_OK;
-}
-
-void DataPipeProducerDispatcher::RemoveAwakable(
-    Awakable* awakable,
-    HandleSignalsState* signals_state) {
-  base::AutoLock lock(lock_);
-  if ((!shared_ring_buffer_ || in_transit_) && signals_state)
-    *signals_state = HandleSignalsState();
-  else if (signals_state)
-    *signals_state = GetHandleSignalsStateNoLock();
-  awakable_list_.Remove(awakable);
-}
-
 void DataPipeProducerDispatcher::StartSerialize(uint32_t* num_bytes,
                                                 uint32_t* num_ports,
                                                 uint32_t* num_handles) {
   base::AutoLock lock(lock_);
   DCHECK(in_transit_);
   *num_bytes = sizeof(SerializedState);
   *num_ports = 1;
   *num_handles = 1;
 }
 
@@ skipping 39 matching lines @@
   CloseNoLock();
 }
 
 void DataPipeProducerDispatcher::CancelTransit() {
   base::AutoLock lock(lock_);
   DCHECK(in_transit_);
   in_transit_ = false;
   buffer_handle_for_transit_.reset();
 
   HandleSignalsState state = GetHandleSignalsStateNoLock();
-  awakable_list_.AwakeForStateChange(state);
   watchers_.NotifyState(state);
 }
 
 // static
 scoped_refptr<DataPipeProducerDispatcher>
 DataPipeProducerDispatcher::Deserialize(const void* data,
                                         size_t num_bytes,
                                         const ports::PortName* ports,
                                         size_t num_ports,
                                         PlatformHandle* handles,
@@ skipping 63 matching lines @@
 }
 
 MojoResult DataPipeProducerDispatcher::CloseNoLock() {
   lock_.AssertAcquired();
   if (is_closed_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
   is_closed_ = true;
   ring_buffer_mapping_.reset();
   shared_ring_buffer_ = nullptr;
 
-  awakable_list_.CancelAll();
   watchers_.NotifyClosed();
   if (!transferred_) {
     base::AutoUnlock unlock(lock_);
     node_controller_->ClosePort(control_port_);
   }
 
   return MOJO_RESULT_OK;
 }
 
 HandleSignalsState DataPipeProducerDispatcher::GetHandleSignalsStateNoLock()
@@ skipping 81 matching lines @@
                  << m->num_bytes << " bytes were read. [control_port="
                  << control_port_.name() << "]";
 
         available_capacity_ += m->num_bytes;
       }
     } while (message);
   }
 
   if (peer_closed_ != was_peer_closed ||
       available_capacity_ != previous_capacity) {
-    HandleSignalsState state = GetHandleSignalsStateNoLock();
-    awakable_list_.AwakeForStateChange(state);
-    watchers_.NotifyState(state);
+    watchers_.NotifyState(GetHandleSignalsStateNoLock());
   }
 }
 
 }  // namespace edk
 }  // namespace mojo