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

mojo/edk/system/data_pipe.cc

Index: mojo/edk/system/data_pipe.cc
diff --git a/mojo/edk/system/data_pipe.cc b/mojo/edk/system/data_pipe.cc
index 9a90c8a549cb6d0cdf6754313a6d3bcb09878bc4..c42b6fd22b2be8884b622f45852fc25b15509655 100644
--- a/mojo/edk/system/data_pipe.cc
+++ b/mojo/edk/system/data_pipe.cc
@@ -281,6 +281,29 @@ void DataPipe::ProducerClose() {
   ProducerCloseNoLock();
 }
 
+MojoResult DataPipe::ProducerSetOptions(uint32_t write_threshold_num_bytes) {
+  MutexLocker locker(&mutex_);
+  DCHECK(has_local_producer_no_lock());
+
+  if (write_threshold_num_bytes % element_num_bytes() != 0)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  HandleSignalsState old_producer_state =
+      impl_->ProducerGetHandleSignalsState();
+  producer_write_threshold_num_bytes_ = write_threshold_num_bytes;
+  HandleSignalsState new_producer_state =
+      impl_->ProducerGetHandleSignalsState();
+  if (!new_producer_state.equals(old_producer_state))
+    AwakeProducerAwakablesForStateChangeNoLock(new_producer_state);
+  return MOJO_RESULT_OK;
+}
+
+void DataPipe::ProducerGetOptions(uint32_t* write_threshold_num_bytes) {
+  MutexLocker locker(&mutex_);
+  DCHECK(has_local_producer_no_lock());
+  *write_threshold_num_bytes = producer_write_threshold_num_bytes_;
+}
+
 MojoResult DataPipe::ProducerWriteData(UserPointer<const void> elements,
                                        UserPointer<uint32_t> num_bytes,
                                        bool all_or_none) {
@@ -686,6 +709,7 @@ DataPipe::DataPipe(bool has_local_producer,
       capacity_num_bytes_(validated_options.capacity_num_bytes),
       producer_open_(true),
       consumer_open_(true),
+      producer_write_threshold_num_bytes_(0),
       consumer_read_threshold_num_bytes_(0),
       producer_awakable_list_(has_local_producer ? new AwakableList()
                                                  : nullptr),