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

mojo/edk/system/core_unittest.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/core.h"
 
 #include <stdint.h>
 
 #include <limits>
 
 #include "mojo/edk/platform/thread_utils.h"
 #include "mojo/edk/system/awakable.h"
 #include "mojo/edk/system/core_test_base.h"
 #include "mojo/edk/system/test/timeouts.h"
 #include "mojo/public/cpp/system/macros.h"
 
 using mojo::platform::ThreadSleep;
 
 namespace mojo {
 namespace system {
 namespace {
 
 const MojoHandleSignalsState kEmptyMojoHandleSignalsState = {0u, 0u};
 const MojoHandleSignalsState kFullMojoHandleSignalsState = {~0u, ~0u};
-const MojoHandleSignals kAllSignals = MOJO_HANDLE_SIGNAL_READABLE |
-                                      MOJO_HANDLE_SIGNAL_WRITABLE |
-                                      MOJO_HANDLE_SIGNAL_PEER_CLOSED;
 
 using CoreTest = test::CoreTestBase;
 
 TEST_F(CoreTest, GetTimeTicksNow) {
   const MojoTimeTicks start = core()->GetTimeTicksNow();
   EXPECT_NE(static_cast<MojoTimeTicks>(0), start)
       << "GetTimeTicksNow should return nonzero value";
   ThreadSleep(test::DeadlineFromMilliseconds(15u));
   const MojoTimeTicks finish = core()->GetTimeTicksNow();
   // Allow for some fuzz in sleep.
@@ skipping 614 matching lines @@
                                   MOJO_HANDLE_SIGNAL_READABLE};
   result_index = static_cast<uint32_t>(-1);
   hss[0] = kEmptyMojoHandleSignalsState;
   hss[1] = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(
       MOJO_RESULT_DEADLINE_EXCEEDED,
       core()->WaitMany(MakeUserPointer(h), MakeUserPointer(signals), 2, 0,
                        MakeUserPointer(&result_index), MakeUserPointer(hss)));
   EXPECT_EQ(static_cast<uint32_t>(-1), result_index);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss[0].satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss[0].satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss[0].satisfiable_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss[1].satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss[1].satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss[1].satisfiable_signals);
 
   // Try to read anyway.
   char buffer[1] = {'a'};
   uint32_t buffer_size = 1;
   EXPECT_EQ(
       MOJO_RESULT_SHOULD_WAIT,
       core()->ReadMessage(h[0], UserPointer<void>(buffer),
                           MakeUserPointer(&buffer_size), NullUserPointer(),
                           NullUserPointer(), MOJO_READ_MESSAGE_FLAG_NONE));
   // Check that it left its inputs alone.
   EXPECT_EQ('a', buffer[0]);
   EXPECT_EQ(1u, buffer_size);
 
   // Both should be writable.
   hss[0] = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK, core()->Wait(h[0], MOJO_HANDLE_SIGNAL_WRITABLE,
                                          1000000000, MakeUserPointer(&hss[0])));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss[0].satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss[0].satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss[0].satisfiable_signals);
   hss[0] = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK, core()->Wait(h[1], MOJO_HANDLE_SIGNAL_WRITABLE,
                                          1000000000, MakeUserPointer(&hss[0])));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss[0].satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss[0].satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss[0].satisfiable_signals);
 
   // Also check that |h[1]| is writable using |WaitMany()|.
   signals[0] = MOJO_HANDLE_SIGNAL_READABLE;
   signals[1] = MOJO_HANDLE_SIGNAL_WRITABLE;
   result_index = static_cast<uint32_t>(-1);
   hss[0] = kEmptyMojoHandleSignalsState;
   hss[1] = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(
       MOJO_RESULT_OK,
       core()->WaitMany(MakeUserPointer(h), MakeUserPointer(signals), 2,
                        MOJO_DEADLINE_INDEFINITE, MakeUserPointer(&result_index),
                        MakeUserPointer(hss)));
   EXPECT_EQ(1u, result_index);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss[0].satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss[0].satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss[0].satisfiable_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss[1].satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss[1].satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss[1].satisfiable_signals);
 
   // Write to |h[1]|.
   buffer[0] = 'b';
   EXPECT_EQ(
       MOJO_RESULT_OK,
       core()->WriteMessage(h[1], UserPointer<const void>(buffer), 1,
                            NullUserPointer(), 0, MOJO_WRITE_MESSAGE_FLAG_NONE));
 
   // Check that |h[0]| is now readable.
   signals[0] = MOJO_HANDLE_SIGNAL_READABLE;
   signals[1] = MOJO_HANDLE_SIGNAL_READABLE;
   result_index = static_cast<uint32_t>(-1);
   hss[0] = kEmptyMojoHandleSignalsState;
   hss[1] = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(
       MOJO_RESULT_OK,
       core()->WaitMany(MakeUserPointer(h), MakeUserPointer(signals), 2,
                        MOJO_DEADLINE_INDEFINITE, MakeUserPointer(&result_index),
                        MakeUserPointer(hss)));
   EXPECT_EQ(0u, result_index);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
             hss[0].satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss[0].satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss[0].satisfiable_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss[1].satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss[1].satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss[1].satisfiable_signals);
 
   // Read from |h[0]|.
   // First, get only the size.
   buffer_size = 0;
   EXPECT_EQ(
       MOJO_RESULT_RESOURCE_EXHAUSTED,
       core()->ReadMessage(h[0], NullUserPointer(),
                           MakeUserPointer(&buffer_size), NullUserPointer(),
                           NullUserPointer(), MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(1u, buffer_size);
   // Then actually read it.
   buffer[0] = 'c';
   buffer_size = 1;
   EXPECT_EQ(
       MOJO_RESULT_OK,
       core()->ReadMessage(h[0], UserPointer<void>(buffer),
                           MakeUserPointer(&buffer_size), NullUserPointer(),
                           NullUserPointer(), MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ('b', buffer[0]);
   EXPECT_EQ(1u, buffer_size);
 
   // |h[0]| should no longer be readable.
   hss[0] = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED,
             core()->Wait(h[0], MOJO_HANDLE_SIGNAL_READABLE, 0,
                          MakeUserPointer(&hss[0])));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss[0].satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss[0].satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss[0].satisfiable_signals);
 
   // Write to |h[0]|.
   buffer[0] = 'd';
   EXPECT_EQ(
       MOJO_RESULT_OK,
       core()->WriteMessage(h[0], UserPointer<const void>(buffer), 1,
                            NullUserPointer(), 0, MOJO_WRITE_MESSAGE_FLAG_NONE));
 
   // Close |h[0]|.
   EXPECT_EQ(MOJO_RESULT_OK, core()->Close(h[0]));
@@ skipping 65 matching lines @@
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->WriteMessage(h_passing[0], UserPointer<const void>(kHello),
                                  kHelloSize, NullUserPointer(), 0,
                                  MOJO_WRITE_MESSAGE_FLAG_NONE));
   hss = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->Wait(h_passing[1], MOJO_HANDLE_SIGNAL_READABLE, 1000000000,
                          MakeUserPointer(&hss)));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
             hss.satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
   num_bytes = kBufferSize;
   num_handles = MOJO_ARRAYSIZE(handles);
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->ReadMessage(
                 h_passing[1], UserPointer<void>(buffer),
                 MakeUserPointer(&num_bytes), MakeUserPointer(handles),
                 MakeUserPointer(&num_handles), MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(kHelloSize, num_bytes);
   EXPECT_STREQ(kHello, buffer);
   EXPECT_EQ(0u, num_handles);
@@ skipping 19 matching lines @@
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->WriteMessage(h_passed[0], UserPointer<const void>(kHello),
                                  kHelloSize, NullUserPointer(), 0,
                                  MOJO_WRITE_MESSAGE_FLAG_NONE));
   hss = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->Wait(h_passed[1], MOJO_HANDLE_SIGNAL_READABLE, 1000000000,
                          MakeUserPointer(&hss)));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
             hss.satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
   num_bytes = kBufferSize;
   num_handles = MOJO_ARRAYSIZE(handles);
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->ReadMessage(
                 h_passed[1], UserPointer<void>(buffer),
                 MakeUserPointer(&num_bytes), MakeUserPointer(handles),
                 MakeUserPointer(&num_handles), MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(kHelloSize, num_bytes);
   EXPECT_STREQ(kHello, buffer);
   EXPECT_EQ(0u, num_handles);
 
   // Send |h_passed[1]| from |h_passing[0]| to |h_passing[1]|.
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->WriteMessage(h_passing[0], UserPointer<const void>(kWorld),
                                  kWorldSize, MakeUserPointer(&h_passed[1]), 1,
                                  MOJO_WRITE_MESSAGE_FLAG_NONE));
   hss = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->Wait(h_passing[1], MOJO_HANDLE_SIGNAL_READABLE, 1000000000,
                          MakeUserPointer(&hss)));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
             hss.satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
   num_bytes = kBufferSize;
   num_handles = MOJO_ARRAYSIZE(handles);
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->ReadMessage(
                 h_passing[1], UserPointer<void>(buffer),
                 MakeUserPointer(&num_bytes), MakeUserPointer(handles),
                 MakeUserPointer(&num_handles), MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(kWorldSize, num_bytes);
   EXPECT_STREQ(kWorld, buffer);
   EXPECT_EQ(1u, num_handles);
@@ skipping 14 matching lines @@
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->WriteMessage(h_passed[0], UserPointer<const void>(kHello),
                                  kHelloSize, NullUserPointer(), 0,
                                  MOJO_WRITE_MESSAGE_FLAG_NONE));
   hss = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->Wait(h_received, MOJO_HANDLE_SIGNAL_READABLE, 1000000000,
                          MakeUserPointer(&hss)));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
             hss.satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
   num_bytes = kBufferSize;
   num_handles = MOJO_ARRAYSIZE(handles);
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->ReadMessage(
                 h_received, UserPointer<void>(buffer),
                 MakeUserPointer(&num_bytes), MakeUserPointer(handles),
                 MakeUserPointer(&num_handles), MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(kHelloSize, num_bytes);
   EXPECT_STREQ(kHello, buffer);
   EXPECT_EQ(0u, num_handles);
@@ skipping 30 matching lines @@
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
             hss.satisfiable_signals);
 
   // Consumer should be never-writable, and not yet readable.
   hss = kFullMojoHandleSignalsState;
   EXPECT_EQ(
       MOJO_RESULT_FAILED_PRECONDITION,
       core()->Wait(ch, MOJO_HANDLE_SIGNAL_WRITABLE, 0, MakeUserPointer(&hss)));
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
             hss.satisfiable_signals);
   hss = kFullMojoHandleSignalsState;
   EXPECT_EQ(
       MOJO_RESULT_DEADLINE_EXCEEDED,
       core()->Wait(ch, MOJO_HANDLE_SIGNAL_READABLE, 0, MakeUserPointer(&hss)));
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
             hss.satisfiable_signals);
 
   // Write.
   signed char elements[2] = {'A', 'B'};
   uint32_t num_bytes = 2u;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->WriteData(ph, UserPointer<const void>(elements),
                               MakeUserPointer(&num_bytes),
                               MOJO_WRITE_DATA_FLAG_NONE));
   EXPECT_EQ(2u, num_bytes);
 
   // Consumer should now be readable.
   hss = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK, core()->Wait(ch, MOJO_HANDLE_SIGNAL_READABLE, 0,
                                          MakeUserPointer(&hss)));
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
             hss.satisfiable_signals);
 
   // Peek one character.
   elements[0] = -1;
   elements[1] = -1;
   num_bytes = 1u;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->ReadData(
                 ch, UserPointer<void>(elements), MakeUserPointer(&num_bytes),
                 MOJO_READ_DATA_FLAG_NONE | MOJO_READ_DATA_FLAG_PEEK));
@@ skipping 97 matching lines @@
   EXPECT_EQ('D', static_cast<const char*>(read_ptr)[0]);
   EXPECT_EQ('E', static_cast<const char*>(read_ptr)[1]);
   EXPECT_EQ(MOJO_RESULT_OK, core()->EndReadData(ch, 2u));
 
   // Consumer should now be no longer readable.
   hss = kFullMojoHandleSignalsState;
   EXPECT_EQ(
       MOJO_RESULT_DEADLINE_EXCEEDED,
       core()->Wait(ch, MOJO_HANDLE_SIGNAL_READABLE, 0, MakeUserPointer(&hss)));
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
             hss.satisfiable_signals);
 
   // TODO(vtl): More.
 
   // Close the producer.
   EXPECT_EQ(MOJO_RESULT_OK, core()->Close(ph));
 
   // The consumer should now be never-readable.
   hss = kFullMojoHandleSignalsState;
   EXPECT_EQ(
       MOJO_RESULT_FAILED_PRECONDITION,
       core()->Wait(ch, MOJO_HANDLE_SIGNAL_READABLE, 0, MakeUserPointer(&hss)));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
   EXPECT_EQ(MOJO_RESULT_OK, core()->Close(ch));
 }
 
+TEST_F(CoreTest, DataPipeSetGetConsumerOptions) {
+  MojoCreateDataPipeOptions options = {
+      sizeof(MojoCreateDataPipeOptions),        // |struct_size|.
+      MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
+      8,                                        // |element_num_bytes|.
+      800                                       // |capacity_num_bytes|.
+  };
+  MojoHandle ph, ch;  // p is for producer and c is for consumer.
+
+  EXPECT_EQ(MOJO_RESULT_OK,
+            core()->CreateDataPipe(MakeUserPointer(&options),
+                                   MakeUserPointer(&ph), MakeUserPointer(&ch)));
+  // Should get two distinct, valid handles.
+  EXPECT_NE(ph, MOJO_HANDLE_INVALID);
+  EXPECT_NE(ch, MOJO_HANDLE_INVALID);
+  EXPECT_NE(ph, ch);
+
+  // Read it.
+  MojoDataPipeConsumerOptions copts = {};
+  const uint32_t kCoptsSize = static_cast<uint32_t>(sizeof(copts));
+  EXPECT_EQ(MOJO_RESULT_OK, core()->GetDataPipeConsumerOptions(
+                                ch, MakeUserPointer(&copts), kCoptsSize));
+  EXPECT_EQ(kCoptsSize, copts.struct_size);
+  EXPECT_EQ(0u, copts.read_threshold_num_bytes);
+
+  // Invalid read threshold.
+  copts.struct_size = kCoptsSize;
+  copts.read_threshold_num_bytes = 4;
+  EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
+            core()->SetDataPipeConsumerOptions(ch, MakeUserPointer(&copts)));
+  // The options shouldn't change.
+  copts = MojoDataPipeConsumerOptions();
+  EXPECT_EQ(MOJO_RESULT_OK, core()->GetDataPipeConsumerOptions(
+                                ch, MakeUserPointer(&copts), kCoptsSize));
+  EXPECT_EQ(kCoptsSize, copts.struct_size);
+  EXPECT_EQ(0u, copts.read_threshold_num_bytes);
+
+  // Valid read threshold.
+  copts.struct_size = kCoptsSize;
+  copts.read_threshold_num_bytes = 8;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            core()->SetDataPipeConsumerOptions(ch, MakeUserPointer(&copts)));
+  copts = MojoDataPipeConsumerOptions();
+  EXPECT_EQ(MOJO_RESULT_OK, core()->GetDataPipeConsumerOptions(
+                                ch, MakeUserPointer(&copts), kCoptsSize));
+  EXPECT_EQ(kCoptsSize, copts.struct_size);
+  EXPECT_EQ(8u, copts.read_threshold_num_bytes);
+
+  // Invalid read threshold.
+  copts.struct_size = kCoptsSize;
+  copts.read_threshold_num_bytes = 9;
+  EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
+            core()->SetDataPipeConsumerOptions(ch, MakeUserPointer(&copts)));
+  // The options shouldn't change.
+  copts = MojoDataPipeConsumerOptions();
+  EXPECT_EQ(MOJO_RESULT_OK, core()->GetDataPipeConsumerOptions(
+                                ch, MakeUserPointer(&copts), kCoptsSize));
+  EXPECT_EQ(kCoptsSize, copts.struct_size);
+  EXPECT_EQ(8u, copts.read_threshold_num_bytes);
+
+  // Valid read threshold.
+  copts.struct_size = kCoptsSize;
+  copts.read_threshold_num_bytes = 16;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            core()->SetDataPipeConsumerOptions(ch, MakeUserPointer(&copts)));
+  copts = MojoDataPipeConsumerOptions();
+  EXPECT_EQ(MOJO_RESULT_OK, core()->GetDataPipeConsumerOptions(
+                                ch, MakeUserPointer(&copts), kCoptsSize));
+  EXPECT_EQ(kCoptsSize, copts.struct_size);
+  EXPECT_EQ(16u, copts.read_threshold_num_bytes);
+
+  // Default read threshold.
+  copts.struct_size = kCoptsSize;
+  copts.read_threshold_num_bytes = 0;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            core()->SetDataPipeConsumerOptions(ch, MakeUserPointer(&copts)));
+  copts = MojoDataPipeConsumerOptions();
+  EXPECT_EQ(MOJO_RESULT_OK, core()->GetDataPipeConsumerOptions(
+                                ch, MakeUserPointer(&copts), kCoptsSize));
+  EXPECT_EQ(kCoptsSize, copts.struct_size);
+  // Note: Should be reported as 0 ("default"), even if it means the element
+  // struct_size.
+  EXPECT_EQ(0u, copts.read_threshold_num_bytes);
+
+  EXPECT_EQ(MOJO_RESULT_OK, core()->Close(ph));
+  EXPECT_EQ(MOJO_RESULT_OK, core()->Close(ch));
+}
+
 // Tests passing data pipe producer and consumer handles.
 TEST_F(CoreTest, MessagePipeBasicLocalHandlePassing2) {
   const char kHello[] = "hello";
   const uint32_t kHelloSize = static_cast<uint32_t>(sizeof(kHello));
   const char kWorld[] = "world!!!";
   const uint32_t kWorldSize = static_cast<uint32_t>(sizeof(kWorld));
   char buffer[100];
   const uint32_t kBufferSize = static_cast<uint32_t>(sizeof(buffer));
   uint32_t num_bytes;
   MojoHandle handles[10];
@@ skipping 15 matching lines @@
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->WriteMessage(h_passing[0], UserPointer<const void>(kHello),
                                  kHelloSize, MakeUserPointer(&ch), 1,
                                  MOJO_WRITE_MESSAGE_FLAG_NONE));
   hss = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->Wait(h_passing[1], MOJO_HANDLE_SIGNAL_READABLE, 1000000000,
                          MakeUserPointer(&hss)));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
             hss.satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
   num_bytes = kBufferSize;
   num_handles = MOJO_ARRAYSIZE(handles);
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->ReadMessage(
                 h_passing[1], UserPointer<void>(buffer),
                 MakeUserPointer(&num_bytes), MakeUserPointer(handles),
                 MakeUserPointer(&num_handles), MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(kHelloSize, num_bytes);
   EXPECT_STREQ(kHello, buffer);
   EXPECT_EQ(1u, num_handles);
@@ skipping 13 matching lines @@
   // Write to |ph|. Should receive on |ch_received|.
   num_bytes = kWorldSize;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->WriteData(ph, UserPointer<const void>(kWorld),
                               MakeUserPointer(&num_bytes),
                               MOJO_WRITE_DATA_FLAG_ALL_OR_NONE));
   hss = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->Wait(ch_received, MOJO_HANDLE_SIGNAL_READABLE, 1000000000,
                          MakeUserPointer(&hss)));
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
             hss.satisfiable_signals);
   num_bytes = kBufferSize;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->ReadData(ch_received, UserPointer<void>(buffer),
                              MakeUserPointer(&num_bytes),
                              MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(kWorldSize, num_bytes);
   EXPECT_STREQ(kWorld, buffer);
 
   // Now pass |ph| in the same direction.
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->WriteMessage(h_passing[0], UserPointer<const void>(kWorld),
                                  kWorldSize, MakeUserPointer(&ph), 1,
                                  MOJO_WRITE_MESSAGE_FLAG_NONE));
   hss = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->Wait(h_passing[1], MOJO_HANDLE_SIGNAL_READABLE, 1000000000,
                          MakeUserPointer(&hss)));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
             hss.satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
   num_bytes = kBufferSize;
   num_handles = MOJO_ARRAYSIZE(handles);
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->ReadMessage(
                 h_passing[1], UserPointer<void>(buffer),
                 MakeUserPointer(&num_bytes), MakeUserPointer(handles),
                 MakeUserPointer(&num_handles), MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(kWorldSize, num_bytes);
   EXPECT_STREQ(kWorld, buffer);
   EXPECT_EQ(1u, num_handles);
@@ skipping 13 matching lines @@
   // Write to |ph_received|. Should receive on |ch_received|.
   num_bytes = kHelloSize;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->WriteData(ph_received, UserPointer<const void>(kHello),
                               MakeUserPointer(&num_bytes),
                               MOJO_WRITE_DATA_FLAG_ALL_OR_NONE));
   hss = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->Wait(ch_received, MOJO_HANDLE_SIGNAL_READABLE, 1000000000,
                          MakeUserPointer(&hss)));
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
             hss.satisfiable_signals);
   num_bytes = kBufferSize;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->ReadData(ch_received, UserPointer<void>(buffer),
                              MakeUserPointer(&num_bytes),
                              MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(kHelloSize, num_bytes);
   EXPECT_STREQ(kHello, buffer);
 
   ph = ph_received;
@@ skipping 37 matching lines @@
   EXPECT_NE(ch, MOJO_HANDLE_INVALID);
 
   // Complete the two-phase write.
   static_cast<char*>(write_ptr)[0] = 'x';
   EXPECT_EQ(MOJO_RESULT_OK, core()->EndWriteData(ph, 1));
 
   // Wait for |ch| to be readable.
   hss = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK, core()->Wait(ch, MOJO_HANDLE_SIGNAL_READABLE,
                                          1000000000, MakeUserPointer(&hss)));
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
             hss.satisfiable_signals);
 
   // Make sure that |ch| can't be sent if it's in a two-phase read.
   const void* read_ptr = nullptr;
   num_bytes = 1;
   ASSERT_EQ(MOJO_RESULT_OK,
             core()->BeginReadData(ch, MakeUserPointer(&read_ptr),
                                   MakeUserPointer(&num_bytes),
                                   MOJO_READ_DATA_FLAG_NONE));
   EXPECT_EQ(MOJO_RESULT_BUSY,
             core()->WriteMessage(h_passing[0], UserPointer<const void>(kHello),
                                  kHelloSize, MakeUserPointer(&ch), 1,
                                  MOJO_WRITE_MESSAGE_FLAG_NONE));
 
   // But |ph| can, even if |ch| is in a two-phase read.
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->WriteMessage(h_passing[0], UserPointer<const void>(kWorld),
                                  kWorldSize, MakeUserPointer(&ph), 1,
                                  MOJO_WRITE_MESSAGE_FLAG_NONE));
   ph = MOJO_HANDLE_INVALID;
   hss = kEmptyMojoHandleSignalsState;
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->Wait(h_passing[1], MOJO_HANDLE_SIGNAL_READABLE, 1000000000,
                          MakeUserPointer(&hss)));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
             hss.satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
   num_bytes = kBufferSize;
   num_handles = MOJO_ARRAYSIZE(handles);
   EXPECT_EQ(MOJO_RESULT_OK,
             core()->ReadMessage(
                 h_passing[1], UserPointer<void>(buffer),
                 MakeUserPointer(&num_bytes), MakeUserPointer(handles),
                 MakeUserPointer(&num_handles), MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(kWorldSize, num_bytes);
   EXPECT_STREQ(kWorld, buffer);
   EXPECT_EQ(1u, num_handles);
@@ skipping 40 matching lines @@
   EXPECT_EQ(MOJO_RESULT_BUSY, waiter.result);
 
   EXPECT_EQ(MOJO_RESULT_OK, core()->Close(h));
 }
 
 // TODO(vtl): Test |DuplicateBufferHandle()| and |MapBuffer()|.
 
 }  // namespace
 }  // namespace system
 }  // namespace mojo