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

mojo/edk/system/data_pipe_impl_unittest.cc

 // Copyright 2015 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.
 
 // This file contains tests that are shared between different implementations of
 // |DataPipeImpl|.
 
 #include "mojo/edk/system/data_pipe_impl.h"
 
 #include <stdint.h>
@@ skipping 22 matching lines @@
 
 using mojo::platform::PlatformPipe;
 using mojo::platform::ThreadSleep;
 using mojo::util::MakeRefCounted;
 using mojo::util::RefPtr;
 
 namespace mojo {
 namespace system {
 namespace {
 
-const MojoHandleSignals kAllSignals = MOJO_HANDLE_SIGNAL_READABLE |
-                                      MOJO_HANDLE_SIGNAL_WRITABLE |
-                                      MOJO_HANDLE_SIGNAL_PEER_CLOSED;
 const uint32_t kSizeOfOptions =
     static_cast<uint32_t>(sizeof(MojoCreateDataPipeOptions));
 
 // In various places, we have to poll (since, e.g., we can't yet wait for a
 // certain amount of data to be available). This is the maximum number of
 // iterations (separated by a short sleep).
 // TODO(vtl): Get rid of this.
 const size_t kMaxPoll = 100;
 
 // DataPipeImplTestHelper ------------------------------------------------------
@@ skipping 85 matching lines @@
                                  HandleSignalsState* signals_state) {
     return dpp()->ProducerAddAwakable(awakable, signals, context,
                                       signals_state);
   }
   void ProducerRemoveAwakable(Awakable* awakable,
                               HandleSignalsState* signals_state) {
     return dpp()->ProducerRemoveAwakable(awakable, signals_state);
   }
 
   void ConsumerClose() { helper_->ConsumerClose(); }
+  MojoResult ConsumerSetOptions(uint32_t read_threshold_num_bytes) {
+    return dpc()->ConsumerSetOptions(read_threshold_num_bytes);
+  }
+  void ConsumerGetOptions(uint32_t* read_threshold_num_bytes) {
+    dpc()->ConsumerGetOptions(read_threshold_num_bytes);
+  }
   MojoResult ConsumerReadData(UserPointer<void> elements,
                               UserPointer<uint32_t> num_bytes,
                               bool all_or_none,
                               bool peek) {
     return dpc()->ConsumerReadData(elements, num_bytes, all_or_none, peek);
   }
   MojoResult ConsumerDiscardData(UserPointer<uint32_t> num_bytes,
                                  bool all_or_none) {
     return dpc()->ConsumerDiscardData(num_bytes, all_or_none);
   }
@@ skipping 120 matching lines @@
                                     MOJO_WRITE_MESSAGE_FLAG_NONE));
       transport.End();
     }
     uint32_t context = 0;
     ASSERT_EQ(MOJO_RESULT_OK, waiter.Wait(test::ActionTimeout(), &context));
     EXPECT_EQ(987u, context);
     HandleSignalsState hss = HandleSignalsState();
     message_pipe(dest_i)->RemoveAwakable(0, &waiter, &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);
     char read_buffer[100] = {};
     uint32_t read_buffer_size = static_cast<uint32_t>(sizeof(read_buffer));
     DispatcherVector read_dispatchers;
     uint32_t read_num_dispatchers = 10;  // Maximum to get.
     ASSERT_EQ(MOJO_RESULT_OK,
               message_pipe(dest_i)->ReadMessage(
                   0, UserPointer<void>(read_buffer),
                   MakeUserPointer(&read_buffer_size), &read_dispatchers,
                   &read_num_dispatchers, MOJO_READ_MESSAGE_FLAG_NONE));
     EXPECT_EQ(0u, static_cast<size_t>(read_buffer_size));
@@ skipping 352 matching lines @@
                                     MakeUserPointer(&num_bytes), false));
   // It should have written everything (even without "all or none").
   EXPECT_EQ(2u * sizeof(elements[0]), num_bytes);
 
   // Wait.
   context = 0;
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::ActionTimeout(), &context));
   EXPECT_EQ(123u, context);
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &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);
 
   // Query.
   // TODO(vtl): It's theoretically possible (though not with the current
   // implementation/configured limits) that not all the data has arrived yet.
   // (The theoretically-correct assertion here is that |num_bytes| is |1 * ...|
   // or |2 * ...|.)
   num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerQueryData(MakeUserPointer(&num_bytes)));
@@ skipping 131 matching lines @@
   EXPECT_EQ(0u, hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
             hss.satisfiable_signals);
 
   // Wait for data to become available to the consumer.
   context = 0;
   EXPECT_EQ(MOJO_RESULT_OK, cwaiter.Wait(test::TinyTimeout(), &context));
   EXPECT_EQ(1234u, context);
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&cwaiter, &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 element.
   elements[0] = -1;
   elements[1] = -1;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerReadData(UserPointer<void>(elements),
                                    MakeUserPointer(&num_bytes), true, true));
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
@@ skipping 199 matching lines @@
   HandleSignalsState hss;
   uint32_t context;
 
   // Never writable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
             this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 12,
                                       &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);
 
   // Add waiter: not yet readable.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
             this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 34,
                                       nullptr));
 
   // Write two elements.
   int32_t elements[2] = {123, 456};
   uint32_t num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerWriteData(UserPointer<const void>(elements),
                                     MakeUserPointer(&num_bytes), true));
 
   // Wait for readability (needed for remote cases).
   context = 0;
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), &context));
   EXPECT_EQ(34u, context);
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &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);
 
   // Discard one element.
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerDiscardData(MakeUserPointer(&num_bytes), true));
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
 
   // Should still be readable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
             this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 78,
                                       &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 element.
   elements[0] = -1;
   elements[1] = -1;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerReadData(UserPointer<void>(elements),
                                    MakeUserPointer(&num_bytes), true, true));
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
   EXPECT_EQ(456, elements[0]);
   EXPECT_EQ(-1, elements[1]);
 
   // Should still be readable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
             this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 78,
                                       &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);
 
   // Read one element.
   elements[0] = -1;
   elements[1] = -1;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerReadData(UserPointer<void>(elements),
                                    MakeUserPointer(&num_bytes), true, false));
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
@@ skipping 13 matching lines @@
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerWriteData(UserPointer<const void>(elements),
                                     MakeUserPointer(&num_bytes), true));
 
   // Waiting should now succeed.
   context = 0;
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), &context));
   EXPECT_EQ(90u, context);
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &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);
 
   // We'll want to wait for the peer closed signal to propagate.
   waiter.Init();
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
                                       12, nullptr));
 
   // Close the producer.
   this->ProducerClose();
 
   // Should still be readable, even if the peer closed signal hasn't propagated
   // yet.
   waiter2.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
             this->ConsumerAddAwakable(&waiter2, MOJO_HANDLE_SIGNAL_READABLE, 34,
                                       &hss));
   // We don't know if the peer closed signal has propagated yet (for the remote
   // cases).
   EXPECT_TRUE((hss.satisfied_signals & MOJO_HANDLE_SIGNAL_READABLE));
-  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);
 
   // Wait for the peer closed signal.
   context = 0;
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), &context));
   EXPECT_EQ(12u, context);
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &hss);
-  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.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);
 
   // Read one element.
   elements[0] = -1;
   elements[1] = -1;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerReadData(UserPointer<void>(elements),
                                    MakeUserPointer(&num_bytes), true, false));
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
@@ skipping 48 matching lines @@
   elements[1] = 456;
   EXPECT_EQ(MOJO_RESULT_OK, this->ProducerEndWriteData(static_cast<uint32_t>(
                                 2u * sizeof(elements[0]))));
 
   // Wait for readability (needed for remote cases).
   context = 0;
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), &context));
   EXPECT_EQ(12u, context);
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &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);
 
   // Read one element.
   const void* read_buffer = nullptr;
   num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerBeginReadData(MakeUserPointer(&read_buffer),
                                         MakeUserPointer(&num_bytes)));
   EXPECT_TRUE(read_buffer);
   EXPECT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes);
   const int32_t* read_elements = static_cast<const int32_t*>(read_buffer);
   EXPECT_EQ(123, read_elements[0]);
   EXPECT_EQ(MOJO_RESULT_OK, this->ConsumerEndReadData(static_cast<uint32_t>(
                                 1u * sizeof(elements[0]))));
 
   // Should still be readable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
             this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 34,
                                       &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);
 
   // Read one element.
   // Request three, but not in all-or-none mode.
   read_buffer = nullptr;
   num_bytes = static_cast<uint32_t>(3u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerBeginReadData(MakeUserPointer(&read_buffer),
                                         MakeUserPointer(&num_bytes)));
   EXPECT_TRUE(read_buffer);
@@ skipping 87 matching lines @@
             this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 3,
                                       &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
             hss.satisfiable_signals);
 
   // It should become readable.
   EXPECT_EQ(MOJO_RESULT_OK, cwaiter.Wait(test::TinyTimeout(), nullptr));
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&cwaiter, &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);
 
   // Start another two-phase write and check that it's readable even in the
   // middle of it.
   write_ptr = nullptr;
   num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerBeginWriteData(MakeUserPointer(&write_ptr),
                                          MakeUserPointer(&num_bytes)));
   EXPECT_TRUE(write_ptr);
   EXPECT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(int32_t)));
 
   // It should be readable.
   cwaiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
             this->ConsumerAddAwakable(&cwaiter, MOJO_HANDLE_SIGNAL_READABLE, 5,
                                       &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);
 
   // End the two-phase write without writing anything.
   EXPECT_EQ(MOJO_RESULT_OK, this->ProducerEndWriteData(0u));
 
   // Start a two-phase read.
   const void* read_ptr = nullptr;
   num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerBeginReadData(MakeUserPointer(&read_ptr),
@@ skipping 13 matching lines @@
 
   // But not readable.
   cwaiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
             this->ConsumerAddAwakable(&cwaiter, MOJO_HANDLE_SIGNAL_READABLE, 7,
                                       nullptr));
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, cwaiter.Wait(0, nullptr));
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&cwaiter, &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);
 
   // End the two-phase read without reading anything.
   EXPECT_EQ(MOJO_RESULT_OK, this->ConsumerEndReadData(0u));
 
   // It should be readable again.
   cwaiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
             this->ConsumerAddAwakable(&cwaiter, MOJO_HANDLE_SIGNAL_READABLE, 8,
                                       &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);
 
   this->ProducerClose();
   this->ConsumerClose();
 }
 
 void Seq(int32_t start, size_t count, int32_t* out) {
   for (size_t i = 0; i < count; i++)
     out[i] = start + static_cast<int32_t>(i);
 }
@@ skipping 40 matching lines @@
   EXPECT_EQ(5u * sizeof(int32_t), num_bytes);
 
   // Wait for data.
   // TODO(vtl): There's no real guarantee that all the data will become
   // available at once (except that in current implementations, with reasonable
   // limits, it will). Eventually, we'll be able to wait for a specified amount
   // of data to become available.
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr));
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &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);
 
   // Half full.
   num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerQueryData(MakeUserPointer(&num_bytes)));
   EXPECT_EQ(5u * sizeof(int32_t), num_bytes);
 
   // Too much.
   num_bytes = 6u * sizeof(int32_t);
@@ skipping 89 matching lines @@
             this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
                                       2, nullptr));
 
   // Close the producer, then test producer-closed cases.
   this->ProducerClose();
 
   // Wait.
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr));
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &hss);
-  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.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);
 
   // Try reading too much; "failed precondition" since the producer is closed.
   num_bytes = 4u * sizeof(int32_t);
   memset(buffer, 0xab, sizeof(buffer));
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
             this->ConsumerReadData(UserPointer<void>(buffer),
                                    MakeUserPointer(&num_bytes), true, false));
   memset(expected_buffer, 0xab, sizeof(expected_buffer));
   EXPECT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
@@ skipping 67 matching lines @@
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerWriteData(UserPointer<const void>(&test_data[0]),
                                     MakeUserPointer(&num_bytes), false));
   EXPECT_EQ(20u, num_bytes);
 
   // Wait for data.
   // TODO(vtl): (See corresponding TODO in AllOrNone.)
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr));
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &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);
 
   // Read 10 bytes.
   unsigned char read_buffer[1000] = {0};
   num_bytes = 10u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerReadData(UserPointer<void>(read_buffer),
                                    MakeUserPointer(&num_bytes), false, false));
   EXPECT_EQ(10u, num_bytes);
   EXPECT_EQ(0, memcmp(read_buffer, &test_data[0], 10u));
@@ skipping 190 matching lines @@
             this->ProducerBeginWriteData(MakeUserPointer(&write_buffer_ptr),
                                          MakeUserPointer(&num_bytes)));
   EXPECT_TRUE(write_buffer_ptr);
   ASSERT_GT(num_bytes, kTestDataSize);
 
   // Wait for data.
   // TODO(vtl): (See corresponding TODO in AllOrNone.)
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr));
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &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);
 
   // Start two-phase read.
   const void* read_buffer_ptr = nullptr;
   num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerBeginReadData(MakeUserPointer(&read_buffer_ptr),
                                         MakeUserPointer(&num_bytes)));
   EXPECT_TRUE(read_buffer_ptr);
   EXPECT_EQ(kTestDataSize, num_bytes);
@@ skipping 96 matching lines @@
                                       1, nullptr));
 
   // Close the producer.
   this->ProducerClose();
 
   // Wait. (Note that once the consumer knows that the producer is closed, it
   // must also know about all the data that was sent.)
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr));
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &hss);
-  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.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);
 
   // Peek that data.
   char buffer[1000];
   num_bytes = static_cast<uint32_t>(sizeof(buffer));
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerReadData(UserPointer<void>(buffer),
                                    MakeUserPointer(&num_bytes), false, true));
   EXPECT_EQ(kTestDataSize, num_bytes);
   EXPECT_EQ(0, memcmp(buffer, kTestData, kTestDataSize));
@@ skipping 56 matching lines @@
   uint32_t num_bytes = kTestDataSize;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerWriteData(UserPointer<const void>(&kTestData),
                                     MakeUserPointer(&num_bytes), false));
   EXPECT_EQ(kTestDataSize, num_bytes);
 
   // Wait.
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr));
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &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);
 
   // Read that data.
   int64_t data[10] = {};
   num_bytes = static_cast<uint32_t>(sizeof(data));
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerReadData(UserPointer<void>(data),
                                    MakeUserPointer(&num_bytes), false, false));
   EXPECT_EQ(kTestDataSize, num_bytes);
   EXPECT_EQ(kTestData, data[0]);
@@ skipping 47 matching lines @@
   uint32_t num_bytes = kTestDataSize;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerWriteData(UserPointer<const void>(&kTestData),
                                     MakeUserPointer(&num_bytes), false));
   EXPECT_EQ(kTestDataSize, num_bytes);
 
   // Wait.
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr));
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &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);
 
   // Start a two-phase read.
   num_bytes = 0u;
   const void* read_ptr = nullptr;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerBeginReadData(MakeUserPointer(&read_ptr),
                                         MakeUserPointer(&num_bytes)));
   EXPECT_EQ(kTestDataSize, num_bytes);
   EXPECT_EQ(kTestData, static_cast<const int64_t*>(read_ptr)[0]);
 
   // Add waiter (for the producer to be closed).
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
             this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
                                       0, nullptr));
 
   // Close the producer.
   this->ProducerClose();
 
   // Wait for producer close to be detected.
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr));
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, 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);
 
   // Add waiter (for the consumer to become readable).
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
             this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 0,
                                       nullptr));
 
   // Complete the two-phase read.
   EXPECT_EQ(MOJO_RESULT_OK, this->ConsumerEndReadData(kTestDataSize));
@@ skipping 162 matching lines @@
   num_bytes = 1u * sizeof(int32_t);
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerWriteData(UserPointer<const void>(&element),
                                     MakeUserPointer(&num_bytes), false));
 
   // Wait for data.
   // TODO(vtl): (See corresponding TODO in AllOrNone.)
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr));
   hss = HandleSignalsState();
   this->ConsumerRemoveAwakable(&waiter, &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);
 
   // One element available.
   num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerQueryData(MakeUserPointer(&num_bytes)));
   EXPECT_EQ(1u * sizeof(int32_t), num_bytes);
 
   // Try "ending" a two-phase read when one isn't active.
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
@@ skipping 96 matching lines @@
   // |ConsumerEndReadData()| below) we want |producer_open()| to be false but
   // the call to |channel_endpoint_->EnqueueMessage()| to fail. (This race can
   // occur without the sleep, but is much less likely.)
   ThreadSleep(10u);
 
   EXPECT_EQ(MOJO_RESULT_OK, this->ConsumerEndReadData(num_bytes));
 
   this->ConsumerClose();
 }
 
+TYPED_TEST(DataPipeImplTest, ReadThreshold) {
+  const MojoCreateDataPipeOptions options = {
+      kSizeOfOptions,                           // |struct_size|.
+      MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
+      1u,                                       // |element_num_bytes|.
+      1000u                                     // |capacity_num_bytes|.
+  };
+  this->Create(options);
+  this->DoTransfer();
+
+  // The default read threshold should be 0 (which means "default", i.e., one
+  // element).
+  uint32_t read_threshold_num_bytes = 123u;
+  this->ConsumerGetOptions(&read_threshold_num_bytes);
+  EXPECT_EQ(0u, read_threshold_num_bytes);
+
+  Waiter waiter;
+  HandleSignalsState hss;
+
+  // Add a waiter.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            this->ConsumerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, 0, nullptr));
+  // Trivial wait: it shouldn't have the read threshold signal.
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
+  hss = HandleSignalsState();
+  this->ConsumerRemoveAwakable(&waiter, &hss);
+  EXPECT_EQ(0u, hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
+            hss.satisfiable_signals);
+
+  // Add a waiter.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            this->ConsumerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, 0, nullptr));
+
+  // Write a byte.
+  const char kTestData[] = {'x'};
+  const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData));
+  uint32_t num_bytes = kTestDataSize;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            this->ProducerWriteData(UserPointer<const void>(kTestData),
+                                    MakeUserPointer(&num_bytes), false));
+  EXPECT_EQ(kTestDataSize, num_bytes);
+
+  // Wait for the read threshold signal.
+  EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr));
+  hss = HandleSignalsState();
+  this->ConsumerRemoveAwakable(&waiter, &hss);
+  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);
+
+  // Set the read threshold to 1.
+  this->ConsumerSetOptions(1);
+  read_threshold_num_bytes = 123u;
+  this->ConsumerGetOptions(&read_threshold_num_bytes);
+  EXPECT_EQ(1u, read_threshold_num_bytes);
+
+  // Try to add a waiter: it should (still) already have the read threshold
+  // signal.
+  waiter.Init();
+  hss = HandleSignalsState();
+  ASSERT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            this->ConsumerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, 0, &hss));
+  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);
+
+  // Set the read threshold to 2.
+  this->ConsumerSetOptions(2);
+  read_threshold_num_bytes = 123u;
+  this->ConsumerGetOptions(&read_threshold_num_bytes);
+  EXPECT_EQ(2u, read_threshold_num_bytes);
+
+  // Add a waiter.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            this->ConsumerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, 0, nullptr));
+
+  // Write another byte.
+  num_bytes = kTestDataSize;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            this->ProducerWriteData(UserPointer<const void>(kTestData),
+                                    MakeUserPointer(&num_bytes), false));
+  EXPECT_EQ(kTestDataSize, num_bytes);
+
+  // Wait for the read threshold signal.
+  EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr));
+  hss = HandleSignalsState();
+  this->ConsumerRemoveAwakable(&waiter, &hss);
+  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);
+
+  // Read one byte.
+  char read_byte = 'a';
+  num_bytes = sizeof(read_byte);
+  EXPECT_EQ(MOJO_RESULT_OK,
+            this->ConsumerReadData(UserPointer<void>(&read_byte),
+                                   MakeUserPointer(&num_bytes), true, false));
+  EXPECT_EQ(1u, num_bytes);
+  EXPECT_EQ(kTestData[0], read_byte);
+
+  // Add a waiter.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            this->ConsumerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, 0, nullptr));
+  // Trivial wait: it shouldn't have the read threshold signal.
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
+  hss = HandleSignalsState();
+  this->ConsumerRemoveAwakable(&waiter, &hss);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
+            hss.satisfiable_signals);
+
+  // Add a waiter.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            this->ConsumerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, 0, nullptr));
+  // Trivial wait: it shouldn't have the read threshold signal.
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
+  hss = HandleSignalsState();
+  this->ConsumerRemoveAwakable(&waiter, &hss);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
+            hss.satisfiable_signals);
+
+  // Add a waiter.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            this->ConsumerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, 0, nullptr));
+
+  // Close the producer.
+  this->ProducerClose();
+
+  // Wait; the current read threshold becomes never satisfiable.
+  EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
+            waiter.Wait(test::TinyTimeout(), nullptr));
+  hss = HandleSignalsState();
+  this->ConsumerRemoveAwakable(&waiter, &hss);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
+
+  // Set the read threshold back to zero to 0.
+  this->ConsumerSetOptions(0);
+  read_threshold_num_bytes = 123u;
+  this->ConsumerGetOptions(&read_threshold_num_bytes);
+  // "Get options" should preserve 0 (and not set it to the element size).
+  EXPECT_EQ(0u, read_threshold_num_bytes);
+
+  // Try to add a waiter: it should have the read threshold signal.
+  waiter.Init();
+  hss = HandleSignalsState();
+  ASSERT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            this->ConsumerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, 0, &hss));
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                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);
+
+  // Read the other byte.
+  read_byte = 'a';
+  num_bytes = sizeof(read_byte);
+  EXPECT_EQ(MOJO_RESULT_OK,
+            this->ConsumerReadData(UserPointer<void>(&read_byte),
+                                   MakeUserPointer(&num_bytes), true, false));
+  EXPECT_EQ(1u, num_bytes);
+  EXPECT_EQ(kTestData[0], read_byte);
+
+  // Try to add a waiter: the read threshold signal should be unsatisfiable.
+  waiter.Init();
+  hss = HandleSignalsState();
+  ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
+            this->ConsumerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, 0, &hss));
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
+
+  this->ConsumerClose();
+}
+
 }  // namespace
 }  // namespace system
 }  // namespace mojo