EDK: Add implementation of data pipe producer write 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 102 matching lines @@
 
   void Reset() {
     if (helper_)
       helper_->TearDown();
 
     helper_.reset(new Helper());
     helper_->SetUp();
   }
 
   void ProducerClose() { helper_->ProducerClose(); }
+  MojoResult ProducerSetOptions(uint32_t write_threshold_num_bytes) {
+    return dpp()->ProducerSetOptions(write_threshold_num_bytes);
+  }
+  void ProducerGetOptions(uint32_t* write_threshold_num_bytes) {
+    dpp()->ProducerGetOptions(write_threshold_num_bytes);
+  }
   MojoResult ProducerWriteData(UserPointer<const void> elements,
                                UserPointer<uint32_t> num_bytes,
                                bool all_or_none) {
     return dpp()->ProducerWriteData(elements, num_bytes, all_or_none);
   }
   MojoResult ProducerBeginWriteData(UserPointer<void*> buffer,
                                     UserPointer<uint32_t> buffer_num_bytes) {
     return dpp()->ProducerBeginWriteData(buffer, buffer_num_bytes);
   }
   MojoResult ProducerEndWriteData(uint32_t num_bytes_written) {
@@ skipping 637 matching lines @@
   Waiter cwaiter;  // For consumer.
   HandleSignalsState hss;
   uint32_t context;
 
   // Never readable.
   pwaiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
             this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_READABLE, 12,
                                       &hss));
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
             hss.satisfiable_signals);
 
   // Already writable.
   pwaiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
             this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 34,
                                       &hss));
 
   // We'll need to wait for readability for the remote cases.
@@ skipping 13 matching lines @@
   // Adding a waiter should now succeed.
   pwaiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
             this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 56,
                                       nullptr));
   // And it shouldn't be writable yet.
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, pwaiter.Wait(0, nullptr));
   hss = HandleSignalsState();
   this->ProducerRemoveAwakable(&pwaiter, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
             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 | MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
             hss.satisfied_signals);
@@ skipping 15 matching lines @@
   // Add a waiter.
   pwaiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
             this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 56,
                                       nullptr));
   // And it still shouldn't be writable yet.
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, pwaiter.Wait(0, nullptr));
   hss = HandleSignalsState();
   this->ProducerRemoveAwakable(&pwaiter, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
             hss.satisfiable_signals);
 
   // Do it again.
   pwaiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
             this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 78,
                                       nullptr));
 
   // 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);
   EXPECT_EQ(123, elements[0]);
   EXPECT_EQ(-1, elements[1]);
 
   // Waiting should now succeed.
   context = 0;
   EXPECT_EQ(MOJO_RESULT_OK, pwaiter.Wait(test::TinyTimeout(), &context));
   EXPECT_EQ(78u, context);
   hss = HandleSignalsState();
   this->ProducerRemoveAwakable(&pwaiter, &hss);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
             hss.satisfiable_signals);
 
   // Try writing, using a two-phase write.
   void* buffer = nullptr;
   num_bytes = static_cast<uint32_t>(3u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerBeginWriteData(MakeUserPointer(&buffer),
                                          MakeUserPointer(&num_bytes)));
   EXPECT_TRUE(buffer);
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
@@ skipping 22 matching lines @@
   EXPECT_EQ(456, static_cast<const int32_t*>(read_buffer)[0]);
   EXPECT_EQ(MOJO_RESULT_OK, this->ConsumerEndReadData(static_cast<uint32_t>(
                                 1u * sizeof(elements[0]))));
 
   // Waiting should succeed.
   context = 0;
   EXPECT_EQ(MOJO_RESULT_OK, pwaiter.Wait(test::TinyTimeout(), &context));
   EXPECT_EQ(90u, context);
   hss = HandleSignalsState();
   this->ProducerRemoveAwakable(&pwaiter, &hss);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
             hss.satisfiable_signals);
 
   // Write one element.
   elements[0] = 123;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerWriteData(UserPointer<const void>(elements),
                                     MakeUserPointer(&num_bytes), false));
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
 
@@ skipping 401 matching lines @@
   Waiter pwaiter;  // For producer.
   Waiter cwaiter;  // For consumer.
   HandleSignalsState hss;
 
   // It should be writable.
   pwaiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
             this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 0,
                                       &hss));
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
             hss.satisfiable_signals);
 
   void* write_ptr = nullptr;
   uint32_t 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)));
 
   // At this point, it shouldn't be writable.
   pwaiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
             this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 1,
                                       nullptr));
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, pwaiter.Wait(0, nullptr));
   hss = HandleSignalsState();
   this->ProducerRemoveAwakable(&pwaiter, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
             hss.satisfiable_signals);
 
   // It shouldn't be readable yet either (we'll wait later).
   cwaiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
             this->ConsumerAddAwakable(&cwaiter, MOJO_HANDLE_SIGNAL_READABLE, 2,
                                       nullptr));
 
   static_cast<int32_t*>(write_ptr)[0] = 123;
   EXPECT_EQ(MOJO_RESULT_OK, this->ProducerEndWriteData(
                                 static_cast<uint32_t>(1u * sizeof(int32_t))));
 
   // It should immediately be writable again.
   pwaiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
             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,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
             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 | MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
             hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
                 MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
@@ skipping 32 matching lines @@
                                         MakeUserPointer(&num_bytes)));
   EXPECT_TRUE(read_ptr);
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(int32_t)), num_bytes);
 
   // At this point, it should still be writable.
   pwaiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
             this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 6,
                                       &hss));
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
             hss.satisfiable_signals);
 
   // 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);
@@ skipping 1092 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, WriteThreshold) {
+  const MojoCreateDataPipeOptions options = {
+      kSizeOfOptions,                           // |struct_size|.
+      MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
+      1u,                                       // |element_num_bytes|.
+      10u                                       // |capacity_num_bytes|.
+  };
+  this->Create(options);
+  this->DoTransfer();
+
+  // The default write threshold should be 0 (which means "default", i.e., one
+  // element).
+  uint32_t write_threshold_num_bytes = 123;
+  this->ProducerGetOptions(&write_threshold_num_bytes);
+  EXPECT_EQ(0u, write_threshold_num_bytes);
+
+  Waiter waiter;
+  HandleSignalsState hss;
+
+  // Try to wait to the write threshold signal; it should already have it.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            this->ProducerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, 0, &hss));
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
+            hss.satisfiable_signals);
+
+  // Before writing, add a waiter on the consumer (since we'll need to know when
+  // the written bytes have propagated).
+  Waiter read_waiter;
+  read_waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            this->ConsumerAddAwakable(&read_waiter, MOJO_HANDLE_SIGNAL_READABLE,
+                                      0, nullptr));
+
+  // Write 5 bytes.
+  static const char kTestData[5] = {'A', 'B', 'C', 'D', 'E'};
+  uint32_t num_bytes = 5;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            this->ProducerWriteData(UserPointer<const void>(kTestData),
+                                    MakeUserPointer(&num_bytes), false));
+  EXPECT_EQ(5u, num_bytes);
+
+  // It should still have the write threshold signal.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            this->ProducerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, 0, nullptr));
+
+  // Set the write threshold to 5.
+  this->ProducerSetOptions(5);
+  write_threshold_num_bytes = 123u;
+  this->ProducerGetOptions(&write_threshold_num_bytes);
+  EXPECT_EQ(5u, write_threshold_num_bytes);
+
+  // Should still have the write threshold signal.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            this->ProducerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, 0, nullptr));
+
+  // Set the write threshold to 6.
+  this->ProducerSetOptions(6);
+
+  // Should no longer have the write threshold signal.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            this->ProducerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, 0, nullptr));
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
+  hss = HandleSignalsState();
+  this->ProducerRemoveAwakable(&waiter, &hss);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
+            hss.satisfiable_signals);
+
+  // Add a waiter.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            this->ProducerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, 0, nullptr));
+
+  // Wait for the consumer to be readable.
+  EXPECT_EQ(MOJO_RESULT_OK, read_waiter.Wait(test::TinyTimeout(), nullptr));
+  this->ConsumerRemoveAwakable(&read_waiter, nullptr);
+
+  // Read a 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);
+
+  // Wait; should get the write threshold signal.
+  EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr));
+  hss = HandleSignalsState();
+  this->ProducerRemoveAwakable(&waiter, &hss);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
+            hss.satisfiable_signals);
+
+  // Write 6 more bytes.
+  static const char kMoreTestData[6] = {'1', '2', '3', '4', '5', '6'};
+  num_bytes = 6;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            this->ProducerWriteData(UserPointer<const void>(kMoreTestData),
+                                    MakeUserPointer(&num_bytes), false));
+  EXPECT_EQ(6u, num_bytes);
+
+  // Should no longer have the write threshold signal.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            this->ProducerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, 0, nullptr));
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
+  this->ProducerRemoveAwakable(&waiter, nullptr);
+
+  // Set the write threshold to 0 (which means the element size, 1).
+  this->ProducerSetOptions(0);
+  write_threshold_num_bytes = 123u;
+  this->ProducerGetOptions(&write_threshold_num_bytes);
+  EXPECT_EQ(0u, write_threshold_num_bytes);
+
+  // Should still not have the write threshold signal.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            this->ProducerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, 0, nullptr));
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
+  this->ProducerRemoveAwakable(&waiter, nullptr);
+
+  // Add a waiter.
+  waiter.Init();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            this->ProducerAddAwakable(
+                &waiter, MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, 0, nullptr));
+
+  // Close the consumer.
+  this->ConsumerClose();
+
+  // Wait; the condition should now never be satisfiable.
+  EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
+            waiter.Wait(test::TinyTimeout(), nullptr));
+  hss = HandleSignalsState();
+  this->ProducerRemoveAwakable(&waiter, &hss);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
+
+  this->ProducerClose();
+}
+
 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;
+  uint32_t read_threshold_num_bytes = 123;
   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(
@@ skipping 173 matching lines @@
                 &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