EDK: Add implementation of data pipe consumer read threshold stuff.

mojo/edk/system/remote_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 tests both |RemoteProducerDataPipeImpl| and
 // |RemoteConsumerDataPipeImpl|.
 
 #include <stdint.h>
 
 #include <memory>
@@ skipping 18 matching lines @@
 #include "testing/gtest/include/gtest/gtest.h"
 
 using mojo::platform::PlatformPipe;
 using mojo::util::MakeRefCounted;
 using mojo::util::RefPtr;
 
 namespace mojo {
 namespace system {
 namespace {
 
-const MojoHandleSignals kAllSignals = MOJO_HANDLE_SIGNAL_READABLE |
-                                      MOJO_HANDLE_SIGNAL_WRITABLE |
-                                      MOJO_HANDLE_SIGNAL_PEER_CLOSED;
-
 class RemoteDataPipeImplTest : public testing::Test {
  public:
   RemoteDataPipeImplTest()
       : platform_support_(embedder::CreateSimplePlatformSupport()),
         io_thread_(test::TestIOThread::StartMode::AUTO) {}
   ~RemoteDataPipeImplTest() override {}
 
   void SetUp() override {
     RefPtr<ChannelEndpoint> ep[2];
     message_pipes_[0] = MessagePipe::CreateLocalProxy(&ep[0]);
@@ skipping 95 matching lines @@
   EXPECT_EQ(MOJO_RESULT_OK,
             message_pipe(0)->WriteMessage(0, UserPointer<const void>(kHello),
                                           sizeof(kHello), nullptr,
                                           MOJO_WRITE_MESSAGE_FLAG_NONE));
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::ActionTimeout(), &context));
   EXPECT_EQ(123u, context);
   hss = HandleSignalsState();
   message_pipe(1)->RemoveAwakable(0, &waiter, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
             hss.satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
   EXPECT_EQ(MOJO_RESULT_OK, message_pipe(1)->ReadMessage(
                                 0, UserPointer<void>(read_buffer),
                                 MakeUserPointer(&read_buffer_size), nullptr,
                                 nullptr, MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(sizeof(kHello), static_cast<size_t>(read_buffer_size));
   EXPECT_STREQ(kHello, read_buffer);
 }
 
 TEST_F(RemoteDataPipeImplTest, SendConsumerWithClosedProducer) {
   char read_buffer[100] = {};
@@ skipping 41 matching lines @@
     // destroyed.
     EXPECT_TRUE(consumer->HasOneRef());
     consumer = nullptr;
   }
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::ActionTimeout(), &context));
   EXPECT_EQ(123u, context);
   hss = HandleSignalsState();
   message_pipe(1)->RemoveAwakable(0, &waiter, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
             hss.satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
   EXPECT_EQ(MOJO_RESULT_OK,
             message_pipe(1)->ReadMessage(
                 0, UserPointer<void>(read_buffer),
                 MakeUserPointer(&read_buffer_size), &read_dispatchers,
                 &read_num_dispatchers, MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(0u, static_cast<size_t>(read_buffer_size));
   EXPECT_EQ(1u, read_dispatchers.size());
   EXPECT_EQ(1u, read_num_dispatchers);
   ASSERT_TRUE(read_dispatchers[0]);
   EXPECT_TRUE(read_dispatchers[0]->HasOneRef());
@@ skipping 11 matching lines @@
   if (result == MOJO_RESULT_OK) {
     context = 0;
     EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::ActionTimeout(), &context));
     EXPECT_EQ(456u, context);
     consumer->RemoveAwakable(&waiter, &hss);
   } else {
     ASSERT_EQ(MOJO_RESULT_ALREADY_EXISTS, result);
   }
   // We don't know if the fact that the producer has been closed is known yet.
   EXPECT_TRUE((hss.satisfied_signals & MOJO_HANDLE_SIGNAL_READABLE));
+  EXPECT_TRUE((hss.satisfied_signals & MOJO_HANDLE_SIGNAL_READ_THRESHOLD));
   EXPECT_TRUE((hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_READABLE));
+  EXPECT_TRUE((hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_READ_THRESHOLD));
 
   // Read one element.
   elements[0] = -1;
   elements[1] = -1;
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK, consumer->ReadData(UserPointer<void>(elements),
                                                MakeUserPointer(&num_bytes),
                                                MOJO_READ_DATA_FLAG_NONE));
   EXPECT_EQ(1u * sizeof(elements[0]), num_bytes);
   EXPECT_EQ(123, elements[0]);
@@ skipping 62 matching lines @@
     // destroyed.
     EXPECT_TRUE(consumer->HasOneRef());
     consumer = nullptr;
   }
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::ActionTimeout(), &context));
   EXPECT_EQ(123u, context);
   hss = HandleSignalsState();
   message_pipe(1)->RemoveAwakable(0, &waiter, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
             hss.satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
   EXPECT_EQ(MOJO_RESULT_OK,
             message_pipe(1)->ReadMessage(
                 0, UserPointer<void>(read_buffer),
                 MakeUserPointer(&read_buffer_size), &read_dispatchers,
                 &read_num_dispatchers, MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(0u, static_cast<size_t>(read_buffer_size));
   EXPECT_EQ(1u, read_dispatchers.size());
   EXPECT_EQ(1u, read_num_dispatchers);
   ASSERT_TRUE(read_dispatchers[0]);
   EXPECT_TRUE(read_dispatchers[0]->HasOneRef());
@@ skipping 19 matching lines @@
   if (result == MOJO_RESULT_OK) {
     context = 0;
     EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::ActionTimeout(), &context));
     EXPECT_EQ(456u, context);
     consumer->RemoveAwakable(&waiter, &hss);
   } else {
     ASSERT_EQ(MOJO_RESULT_ALREADY_EXISTS, result);
   }
   // We don't know if the fact that the producer has been closed is known yet.
   EXPECT_TRUE((hss.satisfied_signals & MOJO_HANDLE_SIGNAL_READABLE));
+  EXPECT_TRUE((hss.satisfied_signals & MOJO_HANDLE_SIGNAL_READ_THRESHOLD));
   EXPECT_TRUE((hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_READABLE));
+  EXPECT_TRUE((hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_READ_THRESHOLD));
 
   // Read one element.
   int32_t elements[10] = {-1, -1};
   num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
   EXPECT_EQ(MOJO_RESULT_OK, consumer->ReadData(UserPointer<void>(elements),
                                                MakeUserPointer(&num_bytes),
                                                MOJO_READ_DATA_FLAG_NONE));
   EXPECT_EQ(1u * sizeof(elements[0]), num_bytes);
   EXPECT_EQ(123456, elements[0]);
   EXPECT_EQ(-1, elements[1]);
@@ skipping 58 matching lines @@
     // destroyed.
     EXPECT_TRUE(consumer->HasOneRef());
     consumer = nullptr;
   }
   EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::ActionTimeout(), &context));
   EXPECT_EQ(123u, context);
   hss = HandleSignalsState();
   message_pipe(1)->RemoveAwakable(0, &waiter, &hss);
   EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
             hss.satisfied_signals);
-  EXPECT_EQ(kAllSignals, hss.satisfiable_signals);
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE |
+                MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
   EXPECT_EQ(MOJO_RESULT_OK,
             message_pipe(1)->ReadMessage(
                 0, UserPointer<void>(read_buffer),
                 MakeUserPointer(&read_buffer_size), &read_dispatchers,
                 &read_num_dispatchers, MOJO_READ_MESSAGE_FLAG_NONE));
   EXPECT_EQ(0u, static_cast<size_t>(read_buffer_size));
   EXPECT_EQ(1u, read_dispatchers.size());
   EXPECT_EQ(1u, read_num_dispatchers);
   ASSERT_TRUE(read_dispatchers[0]);
   EXPECT_TRUE(read_dispatchers[0]->HasOneRef());
@@ skipping 17 matching lines @@
   MojoResult result =
       consumer->AddAwakable(&waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED, 456, &hss);
   if (result == MOJO_RESULT_OK) {
     context = 0;
     EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::ActionTimeout(), &context));
     EXPECT_EQ(456u, context);
     consumer->RemoveAwakable(&waiter, &hss);
   } else {
     ASSERT_EQ(MOJO_RESULT_ALREADY_EXISTS, result);
   }
-  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+  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,
+  EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
+                MOJO_HANDLE_SIGNAL_READ_THRESHOLD,
             hss.satisfiable_signals);
 
   // Read some elements.
   int32_t elements[10] = {};
   num_bytes = static_cast<uint32_t>(sizeof(elements));
   EXPECT_EQ(MOJO_RESULT_OK, consumer->ReadData(UserPointer<void>(elements),
                                                MakeUserPointer(&num_bytes),
                                                MOJO_READ_DATA_FLAG_NONE));
   EXPECT_EQ(2u * sizeof(elements[0]), num_bytes);
   EXPECT_EQ(123456, elements[0]);
   EXPECT_EQ(789012, elements[1]);
   EXPECT_EQ(0, elements[2]);
 
   consumer->Close();
 }
 
 }  // namespace
 }  // namespace system
 }  // namespace mojo