Mojo: Add signal for new data pipe consumer data

mojo/edk/system/data_pipe_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.
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <memory>
 
 #include "base/bind.h"
@@ skipping 24 matching lines @@
 // certain amount of data to be available). This is the maximum number of
 // iterations (separated by a short sleep).
 // TODO(vtl): Get rid of this.
 const size_t kMaxPoll = 100;
 
 // Used in Multiprocess test.
 const size_t kMultiprocessCapacity = 37;
 const char kMultiprocessTestData[] = "hello i'm a string that is 36 bytes";
 const int kMultiprocessMaxIter = 5;
 
+// TODO(rockot): There are many uses of ASSERT where EXPECT would be more
+// appropriate. Fix this.
+
 class DataPipeTest : public test::MojoTestBase {
  public:
   DataPipeTest() : producer_(MOJO_HANDLE_INVALID),
                    consumer_(MOJO_HANDLE_INVALID) {}
 
   ~DataPipeTest() override {
     if (producer_ != MOJO_HANDLE_INVALID)
       CHECK_EQ(MOJO_RESULT_OK, MojoClose(producer_));
     if (consumer_ != MOJO_HANDLE_INVALID)
       CHECK_EQ(MOJO_RESULT_OK, MojoClose(consumer_));
@@ skipping 99 matching lines @@
   elements[0] = 123;
   elements[1] = 456;
   num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
   ASSERT_EQ(MOJO_RESULT_OK, WriteData(&elements[0], &num_bytes));
 
   // Now wait for the other side to become readable.
   MojoHandleSignalsState state;
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &state));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, state.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            state.satisfied_signals);
 
   elements[0] = -1;
   elements[1] = -1;
   ASSERT_EQ(MOJO_RESULT_OK, ReadData(&elements[0], &num_bytes));
   ASSERT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes);
   ASSERT_EQ(elements[0], 123);
   ASSERT_EQ(elements[1], 456);
 }
 
 // Tests creation of data pipes with various (valid) options.
@@ skipping 65 matching lines @@
   elements[1] = 456;
   num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
   ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes));
   // It should have written everything (even without "all or none").
   ASSERT_EQ(2u * sizeof(elements[0]), num_bytes);
 
   // Wait.
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Query.
   // TODO(vtl): It's theoretically possible (though not with the current
   // implementation/configured limits) that not all the data has arrived yet.
   // (The theoretically-correct assertion here is that |num_bytes| is |1 * ...|
   // or |2 * ...|.)
   num_bytes = 0;
   ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
   ASSERT_EQ(2 * sizeof(elements[0]), num_bytes);
@@ skipping 85 matching lines @@
   // Write two elements.
   int32_t elements[2] = {123, 456};
   uint32_t num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
   ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
   ASSERT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes);
 
   // Wait for data to become available to the consumer.
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Peek one element.
   elements[0] = -1;
   elements[1] = -1;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true, true));
   ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
   ASSERT_EQ(123, elements[0]);
   ASSERT_EQ(-1, elements[1]);
@@ skipping 103 matching lines @@
       1000 * sizeof(int32_t)                    // |capacity_num_bytes|.
   };
   ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
   MojoHandleSignalsState hss;
 
   // Never writable.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_WRITABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(0u, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(0u, hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Write two elements.
   int32_t elements[2] = {123, 456};
   uint32_t num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
   ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
 
   // Wait for readability.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Discard one element.
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   ASSERT_EQ(MOJO_RESULT_OK, DiscardData(&num_bytes, true));
   ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
 
   // Should still be readable.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Peek one element.
   elements[0] = -1;
   elements[1] = -1;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true, true));
   ASSERT_EQ(456, elements[0]);
   ASSERT_EQ(-1, elements[1]);
 
   // Should still be readable.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Read one element.
   elements[0] = -1;
   elements[1] = -1;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true));
   ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
   ASSERT_EQ(456, elements[0]);
   ASSERT_EQ(-1, elements[1]);
 
   // Write one element.
   elements[0] = 789;
   elements[1] = -1;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
 
   // Waiting should now succeed.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Close the producer.
   CloseProducer();
 
   // Should still be readable.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_TRUE((hss.satisfied_signals & MOJO_HANDLE_SIGNAL_READABLE) != 0);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_TRUE(hss.satisfied_signals & (MOJO_HANDLE_SIGNAL_READABLE |
+                                       MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE));
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Wait for the peer closed signal.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_TRUE((hss.satisfied_signals & MOJO_HANDLE_SIGNAL_PEER_CLOSED) != 0);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Read one element.
   elements[0] = -1;
   elements[1] = -1;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true));
   ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
   ASSERT_EQ(789, elements[0]);
   ASSERT_EQ(-1, elements[1]);
 
   // Should be never-readable.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
   ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
   ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 }
 
+TEST_F(DataPipeTest, ConsumerNewDataReadable) {
+  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|.
+  };
+  EXPECT_EQ(MOJO_RESULT_OK, Create(&options));
+
+  int32_t elements[2] = {123, 456};
+  uint32_t num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
+  EXPECT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
+
+  // The consumer handle should appear to be readable and have new data.
+  EXPECT_EQ(MOJO_RESULT_OK, MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
+                                     MOJO_DEADLINE_INDEFINITE, nullptr));
+  EXPECT_EQ(MOJO_RESULT_OK,
+            MojoWait(consumer_, MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+                     MOJO_DEADLINE_INDEFINITE, nullptr));
+
+  // Now try to read a minimum of 6 elements.
+  int32_t read_elements[6];
+  uint32_t num_read_bytes = sizeof(read_elements);
+  EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE,
+            MojoReadData(consumer_, read_elements, &num_read_bytes,
+                         MOJO_READ_DATA_FLAG_ALL_OR_NONE));
+
+  // The consumer should still appear to be readable, but not with new data.
+  EXPECT_EQ(MOJO_RESULT_OK,
+            MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE, 0, nullptr));
+  EXPECT_EQ(
+      MOJO_RESULT_DEADLINE_EXCEEDED,
+      MojoWait(consumer_, MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, 0, nullptr));
+
+  // Write four more elements.
+  EXPECT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
+  EXPECT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
+
+  // The consumer handle should once again appear to be readable with new data.
+  EXPECT_EQ(MOJO_RESULT_OK, MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
+                                     MOJO_DEADLINE_INDEFINITE, nullptr));
+  EXPECT_EQ(MOJO_RESULT_OK,
+            MojoWait(consumer_, MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+                     MOJO_DEADLINE_INDEFINITE, nullptr));
+
+  // Read should succeed this time.
+  EXPECT_EQ(MOJO_RESULT_OK,
+            MojoReadData(consumer_, read_elements, &num_read_bytes,
+                         MOJO_READ_DATA_FLAG_ALL_OR_NONE));
+
+  // And once again the consumer is unreadable.
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED,
+            MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE, 0, nullptr));
+  EXPECT_EQ(
+      MOJO_RESULT_DEADLINE_EXCEEDED,
+      MojoWait(consumer_, MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, 0, nullptr));
+}
+
 // Test with two-phase APIs and also closing the producer with an active
 // consumer waiter.
 TEST_F(DataPipeTest, 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|.
   };
   ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
   MojoHandleSignalsState hss;
 
   // Write two elements.
   int32_t* elements = nullptr;
   void* buffer = nullptr;
   // Request room for three (but we'll only write two).
   uint32_t num_bytes = static_cast<uint32_t>(3u * sizeof(elements[0]));
-  ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&buffer, &num_bytes, true));
+  ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&buffer, &num_bytes, false));
   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;
   ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(2u * sizeof(elements[0])));
 
   // Wait for readability.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Read one element.
   // Request two in all-or-none mode, but only read one.
   const void* read_buffer = nullptr;
   num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
   ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer, &num_bytes, true));
   EXPECT_TRUE(read_buffer);
   ASSERT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes);
   const int32_t* read_elements = static_cast<const int32_t*>(read_buffer);
   ASSERT_EQ(123, read_elements[0]);
   ASSERT_EQ(MOJO_RESULT_OK, EndReadData(1u * sizeof(elements[0])));
 
   // Should still be readable.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             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]));
   ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer, &num_bytes));
   EXPECT_TRUE(read_buffer);
   ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
   read_elements = static_cast<const int32_t*>(read_buffer);
@@ skipping 43 matching lines @@
             MojoWait(producer_, MOJO_HANDLE_SIGNAL_WRITABLE, 0, &hss));
   ASSERT_EQ(0u, hss.satisfied_signals);
   ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
             hss.satisfiable_signals);
 
   // It shouldn't be readable yet either (we'll wait later).
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE, 0, &hss));
   ASSERT_EQ(0u, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   static_cast<int32_t*>(write_ptr)[0] = 123;
   ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(1u * sizeof(int32_t)));
 
   // It should immediately be writable again.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(producer_, MOJO_HANDLE_SIGNAL_WRITABLE, 0, &hss));
   ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
   ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
             hss.satisfiable_signals);
 
   // It should become readable.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Start another two-phase write and check that it's readable even in the
   // middle of it.
   num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t));
   write_ptr = nullptr;
   ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_ptr, &num_bytes));
   EXPECT_TRUE(write_ptr);
   EXPECT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(int32_t)));
 
   // It should be readable.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // End the two-phase write without writing anything.
   ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(0u));
 
   // Start a two-phase read.
   num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t));
   const void* read_ptr = nullptr;
   ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_ptr, &num_bytes));
   EXPECT_TRUE(read_ptr);
   ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(int32_t)), num_bytes);
 
   // At this point, it should still be writable.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(producer_, MOJO_HANDLE_SIGNAL_WRITABLE, 0, &hss));
   ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
   ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
             hss.satisfiable_signals);
 
   // But not readable.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE, 0, &hss));
   ASSERT_EQ(0u, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // End the two-phase read without reading anything.
   ASSERT_EQ(MOJO_RESULT_OK, EndReadData(0u));
 
   // It should be readable again.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE, 0, &hss));
   ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 }
 
 void Seq(int32_t start, size_t count, int32_t* out) {
   for (size_t i = 0; i < count; i++)
     out[i] = start + static_cast<int32_t>(i);
 }
 
 TEST_F(DataPipeTest, AllOrNone) {
   const MojoCreateDataPipeOptions options = {
@@ skipping 24 matching lines @@
 
   // 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.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
             hss.satisfiable_signals);
 
   // Half full.
   num_bytes = 0u;
   ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
   ASSERT_EQ(5u * sizeof(int32_t), num_bytes);
 
   // Try writing more than the available capacity of the pipe, but less than the
   // total capacity.
   num_bytes = 6u * sizeof(int32_t);
@@ skipping 67 matching lines @@
   ASSERT_EQ(3u * sizeof(int32_t), num_bytes);
 
   // Close the producer, then test producer-closed cases.
   CloseProducer();
 
   // Wait.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
             hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
             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));
   ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
             ReadData(buffer, &num_bytes, true));
   memset(expected_buffer, 0xab, sizeof(expected_buffer));
   ASSERT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
 
@@ skipping 42 matching lines @@
 
   // Write 20 bytes.
   uint32_t num_bytes = 20u;
   ASSERT_EQ(MOJO_RESULT_OK, WriteData(&test_data[0], &num_bytes, true));
   ASSERT_EQ(20u, num_bytes);
 
   // Wait for data.
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_TRUE((hss.satisfied_signals & MOJO_HANDLE_SIGNAL_READABLE) != 0);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_TRUE(hss.satisfied_signals & MOJO_HANDLE_SIGNAL_READABLE);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Read 10 bytes.
   unsigned char read_buffer[1000] = {0};
   num_bytes = 10u;
   ASSERT_EQ(MOJO_RESULT_OK, ReadData(read_buffer, &num_bytes, true));
   ASSERT_EQ(10u, num_bytes);
   ASSERT_EQ(0, memcmp(read_buffer, &test_data[0], 10u));
 
   // Check that a two-phase write can now only write (at most) 80 bytes. (This
@@ skipping 141 matching lines @@
   ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_buffer_ptr, &num_bytes));
   EXPECT_TRUE(write_buffer_ptr);
   ASSERT_GT(num_bytes, kTestDataSize);
 
   // Wait for data.
   // TODO(vtl): (See corresponding TODO in AllOrNone.)
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Start two-phase read.
   const void* read_buffer_ptr = nullptr;
   num_bytes = 0u;
   ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer_ptr, &num_bytes));
   EXPECT_TRUE(read_buffer_ptr);
   ASSERT_EQ(kTestDataSize, num_bytes);
 
   // Close the consumer.
@@ skipping 68 matching lines @@
 
   // Close the producer.
   CloseProducer();
 
   // Wait. (Note that once the consumer knows that the producer is closed, it
   // must also know about all the data that was sent.)
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Peek that data.
   char buffer[1000];
   num_bytes = static_cast<uint32_t>(sizeof(buffer));
   ASSERT_EQ(MOJO_RESULT_OK, ReadData(buffer, &num_bytes, false, true));
   ASSERT_EQ(kTestDataSize, num_bytes);
   ASSERT_EQ(0, memcmp(buffer, kTestData, kTestDataSize));
 
   // Read that data.
@@ skipping 36 matching lines @@
   // Write some data.
   uint32_t num_bytes = kTestDataSize;
   ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes));
   ASSERT_EQ(kTestDataSize, num_bytes);
 
   // Wait for the data.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Begin a two-phase read.
   const void* read_buffer_ptr = nullptr;
   uint32_t read_buffer_size = 0u;
   ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer_ptr, &read_buffer_size));
 
   // Write more data.
   const char kExtraData[] = "bye world";
   const uint32_t kExtraDataSize = static_cast<uint32_t>(sizeof(kExtraData));
   num_bytes = kExtraDataSize;
   ASSERT_EQ(MOJO_RESULT_OK, WriteData(kExtraData, &num_bytes));
   ASSERT_EQ(kExtraDataSize, num_bytes);
 
   // Close the producer.
   CloseProducer();
 
   // Wait. (Note that once the consumer knows that the producer is closed, it
   // must also have received the extra data).
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Read the two phase memory to check it's still valid.
   ASSERT_EQ(0, memcmp(read_buffer_ptr, kTestData, kTestDataSize));
   EndReadData(read_buffer_size);
 }
 
 // Test that two-phase reads/writes behave correctly when given invalid
 // arguments.
 TEST_F(DataPipeTest, TwoPhaseMoreInvalidArguments) {
@@ skipping 65 matching lines @@
   int32_t element = 123;
   num_bytes = 1u * sizeof(int32_t);
   ASSERT_EQ(MOJO_RESULT_OK, WriteData(&element, &num_bytes));
 
   // Wait for data.
   // TODO(vtl): (See corresponding TODO in AllOrNone.)
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // One element available.
   num_bytes = 0u;
   ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
   ASSERT_EQ(1u * sizeof(int32_t), num_bytes);
 
   // Try "ending" a two-phase read when one isn't active.
   ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, EndReadData(1u * sizeof(int32_t)));
 
@@ skipping 46 matching lines @@
   // Write some data.
   uint32_t num_bytes = kTestDataSize;
   ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes));
   ASSERT_EQ(kTestDataSize, num_bytes);
 
   // Wait for the data.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Check the data.
   const void* read_buffer = nullptr;
   num_bytes = 0u;
   ASSERT_EQ(MOJO_RESULT_OK,
             BeginReadData(&read_buffer, &num_bytes, false));
   ASSERT_EQ(0, memcmp(read_buffer, kTestData, kTestDataSize));
   EndReadData(num_bytes);
 
@@ skipping 19 matching lines @@
   const uint32_t kExtraDataSize = static_cast<uint32_t>(sizeof(kExtraData));
   num_bytes = kExtraDataSize;
   ASSERT_EQ(MOJO_RESULT_OK, WriteData(kExtraData, &num_bytes));
   ASSERT_EQ(kExtraDataSize, num_bytes);
 
   // Wait for it.
   hss = MojoHandleSignalsState();
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
                      MOJO_DEADLINE_INDEFINITE, &hss));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
+            hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             hss.satisfiable_signals);
 
   // Check the second write.
   num_bytes = 0u;
   ASSERT_EQ(MOJO_RESULT_OK,
             BeginReadData(&read_buffer, &num_bytes, false));
   ASSERT_EQ(0, memcmp(read_buffer, kExtraData, kExtraDataSize));
   EndReadData(num_bytes);
 
   ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe0));
@@ skipping 17 matching lines @@
   int32_t data = 123;
   uint32_t num_bytes = sizeof(data);
   ASSERT_EQ(MOJO_RESULT_OK, WriteData(&data, &num_bytes));
   ASSERT_EQ(MOJO_RESULT_OK, CloseProducer());
 
   // Now wait for the other side to become readable and to see the peer closed.
   MojoHandleSignalsState state;
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
                      MOJO_DEADLINE_INDEFINITE, &state));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             state.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             state.satisfiable_signals);
 
   // Now send the consumer over a MP so that it's serialized.
   MojoHandle pipe0, pipe1;
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoCreateMessagePipe(nullptr, &pipe0, &pipe1));
 
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWriteMessage(pipe0, nullptr, 0, &consumer_, 1,
                              MOJO_WRITE_MESSAGE_FLAG_NONE));
   consumer_ = MOJO_HANDLE_INVALID;
   ASSERT_EQ(MOJO_RESULT_OK, MojoWait(pipe1, MOJO_HANDLE_SIGNAL_READABLE,
                                      MOJO_DEADLINE_INDEFINITE, &state));
   uint32_t num_handles = 1;
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoReadMessage(pipe1, nullptr, 0, &consumer_, &num_handles,
                             MOJO_READ_MESSAGE_FLAG_NONE));
   ASSERT_EQ(num_handles, 1u);
 
   ASSERT_EQ(MOJO_RESULT_OK,
             MojoWait(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED,
                      MOJO_DEADLINE_INDEFINITE, &state));
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             state.satisfied_signals);
-  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
             state.satisfiable_signals);
 
   int32_t read_data;
   ASSERT_EQ(MOJO_RESULT_OK, ReadData(&read_data, &num_bytes));
   ASSERT_EQ(sizeof(read_data), num_bytes);
   ASSERT_EQ(data, read_data);
 
   ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe0));
   ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe1));
 }
@@ skipping 51 matching lines @@
       }
     } else {
       EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT, result);
     }
 
     MojoHandleSignalsState hss = MojoHandleSignalsState();
     EXPECT_EQ(MOJO_RESULT_OK, MojoWait(consumer, MOJO_HANDLE_SIGNAL_READABLE,
                                        MOJO_DEADLINE_INDEFINITE, &hss));
     // Peer could have become closed while we're still waiting for data.
     EXPECT_TRUE(MOJO_HANDLE_SIGNAL_READABLE & hss.satisfied_signals);
-    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
-              hss.satisfiable_signals);
+    EXPECT_TRUE(hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_READABLE);
+    EXPECT_TRUE(hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_PEER_CLOSED);
   }
 
   return num_bytes == 0;
 }
 
 #if !defined(OS_IOS)
 
 TEST_F(DataPipeTest, Multiprocess) {
   const uint32_t kTestDataSize =
       static_cast<uint32_t>(sizeof(kMultiprocessTestData));
@@ skipping 316 matching lines @@
     for (size_t i = 3; i < 6; ++i)
       CloseHandle(producers[i]);
   END_CHILD()
 }
 
 #endif  // !defined(OS_IOS)
 
 }  // namespace
 }  // namespace edk
 }  // namespace mojo