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

mojo/edk/system/data_pipe.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.
 
 #include "mojo/edk/system/data_pipe.h"
 
 #include <string.h>
 
 #include <algorithm>
 #include <limits>
@@ skipping 440 matching lines @@
   MutexLocker locker(&mutex_);
   DCHECK(has_local_consumer_no_lock());
   consumer_awakable_list_->CancelAll();
 }
 
 void DataPipe::ConsumerClose() {
   MutexLocker locker(&mutex_);
   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,
                                       bool peek) {
   MutexLocker locker(&mutex_);
   DCHECK(has_local_consumer_no_lock());
 
   if (consumer_in_two_phase_read_no_lock())
     return MOJO_RESULT_BUSY;
 
@@ skipping 185 matching lines @@
 }
 
 DataPipe::DataPipe(bool has_local_producer,
                    bool has_local_consumer,
                    const MojoCreateDataPipeOptions& validated_options,
                    std::unique_ptr<DataPipeImpl> impl)
     : element_num_bytes_(validated_options.element_num_bytes),
       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()
                                                  : nullptr),
       producer_two_phase_max_num_bytes_written_(0),
       consumer_two_phase_max_num_bytes_read_(0),
       impl_(std::move(impl)) {
   impl_->set_owner(this);
 
 #if !defined(NDEBUG) || defined(DCHECK_ALWAYS_ON)
@@ skipping 134 matching lines @@
   SetProducerClosedNoLock();
 }
 
 void DataPipe::SetConsumerClosed() {
   MutexLocker locker(&mutex_);
   SetConsumerClosedNoLock();
 }
 
 }  // namespace system
 }  // namespace mojo