Mojo: Add signal for new data pipe consumer data

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 106 matching lines @@
   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;
 
   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;
@@ skipping 60 matching lines @@
     NotifyRead(bytes_to_read);
   }
 
   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))
@@ skipping 195 matching lines @@
 
   scoped_refptr<DataPipeConsumerDispatcher> dispatcher =
       new DataPipeConsumerDispatcher(node_controller, port, ring_buffer,
                                      state->options, false /* initialized */,
                                      state->pipe_id);
 
   {
     base::AutoLock lock(dispatcher->lock_);
     dispatcher->read_offset_ = state->read_offset;
     dispatcher->bytes_available_ = state->bytes_available;
+    dispatcher->new_data_available_ = state->bytes_available > 0;
     dispatcher->peer_closed_ = state->flags & kFlagPeerClosed;
     dispatcher->InitializeNoLock();
     dispatcher->UpdateSignalsStateNoLock();
   }
 
   return dispatcher;
 }
 
 DataPipeConsumerDispatcher::~DataPipeConsumerDispatcher() {
   DCHECK(is_closed_ && !shared_ring_buffer_ && !ring_buffer_mapping_ &&
@@ skipping 35 matching lines @@
 
   return MOJO_RESULT_OK;
 }
 
 HandleSignalsState
 DataPipeConsumerDispatcher::GetHandleSignalsStateNoLock() const {
   lock_.AssertAcquired();
 
   HandleSignalsState rv;
   if (shared_ring_buffer_ && bytes_available_) {
-    if (!in_two_phase_read_)
+    if (!in_two_phase_read_) {
       rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_READABLE;
+      if (new_data_available_)
+        rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE;
+    }
     rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
   } else if (!peer_closed_ && shared_ring_buffer_) {
     rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
   }
 
+  if (shared_ring_buffer_) {
+    if (new_data_available_ || !peer_closed_)
+      rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE;
+  }
+
   if (peer_closed_)
     rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
   rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
+
   return rv;
 }
 
 void DataPipeConsumerDispatcher::NotifyRead(uint32_t num_bytes) {
   DVLOG(1) << "Data pipe consumer " << pipe_id_ << " notifying peer: "
            << num_bytes << " bytes read. [control_port="
            << control_port_.name() << "]";
 
   SendDataPipeControlMessage(node_controller_, control_port_,
                              DataPipeCommand::DATA_WAS_READ, num_bytes);
@@ skipping 59 matching lines @@
 
         DVLOG(1) << "Data pipe consumer " << pipe_id_ << " is aware that "
                  << m->num_bytes << " bytes were written. [control_port="
                  << control_port_.name() << "]";
 
         bytes_available_ += m->num_bytes;
       }
     } while (message);
   }
 
-  if (peer_closed_ != was_peer_closed ||
-      bytes_available_ != previous_bytes_available) {
+  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());
   }
 }
 
 }  // namespace edk
 }  // namespace mojo