Mojo: Armed Watchers

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 61 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),
       available_capacity_(options_.capacity_num_bytes) {
   if (initialized) {
     base::AutoLock lock(lock_);
     InitializeNoLock();
   }
 }
 
 Dispatcher::Type DataPipeProducerDispatcher::GetType() const {
   return Type::DATA_PIPE_PRODUCER;
 }
 
 MojoResult DataPipeProducerDispatcher::Close() {
   base::AutoLock lock(lock_);
   DVLOG(1) << "Closing data pipe producer " << pipe_id_;
   return CloseNoLock();
 }
 
-MojoResult DataPipeProducerDispatcher::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 DataPipeProducerDispatcher::CancelWatch(uintptr_t context) {
-  base::AutoLock lock(lock_);
-
-  if (is_closed_ || in_transit_)
-    return MOJO_RESULT_INVALID_ARGUMENT;
-
-  return awakable_list_.RemoveWatcher(context);
-}
-
 MojoResult DataPipeProducerDispatcher::WriteData(const void* elements,
                                                  uint32_t* num_bytes,
                                                  MojoWriteDataFlags flags) {
   base::AutoLock lock(lock_);
   if (!shared_ring_buffer_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (in_two_phase_write_)
     return MOJO_RESULT_BUSY;
 
   if (peer_closed_)
     return MOJO_RESULT_FAILED_PRECONDITION;
 
   if (*num_bytes % options_.element_num_bytes != 0)
     return MOJO_RESULT_INVALID_ARGUMENT;
   if (*num_bytes == 0)
     return MOJO_RESULT_OK;  // Nothing to do.
 
   if ((flags & MOJO_WRITE_DATA_FLAG_ALL_OR_NONE) &&
       (*num_bytes > available_capacity_)) {
+    if (flags & MOJO_WRITE_DATA_FLAG_CLEAR_SIGNAL) {
+      suppress_writable_signal_ = true;
+      watchers_.NotifyState(GetHandleSignalsStateNoLock());
+    }
     // Don't return "should wait" since you can't wait for a specified amount of
     // data.
+    //
+    // TODO(rockot): Reconsider this case now that we have support for minimum
+    // signaling capacity.
     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();
 
@@ skipping 18 matching lines @@
     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);
 
   base::AutoUnlock unlock(lock_);
   NotifyWrite(num_bytes_to_write);
 
   return MOJO_RESULT_OK;
 }
 
 MojoResult DataPipeProducerDispatcher::BeginWriteData(
     void** buffer,
     uint32_t* buffer_num_bytes,
     MojoWriteDataFlags flags) {
   base::AutoLock lock(lock_);
   if (!shared_ring_buffer_ || in_transit_)
     return MOJO_RESULT_INVALID_ARGUMENT;
+
+  // These flags may not be used in two-phase mode.
+  if ((flags & MOJO_WRITE_DATA_FLAG_ALL_OR_NONE) ||
+      (flags & MOJO_WRITE_DATA_FLAG_CLEAR_SIGNAL))
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
   if (in_two_phase_write_)
     return MOJO_RESULT_BUSY;
   if (peer_closed_)
     return MOJO_RESULT_FAILED_PRECONDITION;
 
   if (available_capacity_ == 0) {
     return peer_closed_ ? MOJO_RESULT_FAILED_PRECONDITION
                         : MOJO_RESULT_SHOULD_WAIT;
   }
 
@@ skipping 38 matching lines @@
     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);
 
   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;
@@ skipping 76 matching lines @@
   in_transit_ = false;
   ignore_result(buffer_handle_for_transit_.release());
   CloseNoLock();
 }
 
 void DataPipeProducerDispatcher::CancelTransit() {
   base::AutoLock lock(lock_);
   DCHECK(in_transit_);
   in_transit_ = false;
   buffer_handle_for_transit_.reset();
-  awakable_list_.AwakeForStateChange(GetHandleSignalsStateNoLock());
+
+  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,
                                         size_t num_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()
     const {
   lock_.AssertAcquired();
   HandleSignalsState rv;
   if (!peer_closed_) {
     if (!in_two_phase_write_ && shared_ring_buffer_ &&
-        available_capacity_ > 0)
+        available_capacity_ > 0 && !suppress_writable_signal_)
       rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
     rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
   } else {
     rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
   }
   rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
   return rv;
 }
 
 void DataPipeProducerDispatcher::NotifyWrite(uint32_t num_bytes) {
@@ skipping 66 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) {
-    awakable_list_.AwakeForStateChange(GetHandleSignalsStateNoLock());
+    suppress_writable_signal_ = false;
+    HandleSignalsState state = GetHandleSignalsStateNoLock();
+    awakable_list_.AwakeForStateChange(state);
+    watchers_.NotifyState(state);
   }
 }
 
 }  // namespace edk
 }  // namespace mojo