EDK: Add implementation of data pipe consumer read threshold stuff.

mojo/edk/system/remote_producer_data_pipe_impl.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/remote_producer_data_pipe_impl.h"
 
 #include <string.h>
 
 #include <algorithm>
 #include <memory>
@@ skipping 263 matching lines @@
   DCHECK_EQ(num_bytes_read % element_num_bytes(), 0u);
   DCHECK_LE(start_index_ + num_bytes_read, capacity_num_bytes());
   MarkDataAsConsumed(num_bytes_read);
   set_consumer_two_phase_max_num_bytes_read(0);
   return MOJO_RESULT_OK;
 }
 
 HandleSignalsState RemoteProducerDataPipeImpl::ConsumerGetHandleSignalsState()
     const {
   HandleSignalsState rv;
-  if (current_num_bytes_ > 0) {
+  // |consumer_read_threshold_num_bytes()| is always at least 1.
+  if (current_num_bytes_ >= consumer_read_threshold_num_bytes()) {
+    if (!consumer_in_two_phase_read()) {
+      rv.satisfied_signals |=
+          MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_READ_THRESHOLD;
+    }
+    rv.satisfiable_signals |=
+        MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_READ_THRESHOLD;
+  } else if (current_num_bytes_ > 0u) {
     if (!consumer_in_two_phase_read())
       rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_READABLE;
     rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
-  } else if (producer_open()) {
-    rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
   }
-  if (!producer_open())
+  if (producer_open()) {
+    rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE |
+                              MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                              MOJO_HANDLE_SIGNAL_READ_THRESHOLD;
+  } else {
     rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
-  rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
+    rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
+  }
   return rv;
 }
 
 void RemoteProducerDataPipeImpl::ConsumerStartSerialize(
     Channel* channel,
     size_t* max_size,
     size_t* max_platform_handles) {
   *max_size = sizeof(SerializedDataPipeConsumerDispatcher) +
               channel->GetSerializedEndpointSize();
   *max_platform_handles = 0;
@@ skipping 158 matching lines @@
   // If the consumer is still open and we still have data, we have to keep the
   // buffer around. Currently, we won't free it even if it empties later. (We
   // could do this -- requiring a check on every read -- but that seems to be
   // optimizing for the uncommon case.)
   if (!consumer_open() || !current_num_bytes_)
     DestroyBuffer();
 }
 
 }  // namespace system
 }  // namespace mojo