Mojo: Armed Watchers

mojo/edk/system/data_pipe_unittest.cc

Index: mojo/edk/system/data_pipe_unittest.cc
diff --git a/mojo/edk/system/data_pipe_unittest.cc b/mojo/edk/system/data_pipe_unittest.cc
index f444621355c527a449e5867605a755d3531d9660..ed228334effdf104950b583676cfe364c5c329cc 100644
--- a/mojo/edk/system/data_pipe_unittest.cc
+++ b/mojo/edk/system/data_pipe_unittest.cc
@@ -21,7 +21,7 @@
 #include "mojo/public/c/system/data_pipe.h"
 #include "mojo/public/c/system/functions.h"
 #include "mojo/public/c/system/message_pipe.h"
-#include "mojo/public/cpp/system/watcher.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace mojo {
@@ -591,6 +591,166 @@ TEST_F(DataPipeTest, BasicConsumerWaiting) {
   ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
 }
 
+TEST_F(DataPipeTest, ConsumerMinimumAvailabilitySignaling) {
+  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.
+  EXPECT_EQ(MOJO_RESULT_OK, MojoWait(consumer_, MOJO_HANDLE_SIGNAL_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 |
+                             MOJO_READ_DATA_FLAG_CLEAR_SIGNAL));
+
+  // The consumer should no longer appear to be readable.
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED,
+            MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE, 0, nullptr));
+
+  // Write two more elements.
+  EXPECT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
+
+  // The consumer handle should once again appear to be readable.
+  EXPECT_EQ(MOJO_RESULT_OK, MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
+                                     MOJO_DEADLINE_INDEFINITE, nullptr));
+
+  // Try again to read a minimum of 6 elements.
+  EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE,
+            MojoReadData(consumer_, read_elements, &num_read_bytes,
+                         MOJO_READ_DATA_FLAG_ALL_OR_NONE |
+                             MOJO_READ_DATA_FLAG_CLEAR_SIGNAL));
+
+  // Write two final elements.
+  EXPECT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
+
+  // Read should succeed this time.
+  EXPECT_EQ(MOJO_RESULT_OK,
+            MojoReadData(consumer_, read_elements, &num_read_bytes,
+                         MOJO_READ_DATA_FLAG_ALL_OR_NONE |
+                             MOJO_READ_DATA_FLAG_CLEAR_SIGNAL));
+
+  // And once again the consumer is unreadable.
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED,
+            MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE, 0, nullptr));
+
+  // Write two elements.
+  EXPECT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
+  EXPECT_EQ(MOJO_RESULT_OK, MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
+                                     MOJO_DEADLINE_INDEFINITE, nullptr));
+
+  // Reading without ALL_OR_NONE now should still be able to succeed. i.e., the
+  // previous usage should have no lasting effect beyond a successful read.
+  EXPECT_EQ(MOJO_RESULT_OK,
+            MojoReadData(consumer_, read_elements, &num_read_bytes,
+                         MOJO_READ_DATA_FLAG_NONE));
+}
+
+TEST_F(DataPipeTest, ProducerMinimumCapacitySignaling) {
+  const int32_t kTestElements[6] = {123, 456, 768, 101, 121, 314};
+  const uint32_t kTestPipeSize = sizeof(kTestElements);
+
+  const MojoCreateDataPipeOptions options = {
+      kSizeOfOptions,                           // |struct_size|.
+      MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
+      static_cast<uint32_t>(sizeof(int32_t)),   // |element_num_bytes|.
+      kTestPipeSize                             // |capacity_num_bytes|.
+  };
+  ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
+
+  // Start by filling the pipe.
+  uint32_t num_bytes_written = kTestPipeSize;
+  EXPECT_EQ(MOJO_RESULT_OK, WriteData(kTestElements, &num_bytes_written, true));
+
+  // The consumer handle should appear to be readable.
+  EXPECT_EQ(MOJO_RESULT_OK, MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
+                                     MOJO_DEADLINE_INDEFINITE, nullptr));
+
+  // The producer handle should appear to be unwritable because the pipe is
+  // filled to capacity.
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED,
+            MojoWait(producer_, MOJO_HANDLE_SIGNAL_WRITABLE, 0, nullptr));
+
+  // Make some room.
+  int32_t read_elements[2];
+  uint32_t num_bytes_read = sizeof(read_elements);
+  EXPECT_EQ(MOJO_RESULT_OK, ReadData(read_elements, &num_bytes_read, true));
+  EXPECT_EQ(sizeof(read_elements), num_bytes_read);
+
+  // The producer handle should appear to be writable now.
+  EXPECT_EQ(MOJO_RESULT_OK, MojoWait(producer_, MOJO_HANDLE_SIGNAL_WRITABLE,
+                                     MOJO_DEADLINE_INDEFINITE, nullptr));
+
+  // Now attempt to write to the pipe, clearing signals on insufficient space.
+  num_bytes_written = kTestPipeSize;
+  EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE,
+            MojoWriteData(producer_, kTestElements, &num_bytes_written,
+                          MOJO_WRITE_DATA_FLAG_ALL_OR_NONE |
+                              MOJO_WRITE_DATA_FLAG_CLEAR_SIGNAL));
+
+  // The producer should no longer appear to be writable even though no new data
+  // was written above.
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED,
+            MojoWait(producer_, MOJO_HANDLE_SIGNAL_WRITABLE, 0, nullptr));
+
+  // Make more room.
+  EXPECT_EQ(MOJO_RESULT_OK, ReadData(read_elements, &num_bytes_read, true));
+  EXPECT_EQ(sizeof(read_elements), num_bytes_read);
+
+  // The producer handle should appear to be writable again...
+  EXPECT_EQ(MOJO_RESULT_OK, MojoWait(producer_, MOJO_HANDLE_SIGNAL_WRITABLE,
+                                     MOJO_DEADLINE_INDEFINITE, nullptr));
+
+  // ..but there's still not enough capacity.
+  num_bytes_written = kTestPipeSize;
+  EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE,
+            MojoWriteData(producer_, kTestElements, &num_bytes_written,
+                          MOJO_WRITE_DATA_FLAG_ALL_OR_NONE |
+                              MOJO_WRITE_DATA_FLAG_CLEAR_SIGNAL));
+
+  // Once again not writable.
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED,
+            MojoWait(producer_, MOJO_HANDLE_SIGNAL_WRITABLE, 0, nullptr));
+
+  // Finally make enough room for the write to succeed.
+  EXPECT_EQ(MOJO_RESULT_OK, ReadData(read_elements, &num_bytes_read, true));
+  EXPECT_EQ(sizeof(read_elements), num_bytes_read);
+  EXPECT_EQ(MOJO_RESULT_OK, MojoWait(producer_, MOJO_HANDLE_SIGNAL_WRITABLE,
+                                     MOJO_DEADLINE_INDEFINITE, nullptr));
+
+  // And now the write should succeed.
+  num_bytes_written = kTestPipeSize;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            MojoWriteData(producer_, kTestElements, &num_bytes_written,
+                          MOJO_WRITE_DATA_FLAG_ALL_OR_NONE |
+                              MOJO_WRITE_DATA_FLAG_CLEAR_SIGNAL));
+
+  // Read two elements.
+  num_bytes_read = sizeof(read_elements);
+  EXPECT_EQ(MOJO_RESULT_OK, ReadData(read_elements, &num_bytes_read, true));
+  EXPECT_EQ(MOJO_RESULT_OK, MojoWait(producer_, MOJO_HANDLE_SIGNAL_WRITABLE,
+                                     MOJO_DEADLINE_INDEFINITE, nullptr));
+
+  // Writing without ALL_OR_NONE now should still be able to succeed, i.e. the
+  // previous usage should have no lasting effect beyond a successful write.
+  num_bytes_written = kTestPipeSize;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            MojoWriteData(producer_, kTestElements, &num_bytes_written,
+                          MOJO_READ_DATA_FLAG_NONE));
+}
+
 // Test with two-phase APIs and also closing the producer with an active
 // consumer waiter.
 TEST_F(DataPipeTest, ConsumerWaitingTwoPhase) {
@@ -608,7 +768,7 @@ TEST_F(DataPipeTest, ConsumerWaitingTwoPhase) {
   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);
@@ -1921,7 +2081,7 @@ DEFINE_TEST_CLIENT_TEST_WITH_PIPE(DataPipeStatusChangeInTransitClient,
 
   base::MessageLoop message_loop;
 
-  // Wait on producer 1 and consumer 1 using Watchers.
+  // Wait on producer 1 and consumer 1 using SimpleWatchers.
   {
     base::RunLoop run_loop;
     int count = 0;
@@ -1932,12 +2092,16 @@ DEFINE_TEST_CLIENT_TEST_WITH_PIPE(DataPipeStatusChangeInTransitClient,
             loop->Quit();
         },
         &run_loop, &count);
-    Watcher producer_watcher(FROM_HERE), consumer_watcher(FROM_HERE);
-    producer_watcher.Start(
-        Handle(producers[1]), MOJO_HANDLE_SIGNAL_PEER_CLOSED, callback);
-    consumer_watcher.Start(
-        Handle(consumers[1]), MOJO_HANDLE_SIGNAL_PEER_CLOSED, callback);
+    SimpleWatcher producer_watcher(FROM_HERE,
+                                   SimpleWatcher::ArmingPolicy::AUTOMATIC);
+    SimpleWatcher consumer_watcher(FROM_HERE,
+                                   SimpleWatcher::ArmingPolicy::AUTOMATIC);
+    producer_watcher.Watch(Handle(producers[1]), MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+                           callback);
+    consumer_watcher.Watch(Handle(consumers[1]), MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+                           callback);
     run_loop.Run();
+    EXPECT_EQ(2, count);
   }
 
   // Wait on producer 2 by polling with MojoWriteData.