Mojo: Armed Watchers

mojo/edk/system/data_pipe_consumer_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_consumer_dispatcher.h"
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <algorithm>
@@ skipping 62 matching lines @@
     NodeController* node_controller,
     const ports::PortRef& control_port,
     scoped_refptr<PlatformSharedBuffer> shared_ring_buffer,
     const MojoCreateDataPipeOptions& options,
     bool initialized,
     uint64_t pipe_id)
     : options_(options),
       node_controller_(node_controller),
       control_port_(control_port),
       pipe_id_(pipe_id),
+      watchers_(this),
       shared_ring_buffer_(shared_ring_buffer) {
   if (initialized) {
     base::AutoLock lock(lock_);
     InitializeNoLock();
   }
 }
 
 Dispatcher::Type DataPipeConsumerDispatcher::GetType() const {
   return Type::DATA_PIPE_CONSUMER;
 }
 
 MojoResult DataPipeConsumerDispatcher::Close() {
   base::AutoLock lock(lock_);
   DVLOG(1) << "Closing data pipe consumer " << pipe_id_;
   return CloseNoLock();
 }
 
-
-MojoResult DataPipeConsumerDispatcher::Watch(
-    MojoHandleSignals signals,
-    const Watcher::WatchCallback& callback,
-    uintptr_t context) {
-  base::AutoLock lock(lock_);
-
-  if (is_closed_ || in_transit_)
-    return MOJO_RESULT_INVALID_ARGUMENT;
-
-  return awakable_list_.AddWatcher(
-      signals, callback, context, GetHandleSignalsStateNoLock());
-}
-
-MojoResult DataPipeConsumerDispatcher::CancelWatch(uintptr_t context) {
-  base::AutoLock lock(lock_);
-
-  if (is_closed_ || in_transit_)
-    return MOJO_RESULT_INVALID_ARGUMENT;
-
-  return awakable_list_.RemoveWatcher(context);
-}
-
 MojoResult DataPipeConsumerDispatcher::ReadData(void* elements,
                                                 uint32_t* num_bytes,
                                                 MojoReadDataFlags flags) {
   base::AutoLock lock(lock_);
-  new_data_available_ = false;
 
   if (!shared_ring_buffer_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (in_two_phase_read_)
     return MOJO_RESULT_BUSY;
 
+  const bool had_new_data = new_data_available_;
+  new_data_available_ = false;
+
   if ((flags & MOJO_READ_DATA_FLAG_QUERY)) {
     if ((flags & MOJO_READ_DATA_FLAG_PEEK) ||
         (flags & MOJO_READ_DATA_FLAG_DISCARD))
       return MOJO_RESULT_INVALID_ARGUMENT;
     DCHECK(!(flags & MOJO_READ_DATA_FLAG_DISCARD));  // Handled above.
     DVLOG_IF(2, elements)
         << "Query mode: ignoring non-null |elements|";
     *num_bytes = static_cast<uint32_t>(bytes_available_);
+
+    if (had_new_data)
+      watchers_.NotifyState(GetHandleSignalsStateNoLock());
     return MOJO_RESULT_OK;
   }
 
   bool discard = false;
   if ((flags & MOJO_READ_DATA_FLAG_DISCARD)) {
     // These flags are mutally exclusive.
     if (flags & MOJO_READ_DATA_FLAG_PEEK)
       return MOJO_RESULT_INVALID_ARGUMENT;
     DVLOG_IF(2, elements)
         << "Discard mode: ignoring non-null |elements|";
     discard = true;
   }
 
   uint32_t max_num_bytes_to_read = *num_bytes;
   if (max_num_bytes_to_read % options_.element_num_bytes != 0)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   bool all_or_none = flags & MOJO_READ_DATA_FLAG_ALL_OR_NONE;
   uint32_t min_num_bytes_to_read =
       all_or_none ? max_num_bytes_to_read : 0;
 
   if (min_num_bytes_to_read > bytes_available_) {
+    if (had_new_data)
+      watchers_.NotifyState(GetHandleSignalsStateNoLock());
     return peer_closed_ ? MOJO_RESULT_FAILED_PRECONDITION
                         : MOJO_RESULT_OUT_OF_RANGE;
   }
 
   uint32_t bytes_to_read = std::min(max_num_bytes_to_read, bytes_available_);
   if (bytes_to_read == 0) {
+    if (had_new_data)
+      watchers_.NotifyState(GetHandleSignalsStateNoLock());
     return peer_closed_ ? MOJO_RESULT_FAILED_PRECONDITION
                         : MOJO_RESULT_SHOULD_WAIT;
   }
 
   if (!discard) {
     uint8_t* data = static_cast<uint8_t*>(ring_buffer_mapping_->GetBase());
     CHECK(data);
 
     uint8_t* destination = static_cast<uint8_t*>(elements);
     CHECK(destination);
@@ skipping 11 matching lines @@
 
   bool peek = !!(flags & MOJO_READ_DATA_FLAG_PEEK);
   if (discard || !peek) {
     read_offset_ = (read_offset_ + bytes_to_read) % options_.capacity_num_bytes;
     bytes_available_ -= bytes_to_read;
 
     base::AutoUnlock unlock(lock_);
     NotifyRead(bytes_to_read);
   }
 
+  // We may have just read the last available data and thus changed the signals
+  // state.
+  watchers_.NotifyState(GetHandleSignalsStateNoLock());
+
   return MOJO_RESULT_OK;
 }
 
 MojoResult DataPipeConsumerDispatcher::BeginReadData(const void** buffer,
                                                      uint32_t* buffer_num_bytes,
                                                      MojoReadDataFlags flags) {
   base::AutoLock lock(lock_);
-  new_data_available_ = false;
   if (!shared_ring_buffer_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (in_two_phase_read_)
     return MOJO_RESULT_BUSY;
 
   // These flags may not be used in two-phase mode.
   if ((flags & MOJO_READ_DATA_FLAG_DISCARD) ||
       (flags & MOJO_READ_DATA_FLAG_QUERY) ||
       (flags & MOJO_READ_DATA_FLAG_PEEK))
     return MOJO_RESULT_INVALID_ARGUMENT;
 
+  const bool had_new_data = new_data_available_;
+  new_data_available_ = false;
+
   if (bytes_available_ == 0) {
+    if (had_new_data)
+      watchers_.NotifyState(GetHandleSignalsStateNoLock());
     return peer_closed_ ? MOJO_RESULT_FAILED_PRECONDITION
                         : MOJO_RESULT_SHOULD_WAIT;
   }
 
   DCHECK_LT(read_offset_, options_.capacity_num_bytes);
   uint32_t bytes_to_read = std::min(bytes_available_,
                                     options_.capacity_num_bytes - read_offset_);
 
   CHECK(ring_buffer_mapping_);
   uint8_t* data = static_cast<uint8_t*>(ring_buffer_mapping_->GetBase());
   CHECK(data);
 
   in_two_phase_read_ = true;
   *buffer = data + read_offset_;
   *buffer_num_bytes = bytes_to_read;
   two_phase_max_bytes_read_ = bytes_to_read;
 
+  if (had_new_data)
+    watchers_.NotifyState(GetHandleSignalsStateNoLock());
+
   return MOJO_RESULT_OK;
 }
 
 MojoResult DataPipeConsumerDispatcher::EndReadData(uint32_t num_bytes_read) {
   base::AutoLock lock(lock_);
   if (!in_two_phase_read_)
     return MOJO_RESULT_FAILED_PRECONDITION;
 
   if (in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
@@ skipping 16 matching lines @@
     base::AutoUnlock unlock(lock_);
     NotifyRead(num_bytes_read);
   }
 
   in_two_phase_read_ = false;
   two_phase_max_bytes_read_ = 0;
 
   HandleSignalsState new_state = GetHandleSignalsStateNoLock();
   if (!new_state.equals(old_state))
     awakable_list_.AwakeForStateChange(new_state);
+  watchers_.NotifyState(new_state);
 
   return rv;
 }
 
 HandleSignalsState DataPipeConsumerDispatcher::GetHandleSignalsState() const {
   base::AutoLock lock(lock_);
   return GetHandleSignalsStateNoLock();
 }
 
+MojoResult DataPipeConsumerDispatcher::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 DataPipeConsumerDispatcher::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 DataPipeConsumerDispatcher::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;
@@ skipping 162 matching lines @@
 
 MojoResult DataPipeConsumerDispatcher::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
 DataPipeConsumerDispatcher::GetHandleSignalsStateNoLock() const {
@@ skipping 97 matching lines @@
         bytes_available_ += m->num_bytes;
       }
     } while (message);
   }
 
   bool has_new_data = bytes_available_ != previous_bytes_available;
   if (has_new_data)
     new_data_available_ = true;
 
   if (peer_closed_ != was_peer_closed || has_new_data) {
-    awakable_list_.AwakeForStateChange(GetHandleSignalsStateNoLock());
+    HandleSignalsState state = GetHandleSignalsStateNoLock();
+    awakable_list_.AwakeForStateChange(state);
+    watchers_.NotifyState(state);
   }
 }
 
 }  // namespace edk
 }  // namespace mojo