Rationalize AddAwakable...() and RemoveAwakable...() methods.

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 122 matching lines @@
     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) {
     return dpp()->ProducerEndWriteData(num_bytes_written);
   }
   MojoResult ProducerAddAwakable(Awakable* awakable,
+                                 uint64_t context,
+                                 bool force,
                                  MojoHandleSignals signals,
-                                 bool force,
-                                 uint64_t context,
                                  HandleSignalsState* signals_state) {
-    return dpp()->ProducerAddAwakable(awakable, signals, force, context,
+    return dpp()->ProducerAddAwakable(awakable, context, force, signals,
                                       signals_state);
   }
-  void ProducerRemoveAwakable(Awakable* awakable,
+  void ProducerRemoveAwakable(bool match_context,
+                              Awakable* awakable,
+                              uint64_t context,
                               HandleSignalsState* signals_state) {
-    return dpp()->ProducerRemoveAwakable(awakable, signals_state);
+    return dpp()->ProducerRemoveAwakable(match_context, awakable, context,
+                                         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,
@@ skipping 10 matching lines @@
     return dpc()->ConsumerQueryData(num_bytes);
   }
   MojoResult ConsumerBeginReadData(UserPointer<const void*> buffer,
                                    UserPointer<uint32_t> buffer_num_bytes) {
     return dpc()->ConsumerBeginReadData(buffer, buffer_num_bytes);
   }
   MojoResult ConsumerEndReadData(uint32_t num_bytes_read) {
     return dpc()->ConsumerEndReadData(num_bytes_read);
   }
   MojoResult ConsumerAddAwakable(Awakable* awakable,
+                                 uint64_t context,
+                                 bool force,
                                  MojoHandleSignals signals,
-                                 bool force,
-                                 uint64_t context,
                                  HandleSignalsState* signals_state) {
-    return dpc()->ConsumerAddAwakable(awakable, signals, force, context,
+    return dpc()->ConsumerAddAwakable(awakable, context, force, signals,
                                       signals_state);
   }
-  void ConsumerRemoveAwakable(Awakable* awakable,
+  void ConsumerRemoveAwakable(bool match_context,
+                              Awakable* awakable,
+                              uint64_t context,
                               HandleSignalsState* signals_state) {
-    return dpc()->ConsumerRemoveAwakable(awakable, signals_state);
+    return dpc()->ConsumerRemoveAwakable(match_context, awakable, context,
+                                         signals_state);
   }
 
  private:
   DataPipe* dpp() { return helper_->DataPipeForProducer(); }
   DataPipe* dpc() { return helper_->DataPipeForConsumer(); }
 
   std::unique_ptr<Helper> helper_;
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(DataPipeImplTest);
 };
@@ skipping 71 matching lines @@
                   Handle* handle_to_receive) {
     DCHECK(source_i == 0 || source_i == 1);
     size_t dest_i = source_i ^ 1;
 
     // Write the dispatcher to MP |source_i| (port 0). Wait and receive on MP
     // |dest_i| (port 0). (Add the waiter first, to avoid any handling the case
     // where it's already readable.)
     Waiter waiter;
     waiter.Init();
     ASSERT_EQ(MOJO_RESULT_OK, message_pipe(dest_i)->AddAwakable(
-                                  0, &waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                  false, 987, nullptr));
+                                  0, &waiter, 987, false,
+                                  MOJO_HANDLE_SIGNAL_READABLE, nullptr));
     {
       HandleTransport transport(test::HandleTryStartTransport(handle_to_send));
       ASSERT_TRUE(transport.is_valid());
 
       std::vector<HandleTransport> transports;
       transports.push_back(transport);
       ASSERT_EQ(MOJO_RESULT_OK, message_pipe(source_i)->WriteMessage(
                                     0, NullUserPointer(), 0, &transports,
                                     MOJO_WRITE_MESSAGE_FLAG_NONE));
       transport.End();
     }
     uint64_t context = 0;
     ASSERT_EQ(MOJO_RESULT_OK,
               waiter.Wait(test::ActionTimeout(), &context, nullptr));
     EXPECT_EQ(987u, context);
     HandleSignalsState hss = HandleSignalsState();
-    message_pipe(dest_i)->RemoveAwakable(0, &waiter, &hss);
+    message_pipe(dest_i)->RemoveAwakable(0, false, &waiter, 0, &hss);
     EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
               hss.satisfied_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));
     HandleVector read_handles;
     uint32_t read_num_handles = 10;  // Maximum to get.
     ASSERT_EQ(MOJO_RESULT_OK,
@@ skipping 353 matching lines @@
 
   // Read with invalid |num_bytes|.
   num_bytes = sizeof(elements[0]) + 1;
   EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
             this->ConsumerReadData(UserPointer<void>(elements),
                                    MakeUserPointer(&num_bytes), false, false));
 
   // For remote data pipes, we'll have to wait; add the waiter before writing.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 123, nullptr));
+            this->ConsumerAddAwakable(&waiter, 123, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
 
   // Write two elements.
   elements[0] = 123;
   elements[1] = 456;
   num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerWriteData(UserPointer<const void>(elements),
                                     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, nullptr));
   EXPECT_EQ(123u, context);
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // 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 * ...|
@@ skipping 89 matching lines @@
 
   Waiter pwaiter;  // For producer.
   Waiter cwaiter;  // For consumer.
   HandleSignalsState hss;
   uint64_t context;
 
   // Never readable.
   pwaiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
-            this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 12, &hss));
+            this->ProducerAddAwakable(&pwaiter, 12, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, &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);
 
   // Already writable.
   pwaiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
-            this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE,
-                                      false, 34, &hss));
+            this->ProducerAddAwakable(&pwaiter, 34, false,
+                                      MOJO_HANDLE_SIGNAL_WRITABLE, &hss));
 
   // We'll need to wait for readability for the remote cases.
   cwaiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&cwaiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 1234, nullptr));
+            this->ConsumerAddAwakable(&cwaiter, 1234, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, 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));
   EXPECT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes);
 
   // Adding a waiter should now succeed.
   pwaiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE,
-                                      false, 56, nullptr));
+            this->ProducerAddAwakable(&pwaiter, 56, false,
+                                      MOJO_HANDLE_SIGNAL_WRITABLE, nullptr));
   // And it shouldn't be writable yet.
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, pwaiter.Wait(0, nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&pwaiter, &hss);
+  this->ProducerRemoveAwakable(false, &pwaiter, 0, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
   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, nullptr));
   EXPECT_EQ(1234u, context);
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&cwaiter, &hss);
+  this->ConsumerRemoveAwakable(false, &cwaiter, 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);
 
   // 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(123, elements[0]);
   EXPECT_EQ(-1, elements[1]);
 
   // Add a waiter.
   pwaiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE,
-                                      false, 56, nullptr));
+            this->ProducerAddAwakable(&pwaiter, 56, false,
+                                      MOJO_HANDLE_SIGNAL_WRITABLE, nullptr));
   // And it still shouldn't be writable yet.
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, pwaiter.Wait(0, nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&pwaiter, &hss);
+  this->ProducerRemoveAwakable(false, &pwaiter, 0, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
   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,
-                                      false, 78, nullptr));
+            this->ProducerAddAwakable(&pwaiter, 78, false,
+                                      MOJO_HANDLE_SIGNAL_WRITABLE, 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, nullptr));
   EXPECT_EQ(78u, context);
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&pwaiter, &hss);
+  this->ProducerRemoveAwakable(false, &pwaiter, 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);
 
   // 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);
 
   static_cast<int32_t*>(buffer)[0] = 789;
   EXPECT_EQ(MOJO_RESULT_OK, this->ProducerEndWriteData(static_cast<uint32_t>(
                                 1u * sizeof(elements[0]))));
 
   // Add a waiter.
   pwaiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE,
-                                      false, 90, nullptr));
+            this->ProducerAddAwakable(&pwaiter, 90, false,
+                                      MOJO_HANDLE_SIGNAL_WRITABLE, nullptr));
 
   // Read one element, using a two-phase read.
   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);
   // Since we only read one element (after having written three in all), the
   // two-phase read should only allow us to read one. This checks an
   // implementation detail!
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
   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, nullptr));
   EXPECT_EQ(90u, context);
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&pwaiter, &hss);
+  this->ProducerRemoveAwakable(false, &pwaiter, 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);
 
   // 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);
 
   // Add a waiter.
   pwaiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ProducerAddAwakable(&pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE,
-                                      false, 12, nullptr));
+            this->ProducerAddAwakable(&pwaiter, 12, false,
+                                      MOJO_HANDLE_SIGNAL_WRITABLE, nullptr));
 
   // Close the consumer.
   this->ConsumerClose();
 
   // It should now be never-writable.
   context = 0;
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
             pwaiter.Wait(test::TinyTimeout(), &context, nullptr));
   EXPECT_EQ(12u, context);
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&pwaiter, &hss);
+  this->ProducerRemoveAwakable(false, &pwaiter, 0, &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, PeerClosedProducerWaiting) {
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
       2 * sizeof(int32_t)                       // |capacity_num_bytes|.
   };
   this->Create(options);
   this->DoTransfer();
 
   Waiter waiter;
   HandleSignalsState hss;
   uint64_t context;
 
   // Add a waiter.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ProducerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
-                                      false, 12, nullptr));
+            this->ProducerAddAwakable(&waiter, 12, false,
+                                      MOJO_HANDLE_SIGNAL_PEER_CLOSED, nullptr));
 
   // Close the consumer.
   this->ConsumerClose();
 
   // It should be signaled.
   context = 0;
   EXPECT_EQ(MOJO_RESULT_OK,
             waiter.Wait(test::TinyTimeout(), &context, nullptr));
   EXPECT_EQ(12u, context);
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&waiter, &hss);
+  this->ProducerRemoveAwakable(false, &waiter, 0, &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, PeerClosedConsumerWaiting) {
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
       2 * sizeof(int32_t)                       // |capacity_num_bytes|.
   };
   this->Create(options);
   this->DoTransfer();
 
   Waiter waiter;
   HandleSignalsState hss;
   uint64_t context;
 
   // Add a waiter.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
-                                      false, 12, nullptr));
+            this->ConsumerAddAwakable(&waiter, 12, false,
+                                      MOJO_HANDLE_SIGNAL_PEER_CLOSED, nullptr));
 
   // Close the producer.
   this->ProducerClose();
 
   // It should be signaled.
   context = 0;
   EXPECT_EQ(MOJO_RESULT_OK,
             waiter.Wait(test::TinyTimeout(), &context, nullptr));
   EXPECT_EQ(12u, context);
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 0, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
   this->ConsumerClose();
 }
 
 TYPED_TEST(DataPipeImplTest, BasicConsumerWaiting) {
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
       1000 * sizeof(int32_t)                    // |capacity_num_bytes|.
   };
   this->Create(options);
   this->DoTransfer();
 
   Waiter waiter;
   Waiter waiter2;
   HandleSignalsState hss;
   uint64_t context;
 
   // Never writable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE,
-                                      false, 12, &hss));
+            this->ConsumerAddAwakable(&waiter, 12, false,
+                                      MOJO_HANDLE_SIGNAL_WRITABLE, &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 waiter: not yet readable.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 34, nullptr));
+            this->ConsumerAddAwakable(&waiter, 34, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, 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, nullptr));
   EXPECT_EQ(34u, context);
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // 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,
-                                      false, 78, &hss));
+            this->ConsumerAddAwakable(&waiter, 78, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, &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);
 
   // 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,
-                                      false, 78, &hss));
+            this->ConsumerAddAwakable(&waiter, 78, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, &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 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(456, elements[0]);
   EXPECT_EQ(-1, elements[1]);
 
   // Adding a waiter should now succeed.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 90, nullptr));
+            this->ConsumerAddAwakable(&waiter, 90, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
 
   // Write one element.
   elements[0] = 789;
   elements[1] = -1;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   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, nullptr));
   EXPECT_EQ(90u, context);
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // 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,
-                                      false, 12, nullptr));
+            this->ConsumerAddAwakable(&waiter, 12, false,
+                                      MOJO_HANDLE_SIGNAL_PEER_CLOSED, 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,
-                                      false, 34, &hss));
+            this->ConsumerAddAwakable(&waiter2, 34, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, &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 |
                 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, nullptr));
   EXPECT_EQ(12u, context);
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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 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(789, elements[0]);
   EXPECT_EQ(-1, elements[1]);
 
   // Should be never-readable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 56, &hss));
+            this->ConsumerAddAwakable(&waiter, 56, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
   this->ConsumerClose();
 }
 
 // Test with two-phase APIs and also closing the producer with an active
 // consumer waiter.
 TYPED_TEST(DataPipeImplTest, ConsumerWaitingTwoPhase) {
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
       1000 * sizeof(int32_t)                    // |capacity_num_bytes|.
   };
   this->Create(options);
   this->DoTransfer();
 
   Waiter waiter;
   HandleSignalsState hss;
   uint64_t context;
 
   // Add waiter: not yet readable.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 12, nullptr));
+            this->ConsumerAddAwakable(&waiter, 12, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
 
   // Write two elements.
   int32_t* elements = nullptr;
   void* buffer = nullptr;
   uint32_t num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerBeginWriteData(MakeUserPointer(&buffer),
                                          MakeUserPointer(&num_bytes)));
   EXPECT_TRUE(buffer);
   EXPECT_GE(num_bytes, static_cast<uint32_t>(3u * sizeof(elements[0])));
   elements = static_cast<int32_t*>(buffer);
   elements[0] = 123;
   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, nullptr));
   EXPECT_EQ(12u, context);
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // 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,
-                                      false, 34, &hss));
+            this->ConsumerAddAwakable(&waiter, 34, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, &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 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);
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
   read_elements = static_cast<const int32_t*>(read_buffer);
   EXPECT_EQ(456, read_elements[0]);
   EXPECT_EQ(MOJO_RESULT_OK, this->ConsumerEndReadData(static_cast<uint32_t>(
                                 1u * sizeof(elements[0]))));
 
   // Adding a waiter should now succeed.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 56, nullptr));
+            this->ConsumerAddAwakable(&waiter, 56, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
 
   // Close the producer.
   this->ProducerClose();
 
   // Should be never-readable.
   context = 0;
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
             waiter.Wait(test::TinyTimeout(), &context, nullptr));
   EXPECT_EQ(56u, context);
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 0, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
   this->ConsumerClose();
 }
 
 // Tests that data pipes aren't writable/readable during two-phase writes/reads.
 TYPED_TEST(DataPipeImplTest, BasicTwoPhaseWaiting) {
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
       1000 * sizeof(int32_t)                    // |capacity_num_bytes|.
   };
   this->Create(options);
   this->DoTransfer();
 
   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,
-                                      false, 0, &hss));
+            this->ProducerAddAwakable(&pwaiter, 0, false,
+                                      MOJO_HANDLE_SIGNAL_WRITABLE, &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);
 
   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,
-                                      false, 1, nullptr));
+            this->ProducerAddAwakable(&pwaiter, 1, false,
+                                      MOJO_HANDLE_SIGNAL_WRITABLE, nullptr));
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, pwaiter.Wait(0, nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&pwaiter, &hss);
+  this->ProducerRemoveAwakable(false, &pwaiter, 0, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
   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,
-                                      false, 2, nullptr));
+            this->ConsumerAddAwakable(&cwaiter, 2, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, 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,
-                                      false, 3, &hss));
+            this->ProducerAddAwakable(&pwaiter, 3, false,
+                                      MOJO_HANDLE_SIGNAL_WRITABLE, &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);
 
   // It should become readable.
   EXPECT_EQ(MOJO_RESULT_OK,
             cwaiter.Wait(test::TinyTimeout(), nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&cwaiter, &hss);
+  this->ConsumerRemoveAwakable(false, &cwaiter, 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);
 
   // 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,
-                                      false, 5, &hss));
+            this->ConsumerAddAwakable(&cwaiter, 5, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, &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);
 
   // 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),
                                         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,
-                                      false, 6, &hss));
+            this->ProducerAddAwakable(&pwaiter, 6, false,
+                                      MOJO_HANDLE_SIGNAL_WRITABLE, &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);
 
   // But not readable.
   cwaiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&cwaiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 7, nullptr));
+            this->ConsumerAddAwakable(&cwaiter, 7, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, cwaiter.Wait(0, nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&cwaiter, &hss);
+  this->ConsumerRemoveAwakable(false, &cwaiter, 0, &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);
 
   // 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,
-                                      false, 8, &hss));
+            this->ConsumerAddAwakable(&cwaiter, 8, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, &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);
 
   this->ProducerClose();
   this->ConsumerClose();
 }
 
@@ skipping 25 matching lines @@
 
   // Should still be empty.
   num_bytes = ~0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerQueryData(MakeUserPointer(&num_bytes)));
   EXPECT_EQ(0u, num_bytes);
 
   // Add waiter.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 1, nullptr));
+            this->ConsumerAddAwakable(&waiter, 1, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
 
   // Write some data.
   num_bytes = 5u * sizeof(int32_t);
   Seq(100, MOJO_ARRAYSIZE(buffer), buffer);
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerWriteData(UserPointer<const void>(buffer),
                                     MakeUserPointer(&num_bytes), true));
   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, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // Half full.
   num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerQueryData(MakeUserPointer(&num_bytes)));
@@ skipping 83 matching lines @@
 
   // Three left.
   num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerQueryData(MakeUserPointer(&num_bytes)));
   EXPECT_EQ(3u * sizeof(int32_t), num_bytes);
 
   // We'll need to wait for the peer closed to propagate.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
-                                      false, 2, nullptr));
+            this->ConsumerAddAwakable(&waiter, 2, false,
+                                      MOJO_HANDLE_SIGNAL_PEER_CLOSED, nullptr));
 
   // Close the producer, then test producer-closed cases.
   this->ProducerClose();
 
   // Wait.
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // Try reading too much; "failed precondition" since the producer is closed.
   num_bytes = 4u * sizeof(int32_t);
   memset(buffer, 0xab, sizeof(buffer));
@@ skipping 57 matching lines @@
   ASSERT_EQ(100u, validated_options.capacity_num_bytes);
   this->Create(options);
   this->DoTransfer();
 
   Waiter waiter;
   HandleSignalsState hss;
 
   // Add waiter.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 1, nullptr));
+            this->ConsumerAddAwakable(&waiter, 1, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
 
   // Write 20 bytes.
   uint32_t num_bytes = 20u;
   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, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // Read 10 bytes.
   unsigned char read_buffer[1000] = {0};
   num_bytes = 10u;
   EXPECT_EQ(MOJO_RESULT_OK,
@@ skipping 170 matching lines @@
   };
   this->Create(options);
   this->DoTransfer();
 
   Waiter waiter;
   HandleSignalsState hss;
 
   // Add waiter.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 1, nullptr));
+            this->ConsumerAddAwakable(&waiter, 1, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
 
   // Write some data, so we'll have something to read.
   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);
 
   // Start two-phase write.
   void* write_buffer_ptr = nullptr;
   num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             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, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // 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);
 
   // Add waiter.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ProducerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
-                                      false, 1, nullptr));
+            this->ProducerAddAwakable(&waiter, 1, false,
+                                      MOJO_HANDLE_SIGNAL_PEER_CLOSED, nullptr));
 
   // Close the consumer.
   this->ConsumerClose();
 
   // Wait for producer to know that the consumer is closed.
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&waiter, &hss);
+  this->ProducerRemoveAwakable(false, &waiter, 0, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
   // Actually write some data. (Note: Premature freeing of the buffer would
   // probably only be detected under ASAN or similar.)
   memcpy(write_buffer_ptr, kTestData, kTestDataSize);
   // Note: Even though the consumer has been closed, ending the two-phase
   // write will report success.
   EXPECT_EQ(MOJO_RESULT_OK, this->ProducerEndWriteData(kTestDataSize));
 
@@ skipping 61 matching lines @@
   // Write some data, so we'll have something to read.
   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);
 
   // Add waiter.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
-                                      false, 1, nullptr));
+            this->ConsumerAddAwakable(&waiter, 1, false,
+                                      MOJO_HANDLE_SIGNAL_PEER_CLOSED, 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, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // Peek that data.
   char buffer[1000];
   num_bytes = static_cast<uint32_t>(sizeof(buffer));
@@ skipping 47 matching lines @@
   };
   this->Create(options);
   this->DoTransfer();
 
   Waiter waiter;
   HandleSignalsState hss;
 
   // Add waiter.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 0, nullptr));
+            this->ConsumerAddAwakable(&waiter, 0, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
 
   // Write some data, so we'll have something to read.
   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, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // 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]);
 
   // Add waiter again.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 0, nullptr));
+            this->ConsumerAddAwakable(&waiter, 0, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
 
   // Close the producer.
   this->ProducerClose();
 
   // Wait.
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
             waiter.Wait(test::TinyTimeout(), nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 0, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
   this->ConsumerClose();
 }
 
 // During a two-phase read, the consumer is not readable so it may be waited
 // upon (to become readable again). If the producer is closed and the two-phase
 // read consumes the remaining data, that wait should become unsatisfiable.
 TYPED_TEST(DataPipeImplTest, BeginReadCloseProducerWaitEndReadNoData) {
   const int64_t kTestData = 123456789012345LL;
   const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData));
 
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                           // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
       kTestDataSize,                            // |element_num_bytes|.
       100u * kTestDataSize                      // |capacity_num_bytes|.
   };
   this->Create(options);
   this->DoTransfer();
 
   Waiter waiter;
   HandleSignalsState hss;
 
   // Add waiter (for the consumer to become readable).
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 0, nullptr));
+            this->ConsumerAddAwakable(&waiter, 0, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
 
   // Write some data, so we'll have something to read.
   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, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // 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,
-                                      false, 0, nullptr));
+            this->ConsumerAddAwakable(&waiter, 0, false,
+                                      MOJO_HANDLE_SIGNAL_PEER_CLOSED, nullptr));
 
   // Close the producer.
   this->ProducerClose();
 
   // Wait for producer close to be detected.
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyTimeout(), nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 0, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
   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,
-                                      false, 0, nullptr));
+            this->ConsumerAddAwakable(&waiter, 0, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
 
   // Complete the two-phase read.
   EXPECT_EQ(MOJO_RESULT_OK, this->ConsumerEndReadData(kTestDataSize));
 
   // Wait.
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
             waiter.Wait(test::TinyTimeout(), nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 0, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
   this->ConsumerClose();
 }
 
 // During a two-phase write, the producer is not writable so it may be waited
 // upon (to become writable again). If the consumer is closed, that wait should
 // become unsatisfiable.
 TYPED_TEST(DataPipeImplTest, BeginWriteCloseConsumerWaitEndWrite) {
@@ skipping 13 matching lines @@
   // Start a two-phase write.
   void* write_ptr = nullptr;
   uint32_t num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ProducerBeginWriteData(MakeUserPointer(&write_ptr),
                                          MakeUserPointer(&num_bytes)));
 
   // Add waiter (for the consumer to be closed).
   waiter1.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ProducerAddAwakable(&waiter1, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
-                                      false, 0, nullptr));
+            this->ProducerAddAwakable(&waiter1, 0, false,
+                                      MOJO_HANDLE_SIGNAL_PEER_CLOSED, nullptr));
 
   // Add a separate waiter (for the producer to become writable).
   waiter2.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ProducerAddAwakable(&waiter2, MOJO_HANDLE_SIGNAL_WRITABLE,
-                                      false, 0, nullptr));
+            this->ProducerAddAwakable(&waiter2, 0, false,
+                                      MOJO_HANDLE_SIGNAL_WRITABLE, nullptr));
 
   // Close the consumer.
   this->ConsumerClose();
 
   // Wait for the consumer close to be detected.
   // Note: If we didn't wait for the consumer close to be detected before
   // completing the two-phase write, wait might succeed (in the remote cases).
   // This is because the first |Awake()| "wins".
   EXPECT_EQ(MOJO_RESULT_OK,
             waiter1.Wait(test::TinyTimeout(), nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&waiter1, &hss);
+  this->ProducerRemoveAwakable(false, &waiter1, 0, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
   // Complete the two-phase write (with nothing written).
   EXPECT_EQ(MOJO_RESULT_OK, this->ProducerEndWriteData(0u));
 
   // Wait.
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
             waiter2.Wait(test::TinyTimeout(), nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&waiter2, &hss);
+  this->ProducerRemoveAwakable(false, &waiter2, 0, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
   this->ProducerClose();
 }
 
 // Test that two-phase reads/writes behave correctly when given invalid
 // arguments.
 TYPED_TEST(DataPipeImplTest, TwoPhaseMoreInvalidArguments) {
   const MojoCreateDataPipeOptions options = {
@@ skipping 68 matching lines @@
 
   // Still no data.
   num_bytes = 1000u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerQueryData(MakeUserPointer(&num_bytes)));
   EXPECT_EQ(0u, num_bytes);
 
   // Add waiter.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK,
-            this->ConsumerAddAwakable(&waiter, MOJO_HANDLE_SIGNAL_READABLE,
-                                      false, 1, nullptr));
+            this->ConsumerAddAwakable(&waiter, 1, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, nullptr));
 
   // Now write some data, so we'll be able to try reading.
   int32_t element = 123;
   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, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // One element available.
   num_bytes = 0u;
   EXPECT_EQ(MOJO_RESULT_OK,
             this->ConsumerQueryData(MakeUserPointer(&num_bytes)));
@@ skipping 123 matching lines @@
   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, false, 0, &hss));
+                &waiter, 0, false, MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, &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,
-                                      false, 0, nullptr));
+            this->ConsumerAddAwakable(&read_waiter, 0, false,
+                                      MOJO_HANDLE_SIGNAL_READABLE, 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,
-                                false, 0, nullptr));
+      this->ProducerAddAwakable(&waiter, 0, false,
+                                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, 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,
-                                false, 0, nullptr));
+      this->ProducerAddAwakable(&waiter, 0, false,
+                                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, 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,
-                                false, 0, nullptr));
+                                &waiter, 0, false,
+                                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, nullptr));
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&waiter, &hss);
+  this->ProducerRemoveAwakable(false, &waiter, 0, &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,
-                                false, 0, nullptr));
+                                &waiter, 0, false,
+                                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, nullptr));
 
   // Wait for the consumer to be readable.
   EXPECT_EQ(MOJO_RESULT_OK,
             read_waiter.Wait(test::TinyTimeout(), nullptr, nullptr));
-  this->ConsumerRemoveAwakable(&read_waiter, nullptr);
+  this->ConsumerRemoveAwakable(false, &read_waiter, 0, 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, nullptr));
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&waiter, &hss);
+  this->ProducerRemoveAwakable(false, &waiter, 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);
 
   // 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,
-                                false, 0, nullptr));
+                                &waiter, 0, false,
+                                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, nullptr));
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr, nullptr));
-  this->ProducerRemoveAwakable(&waiter, nullptr);
+  this->ProducerRemoveAwakable(false, &waiter, 0, 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,
-                                false, 0, nullptr));
+                                &waiter, 0, false,
+                                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, nullptr));
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr, nullptr));
-  this->ProducerRemoveAwakable(&waiter, nullptr);
+  this->ProducerRemoveAwakable(false, &waiter, 0, nullptr);
 
   // Add a waiter.
   waiter.Init();
   ASSERT_EQ(MOJO_RESULT_OK, this->ProducerAddAwakable(
-                                &waiter, MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD,
-                                false, 0, nullptr));
+                                &waiter, 0, false,
+                                MOJO_HANDLE_SIGNAL_WRITE_THRESHOLD, 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, nullptr));
   hss = HandleSignalsState();
-  this->ProducerRemoveAwakable(&waiter, &hss);
+  this->ProducerRemoveAwakable(false, &waiter, 0, &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 = 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(
-                &waiter, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, false, 0, nullptr));
+                &waiter, 0, false, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, nullptr));
   // Trivial wait: it shouldn't have the read threshold signal.
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 0, &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, false, 0, nullptr));
+                &waiter, 0, false, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, 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, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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 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, false, 0, &hss));
+            this->ConsumerAddAwakable(&waiter, 0, false,
+                                      MOJO_HANDLE_SIGNAL_READ_THRESHOLD, &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, false, 0, nullptr));
+                &waiter, 0, false, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, 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, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 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);
 
   // 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, false, 0, nullptr));
+                &waiter, 0, false, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, nullptr));
   // Trivial wait: it shouldn't have the read threshold signal.
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 0, &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, false, 0, nullptr));
+                &waiter, 0, false, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, nullptr));
   // Trivial wait: it shouldn't have the read threshold signal.
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 0, &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, false, 0, nullptr));
+                &waiter, 0, false, MOJO_HANDLE_SIGNAL_READ_THRESHOLD, 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, nullptr));
   hss = HandleSignalsState();
-  this->ConsumerRemoveAwakable(&waiter, &hss);
+  this->ConsumerRemoveAwakable(false, &waiter, 0, &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, false, 0, &hss));
+            this->ConsumerAddAwakable(&waiter, 0, false,
+                                      MOJO_HANDLE_SIGNAL_READ_THRESHOLD, &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, false, 0, &hss));
+            this->ConsumerAddAwakable(&waiter, 0, false,
+                                      MOJO_HANDLE_SIGNAL_READ_THRESHOLD, &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