Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(50)

Unified Diff: third_party/mojo/src/mojo/edk/system/data_pipe_impl_unittest.cc

Issue 975973002: Update mojo sdk to rev f68e697e389943cd9bf9652397312280e96b127a (Closed) Base URL: https://chromium.googlesource.com/chromium/src.git@master
Patch Set: shake fist at msvc Created 5 years, 10 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
Index: third_party/mojo/src/mojo/edk/system/data_pipe_impl_unittest.cc
diff --git a/third_party/mojo/src/mojo/edk/system/data_pipe_impl_unittest.cc b/third_party/mojo/src/mojo/edk/system/data_pipe_impl_unittest.cc
new file mode 100644
index 0000000000000000000000000000000000000000..b14d3af4c4c7d2c78c3873b3bb6f082bc7e468e7
--- /dev/null
+++ b/third_party/mojo/src/mojo/edk/system/data_pipe_impl_unittest.cc
@@ -0,0 +1,1720 @@
+// 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>
+
+#include "base/bind.h"
+#include "base/location.h"
+#include "base/logging.h"
+#include "base/macros.h"
+#include "base/message_loop/message_loop.h"
+#include "base/test/test_io_thread.h"
+#include "base/threading/platform_thread.h" // For |Sleep()|.
+#include "mojo/edk/embedder/platform_channel_pair.h"
+#include "mojo/edk/embedder/simple_platform_support.h"
+#include "mojo/edk/system/channel.h"
+#include "mojo/edk/system/channel_endpoint.h"
+#include "mojo/edk/system/data_pipe.h"
+#include "mojo/edk/system/data_pipe_consumer_dispatcher.h"
+#include "mojo/edk/system/data_pipe_producer_dispatcher.h"
+#include "mojo/edk/system/memory.h"
+#include "mojo/edk/system/message_pipe.h"
+#include "mojo/edk/system/raw_channel.h"
+#include "mojo/edk/system/test_utils.h"
+#include "mojo/edk/system/waiter.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace mojo {
+namespace system {
+namespace {
+
+const MojoHandleSignals kAllSignals = MOJO_HANDLE_SIGNAL_READABLE |
+ MOJO_HANDLE_SIGNAL_WRITABLE |
+ MOJO_HANDLE_SIGNAL_PEER_CLOSED;
+const uint32_t kSizeOfOptions =
+ static_cast<uint32_t>(sizeof(MojoCreateDataPipeOptions));
+
+// DataPipeImplTestHelper ------------------------------------------------------
+
+class DataPipeImplTestHelper {
+ public:
+ virtual ~DataPipeImplTestHelper() {}
+
+ virtual void SetUp() = 0;
+ virtual void TearDown() = 0;
+
+ virtual void Create(const MojoCreateDataPipeOptions& validated_options) = 0;
+
+ // Possibly transfers the producer/consumer.
+ virtual void DoTransfer() = 0;
+
+ // Returns the |DataPipe| object for the producer and consumer, respectively.
+ virtual DataPipe* dpp() = 0;
+ virtual DataPipe* dpc() = 0;
+
+ virtual void ProducerClose() = 0;
+ virtual void ConsumerClose() = 0;
+
+ protected:
+ DataPipeImplTestHelper() {}
+
+ private:
+ DISALLOW_COPY_AND_ASSIGN(DataPipeImplTestHelper);
+};
+
+// DataPipeImplTest ------------------------------------------------------------
+
+template <class Helper>
+class DataPipeImplTest : public testing::Test {
+ public:
+ DataPipeImplTest() {}
+ ~DataPipeImplTest() override {}
+
+ void SetUp() override { helper_.SetUp(); }
+ void TearDown() override { helper_.TearDown(); }
+
+ protected:
+ void Create(const MojoCreateDataPipeOptions& options) {
+ MojoCreateDataPipeOptions validated_options = {};
+ ASSERT_EQ(MOJO_RESULT_OK,
+ DataPipe::ValidateCreateOptions(MakeUserPointer(&options),
+ &validated_options));
+ helper_.Create(validated_options);
+ }
+
+ void DoTransfer() { return helper_.DoTransfer(); }
+
+ DataPipe* dpp() { return helper_.dpp(); }
+ DataPipe* dpc() { return helper_.dpc(); }
+
+ void ProducerClose() { helper_.ProducerClose(); }
+ void ConsumerClose() { helper_.ConsumerClose(); }
+
+ private:
+ Helper helper_;
+
+ DISALLOW_COPY_AND_ASSIGN(DataPipeImplTest);
+};
+
+// LocalDataPipeImplTestHelper -------------------------------------------------
+
+class LocalDataPipeImplTestHelper : public DataPipeImplTestHelper {
+ public:
+ LocalDataPipeImplTestHelper() {}
+ ~LocalDataPipeImplTestHelper() override {}
+
+ void SetUp() override {}
+ void TearDown() override {}
+
+ void Create(const MojoCreateDataPipeOptions& validated_options) override {
+ CHECK(!dp_);
+ dp_ = DataPipe::CreateLocal(validated_options);
+ }
+
+ void DoTransfer() override {}
+
+ // Returns the |DataPipe| object for the producer and consumer, respectively.
+ DataPipe* dpp() override { return dp_.get(); }
+ DataPipe* dpc() override { return dp_.get(); }
+
+ void ProducerClose() override { dp_->ProducerClose(); }
+ void ConsumerClose() override { dp_->ConsumerClose(); }
+
+ private:
+ scoped_refptr<DataPipe> dp_;
+
+ DISALLOW_COPY_AND_ASSIGN(LocalDataPipeImplTestHelper);
+};
+
+// RemoteDataPipeImplTestHelper ------------------------------------------------
+
+// Base class for |Remote{Producer,Consumer}DataPipeImplTestHelper|.
+class RemoteDataPipeImplTestHelper : public DataPipeImplTestHelper {
+ public:
+ RemoteDataPipeImplTestHelper() : io_thread_(base::TestIOThread::kAutoStart) {}
+ ~RemoteDataPipeImplTestHelper() override {}
+
+ void SetUp() override {
+ scoped_refptr<ChannelEndpoint> ep[2];
+ message_pipes_[0] = MessagePipe::CreateLocalProxy(&ep[0]);
+ message_pipes_[1] = MessagePipe::CreateLocalProxy(&ep[1]);
+
+ io_thread_.PostTaskAndWait(
+ FROM_HERE, base::Bind(&RemoteDataPipeImplTestHelper::SetUpOnIOThread,
+ base::Unretained(this), ep[0], ep[1]));
+ }
+
+ void TearDown() override {
+ EnsureMessagePipeClosed(0);
+ EnsureMessagePipeClosed(1);
+ io_thread_.PostTaskAndWait(
+ FROM_HERE, base::Bind(&RemoteDataPipeImplTestHelper::TearDownOnIOThread,
+ base::Unretained(this)));
+ }
+
+ void Create(const MojoCreateDataPipeOptions& validated_options) override {
+ CHECK(!dp_);
+ dp_ = DataPipe::CreateLocal(validated_options);
+ }
+
+ protected:
+ void SendDispatcher(size_t source_i,
+ scoped_refptr<Dispatcher> to_send,
+ scoped_refptr<Dispatcher>* 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, 987, nullptr));
+ {
+ DispatcherTransport transport(
+ test::DispatcherTryStartTransport(to_send.get()));
+ ASSERT_TRUE(transport.is_valid());
+
+ std::vector<DispatcherTransport> 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();
+ }
+ uint32_t context = 0;
+ ASSERT_EQ(MOJO_RESULT_OK, waiter.Wait(test::ActionDeadline(), &context));
+ EXPECT_EQ(987u, context);
+ HandleSignalsState hss = HandleSignalsState();
+ message_pipe(dest_i)->RemoveAwakable(0, &waiter, &hss);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE,
+ hss.satisfied_signals);
+ EXPECT_EQ(kAllSignals, hss.satisfiable_signals);
+ char read_buffer[100] = {};
+ uint32_t read_buffer_size = static_cast<uint32_t>(sizeof(read_buffer));
+ DispatcherVector read_dispatchers;
+ uint32_t read_num_dispatchers = 10; // Maximum to get.
+ ASSERT_EQ(MOJO_RESULT_OK,
+ message_pipe(dest_i)->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));
+ ASSERT_EQ(1u, read_dispatchers.size());
+ ASSERT_EQ(1u, read_num_dispatchers);
+ ASSERT_TRUE(read_dispatchers[0]);
+ EXPECT_TRUE(read_dispatchers[0]->HasOneRef());
+
+ *to_receive = read_dispatchers[0];
+ }
+
+ scoped_refptr<MessagePipe> message_pipe(size_t i) {
+ return message_pipes_[i];
+ }
+ scoped_refptr<DataPipe> dp() { return dp_; }
+
+ private:
+ void EnsureMessagePipeClosed(size_t i) {
+ if (!message_pipes_[i])
+ return;
+ message_pipes_[i]->Close(0);
+ message_pipes_[i] = nullptr;
+ }
+
+ void SetUpOnIOThread(scoped_refptr<ChannelEndpoint> ep0,
+ scoped_refptr<ChannelEndpoint> ep1) {
+ CHECK_EQ(base::MessageLoop::current(), io_thread_.message_loop());
+
+ embedder::PlatformChannelPair channel_pair;
+ channels_[0] = new Channel(&platform_support_);
+ channels_[0]->Init(RawChannel::Create(channel_pair.PassServerHandle()));
+ channels_[0]->SetBootstrapEndpoint(ep0);
+ channels_[1] = new Channel(&platform_support_);
+ channels_[1]->Init(RawChannel::Create(channel_pair.PassClientHandle()));
+ channels_[1]->SetBootstrapEndpoint(ep1);
+ }
+
+ void TearDownOnIOThread() {
+ CHECK_EQ(base::MessageLoop::current(), io_thread_.message_loop());
+
+ if (channels_[0]) {
+ channels_[0]->Shutdown();
+ channels_[0] = nullptr;
+ }
+ if (channels_[1]) {
+ channels_[1]->Shutdown();
+ channels_[1] = nullptr;
+ }
+ }
+
+ embedder::SimplePlatformSupport platform_support_;
+ base::TestIOThread io_thread_;
+ scoped_refptr<Channel> channels_[2];
+ scoped_refptr<MessagePipe> message_pipes_[2];
+
+ scoped_refptr<DataPipe> dp_;
+
+ DISALLOW_COPY_AND_ASSIGN(RemoteDataPipeImplTestHelper);
+};
+
+// RemoteProducerDataPipeImplTestHelper ----------------------------------------
+
+// Note about naming confusion: This class is named after the "local" class,
+// i.e., |dp_| will have a |RemoteProducerDataPipeImpl|. The remote side, of
+// course, will have a |RemoteConsumerDataPipeImpl|.
+class RemoteProducerDataPipeImplTestHelper
+ : public RemoteDataPipeImplTestHelper {
+ public:
+ RemoteProducerDataPipeImplTestHelper() {}
+ ~RemoteProducerDataPipeImplTestHelper() override {}
+
+ void DoTransfer() override {
+ // This is the producer dispatcher we'll send.
+ scoped_refptr<DataPipeProducerDispatcher> to_send =
+ new DataPipeProducerDispatcher();
+ to_send->Init(dp());
+ scoped_refptr<Dispatcher> to_receive;
+ SendDispatcher(0, to_send, &to_receive);
+ // |to_send| should have been closed. This is |DCHECK()|ed when it is
+ // destroyed.
+ EXPECT_TRUE(to_send->HasOneRef());
+ to_send = nullptr;
+
+ ASSERT_EQ(Dispatcher::kTypeDataPipeProducer, to_receive->GetType());
+ producer_dispatcher_ =
+ static_cast<DataPipeProducerDispatcher*>(to_receive.get());
+ }
+
+ DataPipe* dpp() override {
+ if (producer_dispatcher_)
+ return producer_dispatcher_->GetDataPipeForTest();
+ return dp().get();
+ }
+ DataPipe* dpc() override { return dp().get(); }
+
+ void ProducerClose() override {
+ if (producer_dispatcher_)
+ ASSERT_EQ(MOJO_RESULT_OK, producer_dispatcher_->Close());
+ else
+ dp()->ProducerClose();
+ }
+ void ConsumerClose() override { dp()->ConsumerClose(); }
+
+ protected:
+ scoped_refptr<DataPipeProducerDispatcher> producer_dispatcher_;
+
+ private:
+ DISALLOW_COPY_AND_ASSIGN(RemoteProducerDataPipeImplTestHelper);
+};
+
+// RemoteConsumerDataPipeImplTestHelper ----------------------------------------
+
+// Note about naming confusion: This class is named after the "local" class,
+// i.e., |dp_| will have a |RemoteConsumerDataPipeImpl|. The remote side, of
+// course, will have a |RemoteProducerDataPipeImpl|.
+class RemoteConsumerDataPipeImplTestHelper
+ : public RemoteDataPipeImplTestHelper {
+ public:
+ RemoteConsumerDataPipeImplTestHelper() {}
+ ~RemoteConsumerDataPipeImplTestHelper() override {}
+
+ void DoTransfer() override {
+ // This is the consumer dispatcher we'll send.
+ scoped_refptr<DataPipeConsumerDispatcher> to_send =
+ new DataPipeConsumerDispatcher();
+ to_send->Init(dp());
+ scoped_refptr<Dispatcher> to_receive;
+ SendDispatcher(0, to_send, &to_receive);
+ // |to_send| should have been closed. This is |DCHECK()|ed when it is
+ // destroyed.
+ EXPECT_TRUE(to_send->HasOneRef());
+ to_send = nullptr;
+
+ ASSERT_EQ(Dispatcher::kTypeDataPipeConsumer, to_receive->GetType());
+ consumer_dispatcher_ =
+ static_cast<DataPipeConsumerDispatcher*>(to_receive.get());
+ }
+
+ DataPipe* dpp() override { return dp().get(); }
+ DataPipe* dpc() override {
+ if (consumer_dispatcher_)
+ return consumer_dispatcher_->GetDataPipeForTest();
+ return dp().get();
+ }
+
+ void ProducerClose() override { dp()->ProducerClose(); }
+ void ConsumerClose() override {
+ if (consumer_dispatcher_)
+ ASSERT_EQ(MOJO_RESULT_OK, consumer_dispatcher_->Close());
+ else
+ dp()->ConsumerClose();
+ }
+
+ protected:
+ scoped_refptr<DataPipeConsumerDispatcher> consumer_dispatcher_;
+
+ private:
+ DISALLOW_COPY_AND_ASSIGN(RemoteConsumerDataPipeImplTestHelper);
+};
+
+// RemoteProducerDataPipeImplTestHelper2 ---------------------------------------
+
+// This is like |RemoteProducerDataPipeImplTestHelper|, but |DoTransfer()| does
+// a second transfer. This thus tests passing a producer handle twice, and in
+// particular tests (some of) |RemoteConsumerDataPipeImpl|'s
+// |ProducerEndSerialize()| (instead of |LocalDataPipeImpl|'s).
+//
+// Note about naming confusion: This class is named after the "local" class,
+// i.e., |dp_| will have a |RemoteProducerDataPipeImpl|. The remote side, of
+// course, will have a |RemoteConsumerDataPipeImpl|.
+class RemoteProducerDataPipeImplTestHelper2
+ : public RemoteProducerDataPipeImplTestHelper {
+ public:
+ RemoteProducerDataPipeImplTestHelper2() {}
+ ~RemoteProducerDataPipeImplTestHelper2() override {}
+
+ void DoTransfer() override {
+ // This is the producer dispatcher we'll send.
+ scoped_refptr<DataPipeProducerDispatcher> to_send =
+ new DataPipeProducerDispatcher();
+ to_send->Init(dp());
+ scoped_refptr<Dispatcher> to_receive;
+ SendDispatcher(0, to_send, &to_receive);
+ // |to_send| should have been closed. This is |DCHECK()|ed when it is
+ // destroyed.
+ EXPECT_TRUE(to_send->HasOneRef());
+ to_send = nullptr;
+ ASSERT_EQ(Dispatcher::kTypeDataPipeProducer, to_receive->GetType());
+ to_send = static_cast<DataPipeProducerDispatcher*>(to_receive.get());
+ to_receive = nullptr;
+
+ // Now send it back the other way.
+ SendDispatcher(1, to_send, &to_receive);
+ // |producer_dispatcher_| should have been closed. This is |DCHECK()|ed when
+ // it is destroyed.
+ EXPECT_TRUE(to_send->HasOneRef());
+ to_send = nullptr;
+
+ ASSERT_EQ(Dispatcher::kTypeDataPipeProducer, to_receive->GetType());
+ producer_dispatcher_ =
+ static_cast<DataPipeProducerDispatcher*>(to_receive.get());
+ }
+
+ private:
+ DISALLOW_COPY_AND_ASSIGN(RemoteProducerDataPipeImplTestHelper2);
+};
+
+// RemoteConsumerDataPipeImplTestHelper2 ---------------------------------------
+
+// This is like |RemoteConsumerDataPipeImplTestHelper|, but |DoTransfer()| does
+// a second transfer. This thus tests passing a consumer handle twice, and in
+// particular tests (some of) |RemoteProducerDataPipeImpl|'s
+// |ConsumerEndSerialize()| (instead of |LocalDataPipeImpl|'s).
+//
+// Note about naming confusion: This class is named after the "local" class,
+// i.e., |dp_| will have a |RemoteConsumerDataPipeImpl|. The remote side, of
+// course, will have a |RemoteProducerDataPipeImpl|.
+class RemoteConsumerDataPipeImplTestHelper2
+ : public RemoteConsumerDataPipeImplTestHelper {
+ public:
+ RemoteConsumerDataPipeImplTestHelper2() {}
+ ~RemoteConsumerDataPipeImplTestHelper2() override {}
+
+ void DoTransfer() override {
+ // This is the consumer dispatcher we'll send.
+ scoped_refptr<DataPipeConsumerDispatcher> to_send =
+ new DataPipeConsumerDispatcher();
+ to_send->Init(dp());
+ scoped_refptr<Dispatcher> to_receive;
+ SendDispatcher(0, to_send, &to_receive);
+ // |to_send| should have been closed. This is |DCHECK()|ed when it is
+ // destroyed.
+ EXPECT_TRUE(to_send->HasOneRef());
+ to_send = nullptr;
+ ASSERT_EQ(Dispatcher::kTypeDataPipeConsumer, to_receive->GetType());
+ to_send = static_cast<DataPipeConsumerDispatcher*>(to_receive.get());
+ to_receive = nullptr;
+
+ // Now send it back the other way.
+ SendDispatcher(1, to_send, &to_receive);
+ // |consumer_dispatcher_| should have been closed. This is |DCHECK()|ed when
+ // it is destroyed.
+ EXPECT_TRUE(to_send->HasOneRef());
+ to_send = nullptr;
+
+ ASSERT_EQ(Dispatcher::kTypeDataPipeConsumer, to_receive->GetType());
+ consumer_dispatcher_ =
+ static_cast<DataPipeConsumerDispatcher*>(to_receive.get());
+ }
+
+ private:
+ DISALLOW_COPY_AND_ASSIGN(RemoteConsumerDataPipeImplTestHelper2);
+};
+
+// Test case instantiation -----------------------------------------------------
+
+typedef testing::Types<LocalDataPipeImplTestHelper,
+ RemoteProducerDataPipeImplTestHelper,
+ RemoteConsumerDataPipeImplTestHelper,
+ RemoteProducerDataPipeImplTestHelper2,
+ RemoteConsumerDataPipeImplTestHelper2> HelperTypes;
+
+TYPED_TEST_CASE(DataPipeImplTest, HelperTypes);
+
+// Tests -----------------------------------------------------------------------
+
+TYPED_TEST(DataPipeImplTest, SimpleReadWrite) {
+ 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;
+ uint32_t context;
+
+ int32_t elements[10] = {};
+ uint32_t num_bytes = 0;
+
+ // Try reading; nothing there yet.
+ num_bytes = static_cast<uint32_t>(arraysize(elements) * sizeof(elements[0]));
+ EXPECT_EQ(
+ MOJO_RESULT_SHOULD_WAIT,
+ this->dpc()->ConsumerReadData(UserPointer<void>(elements),
+ MakeUserPointer(&num_bytes), false, false));
+
+ // Query; nothing there yet.
+ num_bytes = 0;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerQueryData(MakeUserPointer(&num_bytes)));
+ EXPECT_EQ(0u, num_bytes);
+
+ // Discard; nothing there yet.
+ num_bytes = static_cast<uint32_t>(5u * sizeof(elements[0]));
+ EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT, this->dpc()->ConsumerDiscardData(
+ MakeUserPointer(&num_bytes), false));
+
+ // Read with invalid |num_bytes|.
+ num_bytes = sizeof(elements[0]) + 1;
+ EXPECT_EQ(
+ MOJO_RESULT_INVALID_ARGUMENT,
+ this->dpc()->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->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_READABLE, 123, 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->dpp()->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::ActionDeadline(), &context));
+ EXPECT_EQ(123u, context);
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&waiter, &hss);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+ 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;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerQueryData(MakeUserPointer(&num_bytes)));
+ EXPECT_EQ(2 * sizeof(elements[0]), num_bytes);
+
+ // 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->dpc()->ConsumerReadData(
+ UserPointer<void>(elements),
+ MakeUserPointer(&num_bytes), false, false));
+ EXPECT_EQ(1u * sizeof(elements[0]), num_bytes);
+ EXPECT_EQ(123, elements[0]);
+ EXPECT_EQ(-1, elements[1]);
+
+ // Query.
+ // TODO(vtl): See previous TODO. (If we got 2 elements there, however, we
+ // should get 1 here.)
+ num_bytes = 0;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerQueryData(MakeUserPointer(&num_bytes)));
+ EXPECT_EQ(1 * sizeof(elements[0]), num_bytes);
+
+ // 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->dpc()->ConsumerReadData(
+ UserPointer<void>(elements),
+ MakeUserPointer(&num_bytes), false, true));
+ EXPECT_EQ(1u * sizeof(elements[0]), num_bytes);
+ EXPECT_EQ(456, elements[0]);
+ EXPECT_EQ(-1, elements[1]);
+
+ // Query. Still has 1 element remaining.
+ num_bytes = 0;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerQueryData(MakeUserPointer(&num_bytes)));
+ EXPECT_EQ(1 * sizeof(elements[0]), num_bytes);
+
+ // Try to read two elements, with "all or none".
+ elements[0] = -1;
+ elements[1] = -1;
+ num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
+ EXPECT_EQ(
+ MOJO_RESULT_OUT_OF_RANGE,
+ this->dpc()->ConsumerReadData(UserPointer<void>(elements),
+ MakeUserPointer(&num_bytes), true, false));
+ EXPECT_EQ(-1, elements[0]);
+ EXPECT_EQ(-1, elements[1]);
+
+ // Try to read two elements, without "all or none".
+ elements[0] = -1;
+ elements[1] = -1;
+ num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
+ EXPECT_EQ(MOJO_RESULT_OK, this->dpc()->ConsumerReadData(
+ UserPointer<void>(elements),
+ MakeUserPointer(&num_bytes), false, false));
+ EXPECT_EQ(1u * sizeof(elements[0]), num_bytes);
+ EXPECT_EQ(456, elements[0]);
+ EXPECT_EQ(-1, elements[1]);
+
+ // Query.
+ num_bytes = 0;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerQueryData(MakeUserPointer(&num_bytes)));
+ EXPECT_EQ(0u, num_bytes);
+
+ this->ProducerClose();
+ this->ConsumerClose();
+}
+
+// Note: The "basic" waiting tests test that the "wait states" are correct in
+// various situations; they don't test that waiters are properly awoken on state
+// changes. (For that, we need to use multiple threads.)
+TYPED_TEST(DataPipeImplTest, BasicProducerWaiting) {
+ // Note: We take advantage of the fact that current for current
+ // implementations capacities are strict maximums. This is not guaranteed by
+ // the API.
+
+ 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 pwaiter; // For producer.
+ Waiter cwaiter; // For consumer.
+ HandleSignalsState hss;
+ uint32_t context;
+
+ // Never readable.
+ pwaiter.Init();
+ hss = HandleSignalsState();
+ EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
+ this->dpp()->ProducerAddAwakable(
+ &pwaiter, MOJO_HANDLE_SIGNAL_READABLE, 12, &hss));
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+ hss.satisfiable_signals);
+
+ // Already writable.
+ pwaiter.Init();
+ hss = HandleSignalsState();
+ EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+ this->dpp()->ProducerAddAwakable(
+ &pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 34, &hss));
+
+ // We'll need to wait for readability for the remote cases.
+ cwaiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerAddAwakable(
+ &cwaiter, MOJO_HANDLE_SIGNAL_READABLE, 1234, 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->dpp()->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->dpp()->ProducerAddAwakable(
+ &pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 56, nullptr));
+ // And it shouldn't be writable yet.
+ EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, pwaiter.Wait(0, nullptr));
+ hss = HandleSignalsState();
+ this->dpp()->ProducerRemoveAwakable(&pwaiter, &hss);
+ EXPECT_EQ(0u, hss.satisfied_signals);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+ hss.satisfiable_signals);
+
+ // Wait for data to become available to the consumer.
+ context = 0;
+ EXPECT_EQ(MOJO_RESULT_OK, cwaiter.Wait(test::TinyDeadline(), &context));
+ EXPECT_EQ(1234u, context);
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&cwaiter, &hss);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_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, this->dpc()->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->dpp()->ProducerAddAwakable(
+ &pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 56, nullptr));
+ // And it still shouldn't be writable yet.
+ EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, pwaiter.Wait(0, nullptr));
+ hss = HandleSignalsState();
+ this->dpp()->ProducerRemoveAwakable(&pwaiter, &hss);
+ EXPECT_EQ(0u, hss.satisfied_signals);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+ hss.satisfiable_signals);
+
+ // Do it again.
+ pwaiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpp()->ProducerAddAwakable(
+ &pwaiter, 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, this->dpc()->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::TinyDeadline(), &context));
+ EXPECT_EQ(78u, context);
+ hss = HandleSignalsState();
+ this->dpp()->ProducerRemoveAwakable(&pwaiter, &hss);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_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,
+ this->dpp()->ProducerBeginWriteData(
+ MakeUserPointer(&buffer), MakeUserPointer(&num_bytes), false));
+ 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->dpp()->ProducerEndWriteData(
+ static_cast<uint32_t>(1u * sizeof(elements[0]))));
+
+ // Add a waiter.
+ pwaiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpp()->ProducerAddAwakable(
+ &pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 90, nullptr));
+
+ // Read one element, using a two-phase read.
+ const void* read_buffer = nullptr;
+ num_bytes = 0u;
+ EXPECT_EQ(MOJO_RESULT_OK, this->dpc()->ConsumerBeginReadData(
+ MakeUserPointer(&read_buffer),
+ MakeUserPointer(&num_bytes), false));
+ 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->dpc()->ConsumerEndReadData(
+ static_cast<uint32_t>(1u * sizeof(elements[0]))));
+
+ // Waiting should succeed.
+ context = 0;
+ EXPECT_EQ(MOJO_RESULT_OK, pwaiter.Wait(test::TinyDeadline(), &context));
+ EXPECT_EQ(90u, context);
+ hss = HandleSignalsState();
+ this->dpp()->ProducerRemoveAwakable(&pwaiter, &hss);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_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,
+ this->dpp()->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->dpp()->ProducerAddAwakable(
+ &pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 12, nullptr));
+
+ // Close the consumer.
+ this->ConsumerClose();
+
+ // It should now be never-writable.
+ context = 0;
+ EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
+ pwaiter.Wait(test::TinyDeadline(), &context));
+ EXPECT_EQ(12u, context);
+ hss = HandleSignalsState();
+ this->dpp()->ProducerRemoveAwakable(&pwaiter, &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;
+ uint32_t context;
+
+ // Add a waiter.
+ waiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpp()->ProducerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED, 12, nullptr));
+
+ // Close the consumer.
+ this->ConsumerClose();
+
+ // It should be signaled.
+ context = 0;
+ EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyDeadline(), &context));
+ EXPECT_EQ(12u, context);
+ hss = HandleSignalsState();
+ this->dpp()->ProducerRemoveAwakable(&waiter, &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;
+ uint32_t context;
+
+ // Add a waiter.
+ waiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED, 12, nullptr));
+
+ // Close the producer.
+ this->ProducerClose();
+
+ // It should be signaled.
+ context = 0;
+ EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyDeadline(), &context));
+ EXPECT_EQ(12u, context);
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&waiter, &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;
+ uint32_t context;
+
+ // Never writable.
+ waiter.Init();
+ hss = HandleSignalsState();
+ EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
+ this->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_WRITABLE, 12, &hss));
+ EXPECT_EQ(0u, hss.satisfied_signals);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+ hss.satisfiable_signals);
+
+ // Add waiter: not yet readable.
+ waiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_READABLE, 34, 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->dpp()->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::TinyDeadline(), &context));
+ EXPECT_EQ(34u, context);
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&waiter, &hss);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_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, this->dpc()->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->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_READABLE, 78, &hss));
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_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, this->dpc()->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->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_READABLE, 78, &hss));
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, 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, this->dpc()->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->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_READABLE, 90, 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->dpp()->ProducerWriteData(UserPointer<const void>(elements),
+ MakeUserPointer(&num_bytes), true));
+
+ // Waiting should now succeed.
+ context = 0;
+ EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyDeadline(), &context));
+ EXPECT_EQ(90u, context);
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&waiter, &hss);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+ hss.satisfiable_signals);
+
+ // We'll want to wait for the peer closed signal to propagate.
+ waiter.Init();
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED, 12, 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->dpc()->ConsumerAddAwakable(
+ &waiter2, MOJO_HANDLE_SIGNAL_READABLE, 34, &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,
+ hss.satisfiable_signals);
+
+ // Wait for the peer closed signal.
+ context = 0;
+ EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyDeadline(), &context));
+ EXPECT_EQ(12u, context);
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&waiter, &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);
+
+ // 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->dpc()->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->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_READABLE, 56, &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;
+ uint32_t context;
+
+ // Add waiter: not yet readable.
+ waiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_READABLE, 12, nullptr));
+
+ // 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]));
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpp()->ProducerBeginWriteData(
+ MakeUserPointer(&buffer), MakeUserPointer(&num_bytes), true));
+ 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->dpp()->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::TinyDeadline(), &context));
+ EXPECT_EQ(12u, context);
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&waiter, &hss);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_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, this->dpc()->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,
+ this->dpc()->ConsumerEndReadData(
+ static_cast<uint32_t>(1u * sizeof(elements[0]))));
+
+ // Should still be readable.
+ waiter.Init();
+ hss = HandleSignalsState();
+ EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+ this->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_READABLE, 34, &hss));
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_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, this->dpc()->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,
+ this->dpc()->ConsumerEndReadData(
+ static_cast<uint32_t>(1u * sizeof(elements[0]))));
+
+ // Adding a waiter should now succeed.
+ waiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_READABLE, 56, nullptr));
+
+ // Close the producer.
+ this->ProducerClose();
+
+ // Should be never-readable.
+ context = 0;
+ EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
+ waiter.Wait(test::TinyDeadline(), &context));
+ EXPECT_EQ(56u, context);
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&waiter, &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->dpp()->ProducerAddAwakable(
+ &pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 0, &hss));
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_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, this->dpp()->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.
+ pwaiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpp()->ProducerAddAwakable(
+ &pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 1, nullptr));
+ EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, pwaiter.Wait(0, nullptr));
+ hss = HandleSignalsState();
+ this->dpp()->ProducerRemoveAwakable(&pwaiter, &hss);
+ EXPECT_EQ(0u, hss.satisfied_signals);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+ hss.satisfiable_signals);
+
+ // It shouldn't be readable yet either (we'll wait later).
+ cwaiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerAddAwakable(
+ &cwaiter, MOJO_HANDLE_SIGNAL_READABLE, 2, nullptr));
+
+ static_cast<int32_t*>(write_ptr)[0] = 123;
+ EXPECT_EQ(MOJO_RESULT_OK, this->dpp()->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->dpp()->ProducerAddAwakable(
+ &pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 3, &hss));
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+ hss.satisfiable_signals);
+
+ // It should become readable.
+ EXPECT_EQ(MOJO_RESULT_OK, cwaiter.Wait(test::TinyDeadline(), nullptr));
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&cwaiter, &hss);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_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, this->dpp()->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.
+ cwaiter.Init();
+ hss = HandleSignalsState();
+ EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+ this->dpc()->ConsumerAddAwakable(
+ &cwaiter, MOJO_HANDLE_SIGNAL_READABLE, 5, &hss));
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_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, this->dpp()->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, this->dpc()->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.
+ pwaiter.Init();
+ hss = HandleSignalsState();
+ EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+ this->dpp()->ProducerAddAwakable(
+ &pwaiter, MOJO_HANDLE_SIGNAL_WRITABLE, 6, &hss));
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+ hss.satisfiable_signals);
+
+ // But not readable.
+ cwaiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerAddAwakable(
+ &cwaiter, MOJO_HANDLE_SIGNAL_READABLE, 7, nullptr));
+ EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, cwaiter.Wait(0, nullptr));
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&cwaiter, &hss);
+ EXPECT_EQ(0u, hss.satisfied_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, this->dpc()->ConsumerEndReadData(0u));
+
+ // It should be readable again.
+ cwaiter.Init();
+ hss = HandleSignalsState();
+ EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+ this->dpc()->ConsumerAddAwakable(
+ &cwaiter, MOJO_HANDLE_SIGNAL_READABLE, 8, &hss));
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+ hss.satisfiable_signals);
+
+ this->ProducerClose();
+ this->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);
+}
+
+TYPED_TEST(DataPipeImplTest, AllOrNone) {
+ const MojoCreateDataPipeOptions options = {
+ kSizeOfOptions, // |struct_size|.
+ MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE, // |flags|.
+ static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
+ 10 * sizeof(int32_t) // |capacity_num_bytes|.
+ };
+ this->Create(options);
+ this->DoTransfer();
+
+ Waiter waiter;
+ HandleSignalsState hss;
+
+ // Try writing way too much.
+ uint32_t num_bytes = 20u * sizeof(int32_t);
+ int32_t buffer[100];
+ Seq(0, arraysize(buffer), buffer);
+ EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE,
+ this->dpp()->ProducerWriteData(UserPointer<const void>(buffer),
+ MakeUserPointer(&num_bytes), true));
+
+ // Should still be empty.
+ num_bytes = ~0u;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerQueryData(MakeUserPointer(&num_bytes)));
+ EXPECT_EQ(0u, num_bytes);
+
+ // Add waiter.
+ waiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_READABLE, 1, nullptr));
+
+ // Write some data.
+ num_bytes = 5u * sizeof(int32_t);
+ Seq(100, arraysize(buffer), buffer);
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpp()->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::TinyDeadline(), nullptr));
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&waiter, &hss);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+ hss.satisfiable_signals);
+
+ // Half full.
+ num_bytes = 0u;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerQueryData(MakeUserPointer(&num_bytes)));
+ EXPECT_EQ(5u * sizeof(int32_t), num_bytes);
+
+ // Too much.
+ num_bytes = 6u * sizeof(int32_t);
+ Seq(200, arraysize(buffer), buffer);
+ EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE,
+ this->dpp()->ProducerWriteData(UserPointer<const void>(buffer),
+ MakeUserPointer(&num_bytes), true));
+
+ // Try reading too much.
+ num_bytes = 11u * sizeof(int32_t);
+ memset(buffer, 0xab, sizeof(buffer));
+ EXPECT_EQ(
+ MOJO_RESULT_OUT_OF_RANGE,
+ this->dpc()->ConsumerReadData(UserPointer<void>(buffer),
+ MakeUserPointer(&num_bytes), true, false));
+ int32_t expected_buffer[100];
+ memset(expected_buffer, 0xab, sizeof(expected_buffer));
+ EXPECT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
+
+ // Try discarding too much.
+ num_bytes = 11u * sizeof(int32_t);
+ EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE, this->dpc()->ConsumerDiscardData(
+ MakeUserPointer(&num_bytes), true));
+
+ // Just a little.
+ num_bytes = 2u * sizeof(int32_t);
+ Seq(300, arraysize(buffer), buffer);
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpp()->ProducerWriteData(UserPointer<const void>(buffer),
+ MakeUserPointer(&num_bytes), true));
+ EXPECT_EQ(2u * sizeof(int32_t), num_bytes);
+
+ // Just right.
+ num_bytes = 3u * sizeof(int32_t);
+ Seq(400, arraysize(buffer), buffer);
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpp()->ProducerWriteData(UserPointer<const void>(buffer),
+ MakeUserPointer(&num_bytes), true));
+ EXPECT_EQ(3u * sizeof(int32_t), num_bytes);
+
+ // TODO(vtl): Hack (see also the TODO above): We can't currently wait for a
+ // specified amount of data to be available, so poll.
+ const size_t kMaxPoll = 100;
+ for (size_t i = 0; i < kMaxPoll; i++) {
+ num_bytes = 0u;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerQueryData(MakeUserPointer(&num_bytes)));
+ if (num_bytes >= 10u * sizeof(int32_t))
+ break;
+
+ base::PlatformThread::Sleep(test::EpsilonTimeout());
+ }
+ EXPECT_EQ(10u * sizeof(int32_t), num_bytes);
+
+ // Read half.
+ num_bytes = 5u * sizeof(int32_t);
+ memset(buffer, 0xab, sizeof(buffer));
+ EXPECT_EQ(MOJO_RESULT_OK, this->dpc()->ConsumerReadData(
+ UserPointer<void>(buffer),
+ MakeUserPointer(&num_bytes), true, false));
+ EXPECT_EQ(5u * sizeof(int32_t), num_bytes);
+ memset(expected_buffer, 0xab, sizeof(expected_buffer));
+ Seq(100, 5, expected_buffer);
+ EXPECT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
+
+ // Try reading too much again.
+ num_bytes = 6u * sizeof(int32_t);
+ memset(buffer, 0xab, sizeof(buffer));
+ EXPECT_EQ(
+ MOJO_RESULT_OUT_OF_RANGE,
+ this->dpc()->ConsumerReadData(UserPointer<void>(buffer),
+ MakeUserPointer(&num_bytes), true, false));
+ memset(expected_buffer, 0xab, sizeof(expected_buffer));
+ EXPECT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
+
+ // Try discarding too much again.
+ num_bytes = 6u * sizeof(int32_t);
+ EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE, this->dpc()->ConsumerDiscardData(
+ MakeUserPointer(&num_bytes), true));
+
+ // Discard a little.
+ num_bytes = 2u * sizeof(int32_t);
+ EXPECT_EQ(MOJO_RESULT_OK, this->dpc()->ConsumerDiscardData(
+ MakeUserPointer(&num_bytes), true));
+ EXPECT_EQ(2u * sizeof(int32_t), num_bytes);
+
+ // Three left.
+ num_bytes = 0u;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->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->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED, 2, nullptr));
+
+ // Close the producer, then test producer-closed cases.
+ this->ProducerClose();
+
+ // Wait.
+ EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyDeadline(), nullptr));
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&waiter, &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);
+
+ // Try reading too much; "failed precondition" since the producer is closed.
+ num_bytes = 4u * sizeof(int32_t);
+ memset(buffer, 0xab, sizeof(buffer));
+ EXPECT_EQ(
+ MOJO_RESULT_FAILED_PRECONDITION,
+ this->dpc()->ConsumerReadData(UserPointer<void>(buffer),
+ MakeUserPointer(&num_bytes), true, false));
+ memset(expected_buffer, 0xab, sizeof(expected_buffer));
+ EXPECT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
+
+ // Try discarding too much; "failed precondition" again.
+ num_bytes = 4u * sizeof(int32_t);
+ EXPECT_EQ(
+ MOJO_RESULT_FAILED_PRECONDITION,
+ this->dpc()->ConsumerDiscardData(MakeUserPointer(&num_bytes), true));
+
+ // Read a little.
+ num_bytes = 2u * sizeof(int32_t);
+ memset(buffer, 0xab, sizeof(buffer));
+ EXPECT_EQ(MOJO_RESULT_OK, this->dpc()->ConsumerReadData(
+ UserPointer<void>(buffer),
+ MakeUserPointer(&num_bytes), true, false));
+ EXPECT_EQ(2u * sizeof(int32_t), num_bytes);
+ memset(expected_buffer, 0xab, sizeof(expected_buffer));
+ Seq(400, 2, expected_buffer);
+ EXPECT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
+
+ // Discard the remaining element.
+ num_bytes = 1u * sizeof(int32_t);
+ EXPECT_EQ(MOJO_RESULT_OK, this->dpc()->ConsumerDiscardData(
+ MakeUserPointer(&num_bytes), true));
+ EXPECT_EQ(1u * sizeof(int32_t), num_bytes);
+
+ // Empty again.
+ num_bytes = ~0u;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerQueryData(MakeUserPointer(&num_bytes)));
+ EXPECT_EQ(0u, num_bytes);
+
+ this->ConsumerClose();
+}
+
+TYPED_TEST(DataPipeImplTest, TwoPhaseAllOrNone) {
+ const MojoCreateDataPipeOptions options = {
+ kSizeOfOptions, // |struct_size|.
+ MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE, // |flags|.
+ static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
+ 10 * sizeof(int32_t) // |capacity_num_bytes|.
+ };
+ this->Create(options);
+ this->DoTransfer();
+
+ Waiter waiter;
+ HandleSignalsState hss;
+
+ // Try writing way too much (two-phase).
+ uint32_t num_bytes = 20u * sizeof(int32_t);
+ void* write_ptr = nullptr;
+ EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE, this->dpp()->ProducerBeginWriteData(
+ MakeUserPointer(&write_ptr),
+ MakeUserPointer(&num_bytes), true));
+
+ // Try writing an amount which isn't a multiple of the element size
+ // (two-phase).
+ static_assert(sizeof(int32_t) > 1u, "Wow! int32_t's have size 1");
+ num_bytes = 1u;
+ write_ptr = nullptr;
+ EXPECT_EQ(
+ MOJO_RESULT_INVALID_ARGUMENT,
+ this->dpp()->ProducerBeginWriteData(MakeUserPointer(&write_ptr),
+ MakeUserPointer(&num_bytes), true));
+
+ // Try reading way too much (two-phase).
+ num_bytes = 20u * sizeof(int32_t);
+ const void* read_ptr = nullptr;
+ EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE,
+ this->dpc()->ConsumerBeginReadData(
+ MakeUserPointer(&read_ptr), MakeUserPointer(&num_bytes), true));
+
+ // Add waiter.
+ waiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_READABLE, 1, nullptr));
+
+ // Write half (two-phase).
+ num_bytes = 5u * sizeof(int32_t);
+ write_ptr = nullptr;
+ EXPECT_EQ(MOJO_RESULT_OK, this->dpp()->ProducerBeginWriteData(
+ MakeUserPointer(&write_ptr),
+ MakeUserPointer(&num_bytes), true));
+ // May provide more space than requested.
+ EXPECT_GE(num_bytes, 5u * sizeof(int32_t));
+ EXPECT_TRUE(write_ptr);
+ Seq(0, 5, static_cast<int32_t*>(write_ptr));
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpp()->ProducerEndWriteData(5u * sizeof(int32_t)));
+
+ // Wait for data.
+ // TODO(vtl): (See corresponding TODO in AllOrNone.)
+ EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyDeadline(), nullptr));
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&waiter, &hss);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
+ EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+ hss.satisfiable_signals);
+
+ // Try reading an amount which isn't a multiple of the element size
+ // (two-phase).
+ num_bytes = 1u;
+ read_ptr = nullptr;
+ EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
+ this->dpc()->ConsumerBeginReadData(
+ MakeUserPointer(&read_ptr), MakeUserPointer(&num_bytes), true));
+
+ // Read one (two-phase).
+ num_bytes = 1u * sizeof(int32_t);
+ read_ptr = nullptr;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerBeginReadData(
+ MakeUserPointer(&read_ptr), MakeUserPointer(&num_bytes), true));
+ EXPECT_GE(num_bytes, 1u * sizeof(int32_t));
+ EXPECT_EQ(0, static_cast<const int32_t*>(read_ptr)[0]);
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerEndReadData(1u * sizeof(int32_t)));
+
+ // We should have four left, leaving room for six.
+ num_bytes = 0u;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerQueryData(MakeUserPointer(&num_bytes)));
+ EXPECT_EQ(4u * sizeof(int32_t), num_bytes);
+
+ // Assuming a tight circular buffer of the specified capacity, we can't do a
+ // two-phase write of six now.
+ num_bytes = 6u * sizeof(int32_t);
+ write_ptr = nullptr;
+ EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE, this->dpp()->ProducerBeginWriteData(
+ MakeUserPointer(&write_ptr),
+ MakeUserPointer(&num_bytes), true));
+
+ // TODO(vtl): Hack (see also the TODO above): We can't currently wait for a
+ // specified amount of space to be available, so poll.
+ const size_t kMaxPoll = 100;
+ for (size_t i = 0; i < kMaxPoll; i++) {
+ // Write six elements (simple), filling the buffer.
+ num_bytes = 6u * sizeof(int32_t);
+ int32_t buffer[100];
+ Seq(100, 6, buffer);
+ MojoResult result = this->dpp()->ProducerWriteData(
+ UserPointer<const void>(buffer), MakeUserPointer(&num_bytes), true);
+ if (result == MOJO_RESULT_OK)
+ break;
+ EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE, result);
+
+ base::PlatformThread::Sleep(test::EpsilonTimeout());
+ }
+ EXPECT_EQ(6u * sizeof(int32_t), num_bytes);
+
+ // TODO(vtl): Hack: poll again.
+ for (size_t i = 0; i < kMaxPoll; i++) {
+ // We have ten.
+ num_bytes = 0u;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerQueryData(MakeUserPointer(&num_bytes)));
+ if (num_bytes >= 10u * sizeof(int32_t))
+ break;
+
+ base::PlatformThread::Sleep(test::EpsilonTimeout());
+ }
+ EXPECT_EQ(10u * sizeof(int32_t), num_bytes);
+
+ // Note: Whether a two-phase read of ten would fail here or not is
+ // implementation-dependent.
+
+ // Add waiter.
+ waiter.Init();
+ ASSERT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerAddAwakable(
+ &waiter, MOJO_HANDLE_SIGNAL_PEER_CLOSED, 2, nullptr));
+
+ // Close the producer.
+ this->ProducerClose();
+
+ // A two-phase read of nine should work.
+ num_bytes = 9u * sizeof(int32_t);
+ read_ptr = nullptr;
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerBeginReadData(
+ MakeUserPointer(&read_ptr), MakeUserPointer(&num_bytes), true));
+ EXPECT_GE(num_bytes, 9u * sizeof(int32_t));
+ EXPECT_EQ(1, static_cast<const int32_t*>(read_ptr)[0]);
+ EXPECT_EQ(2, static_cast<const int32_t*>(read_ptr)[1]);
+ EXPECT_EQ(3, static_cast<const int32_t*>(read_ptr)[2]);
+ EXPECT_EQ(4, static_cast<const int32_t*>(read_ptr)[3]);
+ EXPECT_EQ(100, static_cast<const int32_t*>(read_ptr)[4]);
+ EXPECT_EQ(101, static_cast<const int32_t*>(read_ptr)[5]);
+ EXPECT_EQ(102, static_cast<const int32_t*>(read_ptr)[6]);
+ EXPECT_EQ(103, static_cast<const int32_t*>(read_ptr)[7]);
+ EXPECT_EQ(104, static_cast<const int32_t*>(read_ptr)[8]);
+ EXPECT_EQ(MOJO_RESULT_OK,
+ this->dpc()->ConsumerEndReadData(9u * sizeof(int32_t)));
+
+ // Wait for peer closed.
+ EXPECT_EQ(MOJO_RESULT_OK, waiter.Wait(test::TinyDeadline(), nullptr));
+ hss = HandleSignalsState();
+ this->dpc()->ConsumerRemoveAwakable(&waiter, &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);
+
+ // A two-phase read of two should fail, with "failed precondition".
+ num_bytes = 2u * sizeof(int32_t);
+ read_ptr = nullptr;
+ EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
+ this->dpc()->ConsumerBeginReadData(
+ MakeUserPointer(&read_ptr), MakeUserPointer(&num_bytes), true));
+
+ this->ConsumerClose();
+}
+
+} // namespace
+} // namespace system
+} // namespace mojo

Powered by Google App Engine
This is Rietveld 408576698