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

mojo/edk/system/data_pipe.h

Index: mojo/edk/system/data_pipe.h
diff --git a/mojo/edk/system/data_pipe.h b/mojo/edk/system/data_pipe.h
index 3ac3b379e0c50101e65d79ca150b1d0a941de9f1..f559511c5a57a6209000e46eda2c1e491da481e8 100644
--- a/mojo/edk/system/data_pipe.h
+++ b/mojo/edk/system/data_pipe.h
@@ -132,6 +132,8 @@ class DataPipe final : public ChannelEndpointClient {
   // corresponding names.
   void ConsumerCancelAllAwakables();
   void ConsumerClose();
+  MojoResult ConsumerSetOptions(uint32_t read_threshold_num_bytes);
+  void ConsumerGetOptions(uint32_t* read_threshold_num_bytes);
   // This does not validate its arguments, except to check that |*num_bytes| is
   // a multiple of |element_num_bytes_|.
   MojoResult ConsumerReadData(UserPointer<void> elements,
@@ -195,6 +197,14 @@ class DataPipe final : public ChannelEndpointClient {
     mutex_.AssertHeld();
     return consumer_open_;
   }
+  // Note that this returns the "real" read threshold (never zero).
+  size_t consumer_read_threshold_num_bytes_no_lock() const
+      MOJO_SHARED_LOCKS_REQUIRED(mutex_) {
+    mutex_.AssertHeld();
+    return (consumer_read_threshold_num_bytes_ > 0u)
+               ? consumer_read_threshold_num_bytes_
+               : element_num_bytes_;
+  }
   uint32_t producer_two_phase_max_num_bytes_written_no_lock() const
       MOJO_SHARED_LOCKS_REQUIRED(mutex_) {
     mutex_.AssertHeld();
@@ -271,6 +281,8 @@ class DataPipe final : public ChannelEndpointClient {
   // *Known* state of producer or consumer.
   bool producer_open_ MOJO_GUARDED_BY(mutex_);
   bool consumer_open_ MOJO_GUARDED_BY(mutex_);
+  // This may be zero (in which case it "means" |element_num_bytes_|).
+  uint32_t consumer_read_threshold_num_bytes_ MOJO_GUARDED_BY(mutex_);
   // Non-null only if the producer or consumer, respectively, is local.
   std::unique_ptr<AwakableList> producer_awakable_list_ MOJO_GUARDED_BY(mutex_);
   std::unique_ptr<AwakableList> consumer_awakable_list_ MOJO_GUARDED_BY(mutex_);