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

mojo/edk/system/local_data_pipe_impl.cc

Index: mojo/edk/system/local_data_pipe_impl.cc
diff --git a/mojo/edk/system/local_data_pipe_impl.cc b/mojo/edk/system/local_data_pipe_impl.cc
index 7c05c2248eba5d23500c9ff2c66f833fc5ed92cc..dcb527cfbfa897127fd7fd33389ae9c6101f68dd 100644
--- a/mojo/edk/system/local_data_pipe_impl.cc
+++ b/mojo/edk/system/local_data_pipe_impl.cc
@@ -144,10 +144,18 @@ MojoResult LocalDataPipeImpl::ProducerEndWriteData(uint32_t num_bytes_written) {
 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;
   }