Adding MOJO_HANDLE_SIGNAL_PEER_CLOSED to be notified when a peer is closed.

mojo/edk/system/local_data_pipe_unittest.cc

 // Copyright 2013 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "mojo/edk/system/local_data_pipe.h"
 
 #include <string.h>
 
 #include "base/macros.h"
 #include "base/memory/ref_counted.h"
@@ skipping 221 matching lines @@
   uint32_t context = 0;
   HandleSignalsState hss;
 
   // Never readable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_FAILED_PRECONDITION,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 12, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // Already writable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 34, &hss));
 
   // 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,
             dp->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.
   waiter.Init();
   ASSERT_EQ(
       MOJO_RESULT_OK,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 56, nullptr));
   // And it shouldn't be writable yet.
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
   hss = HandleSignalsState();
   dp->ProducerRemoveWaiter(&waiter, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            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,
             dp->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.
   waiter.Init();
   ASSERT_EQ(
       MOJO_RESULT_OK,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 56, nullptr));
   // And it still shouldn't be writable yet.
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
   hss = HandleSignalsState();
   dp->ProducerRemoveWaiter(&waiter, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // Do it again.
   waiter.Init();
   ASSERT_EQ(
       MOJO_RESULT_OK,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 78, nullptr));
 
   // Read one element.
   elements[0] = -1;
   elements[1] = -1;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK,
             dp->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.
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(1000, &context));
   EXPECT_EQ(78u, context);
   hss = HandleSignalsState();
   dp->ProducerRemoveWaiter(&waiter, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            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,
             dp->ProducerBeginWriteData(MakeUserPointer(&buffer),
                                        MakeUserPointer(&num_bytes), false));
   EXPECT_TRUE(buffer);
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
 
@@ skipping 22 matching lines @@
   EXPECT_EQ(
       MOJO_RESULT_OK,
       dp->ConsumerEndReadData(static_cast<uint32_t>(1u * sizeof(elements[0]))));
 
   // Waiting should succeed.
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(1000, &context));
   EXPECT_EQ(90u, context);
   hss = HandleSignalsState();
   dp->ProducerRemoveWaiter(&waiter, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // Write one element.
   elements[0] = 123;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK,
             dp->ProducerWriteData(UserPointer<const void>(elements),
                                   MakeUserPointer(&num_bytes), false));
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
 
   // Add a waiter.
   waiter.Init();
   ASSERT_EQ(
       MOJO_RESULT_OK,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 12, nullptr));
 
   // Close the consumer.
   dp->ConsumerClose();
 
   // It should now be never-writable.
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, waiter.Wait(1000, &context));
   EXPECT_EQ(12u, context);
   hss = HandleSignalsState();
   dp->ProducerRemoveWaiter(&waiter, &hss);
-  EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(0u, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
   dp->ProducerClose();
 }
 
+TEST(LocalDataPipeTest, PeerClosedWaiting) {
+  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|.
+  };
+  MojoCreateDataPipeOptions validated_options = {0};
+  EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
+                                MakeUserPointer(&options), &validated_options));
+
+  Waiter waiter;
+  HandleSignalsState hss;
+
+  // Check MOJO_HANDLE_SIGNAL_PEER_CLOSED on producer.
+  {
+    scoped_refptr<LocalDataPipe> dp(new LocalDataPipe(validated_options));
+    // Add a waiter.
+    waiter.Init();
+    ASSERT_EQ(MOJO_RESULT_OK,
+              dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED, 12,
+                                    nullptr));
+
+    // Close the consumer.
+    dp->ConsumerClose();
+
+    // It should be signaled.
+    uint32_t context = 0;
+    EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(1000, &context));
+    EXPECT_EQ(12u, context);
+    hss = HandleSignalsState();
+    dp->ProducerRemoveWaiter(&waiter, &hss);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
+
+    dp->ProducerClose();
+  }
+
+  // Check MOJO_HANDLE_SIGNAL_PEER_CLOSED on consumer.
+  {
+    scoped_refptr<LocalDataPipe> dp(new LocalDataPipe(validated_options));
+    // Add a waiter.
+    waiter.Init();
+    ASSERT_EQ(MOJO_RESULT_OK,
+              dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED, 12,
+                                    nullptr));
+
+    // Close the producer.
+    dp->ProducerClose();
+
+    // It should be signaled.
+    uint32_t context = 0;
+    EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(1000, &context));
+    EXPECT_EQ(12u, context);
+    hss = HandleSignalsState();
+    dp->ConsumerRemoveWaiter(&waiter, &hss);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
+
+    dp->ConsumerClose();
+  }
+}
+
 TEST(LocalDataPipeTest, 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|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
   {
     scoped_refptr<LocalDataPipe> dp(new LocalDataPipe(validated_options));
     Waiter waiter;
     uint32_t context = 0;
     HandleSignalsState hss;
 
     // Never writable.
     waiter.Init();
     hss = HandleSignalsState();
     EXPECT_EQ(
         MOJO_RESULT_FAILED_PRECONDITION,
         dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 12, &hss));
     EXPECT_EQ(0u, hss.satisfied_signals);
-    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+              hss.satisfiable_signals);
 
     // Not yet readable.
     waiter.Init();
     ASSERT_EQ(MOJO_RESULT_OK,
               dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 34,
                                     nullptr));
     EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
     hss = HandleSignalsState();
     dp->ConsumerRemoveWaiter(&waiter, &hss);
     EXPECT_EQ(0u, hss.satisfied_signals);
-    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+              hss.satisfiable_signals);
 
     // 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,
               dp->ProducerWriteData(UserPointer<const void>(elements),
                                     MakeUserPointer(&num_bytes), true));
 
     // Should already be readable.
     waiter.Init();
     hss = HandleSignalsState();
     EXPECT_EQ(
         MOJO_RESULT_ALREADY_EXISTS,
         dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 56, &hss));
     EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+              hss.satisfiable_signals);
 
     // Discard one element.
     num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
     EXPECT_EQ(MOJO_RESULT_OK,
               dp->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,
         dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 78, &hss));
     EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+              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,
               dp->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,
         dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 78, &hss));
     EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+              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,
               dp->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]);
@@ skipping 12 matching lines @@
     EXPECT_EQ(MOJO_RESULT_OK,
               dp->ProducerWriteData(UserPointer<const void>(elements),
                                     MakeUserPointer(&num_bytes), true));
 
     // Waiting should now succeed.
     EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(1000, &context));
     EXPECT_EQ(90u, context);
     hss = HandleSignalsState();
     dp->ConsumerRemoveWaiter(&waiter, &hss);
     EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+              hss.satisfiable_signals);
 
     // Close the producer.
     dp->ProducerClose();
 
     // Should still be readable.
     waiter.Init();
     hss = HandleSignalsState();
     EXPECT_EQ(
         MOJO_RESULT_ALREADY_EXISTS,
         dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 12, &hss));
-    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+    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);
 
     // Read one element.
     elements[0] = -1;
     elements[1] = -1;
     num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
     EXPECT_EQ(MOJO_RESULT_OK,
               dp->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,
         dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 34, &hss));
-    EXPECT_EQ(0u, hss.satisfied_signals);
-    EXPECT_EQ(0u, hss.satisfiable_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
     dp->ConsumerClose();
   }
 
   // Test with two-phase APIs and closing the producer with an active consumer
   // waiter.
   {
     scoped_refptr<LocalDataPipe> dp(new LocalDataPipe(validated_options));
     Waiter waiter;
     uint32_t context = 0;
@@ skipping 15 matching lines @@
     EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerEndWriteData(static_cast<uint32_t>(
                                   2u * sizeof(elements[0]))));
 
     // Should already be readable.
     waiter.Init();
     hss = HandleSignalsState();
     EXPECT_EQ(
         MOJO_RESULT_ALREADY_EXISTS,
         dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 12, &hss));
     EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+              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]));
     EXPECT_EQ(MOJO_RESULT_OK,
               dp->ConsumerBeginReadData(MakeUserPointer(&read_buffer),
                                         MakeUserPointer(&num_bytes), true));
     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, dp->ConsumerEndReadData(static_cast<uint32_t>(
                                   1u * sizeof(elements[0]))));
 
     // Should still be readable.
     waiter.Init();
     hss = HandleSignalsState();
     EXPECT_EQ(
         MOJO_RESULT_ALREADY_EXISTS,
         dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 34, &hss));
     EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+              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,
               dp->ConsumerBeginReadData(MakeUserPointer(&read_buffer),
                                         MakeUserPointer(&num_bytes), false));
     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, dp->ConsumerEndReadData(static_cast<uint32_t>(
                                   1u * sizeof(elements[0]))));
 
     // Adding a waiter should now succeed.
     waiter.Init();
     ASSERT_EQ(MOJO_RESULT_OK,
               dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 56,
                                     nullptr));
 
     // Close the producer.
     dp->ProducerClose();
 
     // Should be never-readable.
     EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, waiter.Wait(1000, &context));
     EXPECT_EQ(56u, context);
     hss = HandleSignalsState();
     dp->ConsumerRemoveWaiter(&waiter, &hss);
-    EXPECT_EQ(0u, hss.satisfied_signals);
-    EXPECT_EQ(0u, hss.satisfiable_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
+    EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 
     dp->ConsumerClose();
   }
 }
 
 // Tests that data pipes aren't writable/readable during two-phase writes/reads.
 TEST(LocalDataPipeTest, 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|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
   scoped_refptr<LocalDataPipe> dp(new LocalDataPipe(validated_options));
   Waiter waiter;
   HandleSignalsState hss;
 
   // It should be writable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 0, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   uint32_t num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t));
   void* write_ptr = nullptr;
   EXPECT_EQ(MOJO_RESULT_OK,
             dp->ProducerBeginWriteData(MakeUserPointer(&write_ptr),
                                        MakeUserPointer(&num_bytes), false));
   EXPECT_TRUE(write_ptr);
   EXPECT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(int32_t)));
 
   // At this point, it shouldn't be writable.
   waiter.Init();
   ASSERT_EQ(
       MOJO_RESULT_OK,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 1, nullptr));
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
   hss = HandleSignalsState();
   dp->ProducerRemoveWaiter(&waiter, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // It shouldn't be readable yet either.
   waiter.Init();
   ASSERT_EQ(
       MOJO_RESULT_OK,
       dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 2, nullptr));
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
   hss = HandleSignalsState();
   dp->ConsumerRemoveWaiter(&waiter, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   static_cast<int32_t*>(write_ptr)[0] = 123;
   EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerEndWriteData(
                                 static_cast<uint32_t>(1u * sizeof(int32_t))));
 
   // It should be writable again.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 3, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // And readable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 4, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            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;
   EXPECT_EQ(MOJO_RESULT_OK,
             dp->ProducerBeginWriteData(MakeUserPointer(&write_ptr),
                                        MakeUserPointer(&num_bytes), false));
   EXPECT_TRUE(write_ptr);
   EXPECT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(int32_t)));
 
   // It should be readable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 5, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // End the two-phase write without writing anything.
   EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerEndWriteData(0u));
 
   // Start a two-phase read.
   num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t));
   const void* read_ptr = nullptr;
   EXPECT_EQ(MOJO_RESULT_OK,
             dp->ConsumerBeginReadData(MakeUserPointer(&read_ptr),
                                       MakeUserPointer(&num_bytes), false));
   EXPECT_TRUE(read_ptr);
   EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(int32_t)), num_bytes);
 
   // At this point, it should still be writable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 6, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // But not readable.
   waiter.Init();
   ASSERT_EQ(
       MOJO_RESULT_OK,
       dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 7, nullptr));
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
   hss = HandleSignalsState();
   dp->ConsumerRemoveWaiter(&waiter, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // End the two-phase read without reading anything.
   EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerEndReadData(0u));
 
   // It should be readable again.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 8, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   dp->ProducerClose();
   dp->ConsumerClose();
 }
 
 // Test that a "may discard" data pipe is writable even when it's full.
 TEST(LocalDataPipeTest, BasicMayDiscardWaiting) {
   const MojoCreateDataPipeOptions options = {
       kSizeOfOptions,                                  // |struct_size|.
       MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD,  // |flags|.
       static_cast<uint32_t>(sizeof(int32_t)),          // |element_num_bytes|.
       1 * sizeof(int32_t)                              // |capacity_num_bytes|.
   };
   MojoCreateDataPipeOptions validated_options = {0};
   EXPECT_EQ(MOJO_RESULT_OK, DataPipe::ValidateCreateOptions(
                                 MakeUserPointer(&options), &validated_options));
 
   scoped_refptr<LocalDataPipe> dp(new LocalDataPipe(validated_options));
   Waiter waiter;
   HandleSignalsState hss;
 
   // Writable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 0, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // Not readable.
   waiter.Init();
   ASSERT_EQ(
       MOJO_RESULT_OK,
       dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 1, nullptr));
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
   hss = HandleSignalsState();
   dp->ConsumerRemoveWaiter(&waiter, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   uint32_t num_bytes = static_cast<uint32_t>(sizeof(int32_t));
   int32_t element = 123;
   EXPECT_EQ(MOJO_RESULT_OK,
             dp->ProducerWriteData(UserPointer<const void>(&element),
                                   MakeUserPointer(&num_bytes), false));
   EXPECT_EQ(static_cast<uint32_t>(sizeof(int32_t)), num_bytes);
 
   // Still writable (even though it's full).
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 2, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // Now readable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 3, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // Overwrite that element.
   num_bytes = static_cast<uint32_t>(sizeof(int32_t));
   element = 456;
   EXPECT_EQ(MOJO_RESULT_OK,
             dp->ProducerWriteData(UserPointer<const void>(&element),
                                   MakeUserPointer(&num_bytes), false));
   EXPECT_EQ(static_cast<uint32_t>(sizeof(int32_t)), num_bytes);
 
   // Still writable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 4, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // And still readable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 5, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // Read that element.
   num_bytes = static_cast<uint32_t>(sizeof(int32_t));
   element = 0;
   EXPECT_EQ(MOJO_RESULT_OK,
             dp->ConsumerReadData(UserPointer<void>(&element),
                                  MakeUserPointer(&num_bytes), false, false));
   EXPECT_EQ(static_cast<uint32_t>(sizeof(int32_t)), num_bytes);
   EXPECT_EQ(456, element);
 
   // Still writable.
   waiter.Init();
   hss = HandleSignalsState();
   EXPECT_EQ(
       MOJO_RESULT_ALREADY_EXISTS,
       dp->ProducerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 6, &hss));
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   // No longer readable.
   waiter.Init();
   ASSERT_EQ(
       MOJO_RESULT_OK,
       dp->ConsumerAddWaiter(&waiter, MOJO_HANDLE_SIGNAL_READABLE, 7, nullptr));
   EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0, nullptr));
   hss = HandleSignalsState();
   dp->ConsumerRemoveWaiter(&waiter, &hss);
   EXPECT_EQ(0u, hss.satisfied_signals);
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
 
   dp->ProducerClose();
   dp->ConsumerClose();
 }
 
 void Seq(int32_t start, size_t count, int32_t* out) {
   for (size_t i = 0; i < count; i++)
     out[i] = start + static_cast<int32_t>(i);
 }
 
@@ skipping 1061 matching lines @@
   // End reading.
   EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerEndReadData(2u));
 
   dp->ProducerClose();
   dp->ConsumerClose();
 }
 
 }  // namespace
 }  // namespace system
 }  // namespace mojo