EDK: Add implementation of data pipe consumer read 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 61a8793a1f9c1cbce1a6a5d41c6a1d8277172754..9a90c8a549cb6d0cdf6754313a6d3bcb09878bc4 100644
--- a/mojo/edk/system/data_pipe.cc
+++ b/mojo/edk/system/data_pipe.cc
@@ -458,6 +458,29 @@ void DataPipe::ConsumerClose() {
   ConsumerCloseNoLock();
 }
 
+MojoResult DataPipe::ConsumerSetOptions(uint32_t read_threshold_num_bytes) {
+  MutexLocker locker(&mutex_);
+  DCHECK(has_local_consumer_no_lock());
+
+  if (read_threshold_num_bytes % element_num_bytes() != 0)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  HandleSignalsState old_consumer_state =
+      impl_->ConsumerGetHandleSignalsState();
+  consumer_read_threshold_num_bytes_ = read_threshold_num_bytes;
+  HandleSignalsState new_consumer_state =
+      impl_->ConsumerGetHandleSignalsState();
+  if (!new_consumer_state.equals(old_consumer_state))
+    AwakeConsumerAwakablesForStateChangeNoLock(new_consumer_state);
+  return MOJO_RESULT_OK;
+}
+
+void DataPipe::ConsumerGetOptions(uint32_t* read_threshold_num_bytes) {
+  MutexLocker locker(&mutex_);
+  DCHECK(has_local_consumer_no_lock());
+  *read_threshold_num_bytes = consumer_read_threshold_num_bytes_;
+}
+
 MojoResult DataPipe::ConsumerReadData(UserPointer<void> elements,
                                       UserPointer<uint32_t> num_bytes,
                                       bool all_or_none,
@@ -663,6 +686,7 @@ DataPipe::DataPipe(bool has_local_producer,
       capacity_num_bytes_(validated_options.capacity_num_bytes),
       producer_open_(true),
       consumer_open_(true),
+      consumer_read_threshold_num_bytes_(0),
       producer_awakable_list_(has_local_producer ? new AwakableList()
                                                  : nullptr),
       consumer_awakable_list_(has_local_consumer ? new AwakableList()