EDK: Add implementation of data pipe producer write threshold stuff.

mojo/edk/system/local_data_pipe_impl.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.
 
 // TODO(vtl): I currently potentially overflow in doing index calculations.
 // E.g., |start_index_| and |current_num_bytes_| fit into a |uint32_t|, but
 // their sum may not. This is bad and poses a security risk. (We're currently
 // saved by the limit on capacity -- the maximum size of the buffer, checked in
 // |DataPipe::ValidateOptions()|, is currently sufficiently small.)
 
@@ skipping 126 matching lines @@
   DCHECK_EQ(num_bytes_written % element_num_bytes(), 0u);
   current_num_bytes_ += num_bytes_written;
   DCHECK_LE(current_num_bytes_, capacity_num_bytes());
   set_producer_two_phase_max_num_bytes_written(0);
   return MOJO_RESULT_OK;
 }
 
 HandleSignalsState LocalDataPipeImpl::ProducerGetHandleSignalsState() const {
   HandleSignalsState rv;
   if (consumer_open()) {
-    if (current_num_bytes_ < capacity_num_bytes() &&
-        !producer_in_two_phase_write())
-      rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
-    rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
+    if (!producer_in_two_phase_write()) {
+      // |producer_write_threshold_num_bytes()| is always at least 1.
+      if (capacity_num_bytes() - current_num_bytes_ >=
+          producer_write_threshold_num_bytes()) {
+        rv.satisfied_signals |=
+            MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD;
+      } else if (current_num_bytes_ < capacity_num_bytes()) {
+        rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
+      }
+    }
+    rv.satisfiable_signals |=
+        MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD;
   } else {
     rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
   }
   rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
   return rv;
 }
 
 void LocalDataPipeImpl::ProducerStartSerialize(Channel* channel,
                                                size_t* max_size,
                                                size_t* max_platform_handles) {
@@ skipping 283 matching lines @@
 
 void LocalDataPipeImpl::MarkDataAsConsumed(size_t num_bytes) {
   DCHECK_LE(num_bytes, current_num_bytes_);
   start_index_ += num_bytes;
   start_index_ %= capacity_num_bytes();
   current_num_bytes_ -= num_bytes;
 }
 
 }  // namespace system
 }  // namespace mojo