| // Copyright 2015 The Chromium Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| #include <stddef.h> |
| #include <stdint.h> |
| |
| #include <memory> |
| |
| #include "base/bind.h" |
| #include "base/location.h" |
| #include "base/logging.h" |
| #include "base/message_loop/message_loop.h" |
| #include "base/run_loop.h" |
| #include "base/stl_util.h" |
| #include "build/build_config.h" |
| #include "mojo/core/embedder/embedder.h" |
| #include "mojo/core/test/mojo_test_base.h" |
| #include "mojo/core/test_utils.h" |
| #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/message_pipe.h" |
| #include "mojo/public/cpp/system/simple_watcher.h" |
| #include "testing/gtest/include/gtest/gtest.h" |
| |
| namespace mojo { |
| namespace core { |
| namespace { |
| |
| const uint32_t kSizeOfOptions = |
| static_cast<uint32_t>(sizeof(MojoCreateDataPipeOptions)); |
| |
| // In various places, we have to poll (since, e.g., we can't yet wait for a |
| // certain amount of data to be available). This is the maximum number of |
| // iterations (separated by a short sleep). |
| // TODO(vtl): Get rid of this. |
| const size_t kMaxPoll = 100; |
| |
| // Used in Multiprocess test. |
| const size_t kMultiprocessCapacity = 37; |
| const char kMultiprocessTestData[] = "hello i'm a string that is 36 bytes"; |
| const int kMultiprocessMaxIter = 5; |
| |
| // TODO(rockot): There are many uses of ASSERT where EXPECT would be more |
| // appropriate. Fix this. |
| |
| class DataPipeTest : public test::MojoTestBase { |
| public: |
| DataPipeTest() |
| : producer_(MOJO_HANDLE_INVALID), consumer_(MOJO_HANDLE_INVALID) {} |
| |
| ~DataPipeTest() override { |
| if (producer_ != MOJO_HANDLE_INVALID) |
| CHECK_EQ(MOJO_RESULT_OK, MojoClose(producer_)); |
| if (consumer_ != MOJO_HANDLE_INVALID) |
| CHECK_EQ(MOJO_RESULT_OK, MojoClose(consumer_)); |
| } |
| |
| MojoResult ReadEmptyMessageWithHandles(MojoHandle pipe, |
| MojoHandle* out_handles, |
| uint32_t num_handles) { |
| std::vector<uint8_t> bytes; |
| std::vector<ScopedHandle> handles; |
| MojoResult rv = ReadMessageRaw(MessagePipeHandle(pipe), &bytes, &handles, |
| MOJO_READ_MESSAGE_FLAG_NONE); |
| if (rv == MOJO_RESULT_OK) { |
| CHECK_EQ(0u, bytes.size()); |
| CHECK_EQ(num_handles, handles.size()); |
| for (size_t i = 0; i < num_handles; ++i) |
| out_handles[i] = handles[i].release().value(); |
| } |
| return rv; |
| } |
| |
| MojoResult Create(const MojoCreateDataPipeOptions* options) { |
| return MojoCreateDataPipe(options, &producer_, &consumer_); |
| } |
| |
| MojoResult WriteData(const void* elements, |
| uint32_t* num_bytes, |
| bool all_or_none = false) { |
| MojoWriteDataOptions options; |
| options.struct_size = sizeof(options); |
| options.flags = all_or_none ? MOJO_WRITE_DATA_FLAG_ALL_OR_NONE |
| : MOJO_WRITE_DATA_FLAG_NONE; |
| return MojoWriteData(producer_, elements, num_bytes, &options); |
| } |
| |
| MojoResult ReadData(void* elements, |
| uint32_t* num_bytes, |
| bool all_or_none = false, |
| bool peek = false) { |
| MojoReadDataFlags flags = MOJO_READ_DATA_FLAG_NONE; |
| if (all_or_none) |
| flags |= MOJO_READ_DATA_FLAG_ALL_OR_NONE; |
| if (peek) |
| flags |= MOJO_READ_DATA_FLAG_PEEK; |
| |
| MojoReadDataOptions options; |
| options.struct_size = sizeof(options); |
| options.flags = flags; |
| return MojoReadData(consumer_, &options, elements, num_bytes); |
| } |
| |
| MojoResult QueryData(uint32_t* num_bytes) { |
| MojoReadDataOptions options; |
| options.struct_size = sizeof(options); |
| options.flags = MOJO_READ_DATA_FLAG_QUERY; |
| return MojoReadData(consumer_, &options, nullptr, num_bytes); |
| } |
| |
| MojoResult DiscardData(uint32_t* num_bytes, bool all_or_none = false) { |
| MojoReadDataFlags flags = MOJO_READ_DATA_FLAG_DISCARD; |
| if (all_or_none) |
| flags |= MOJO_READ_DATA_FLAG_ALL_OR_NONE; |
| MojoReadDataOptions options; |
| options.struct_size = sizeof(options); |
| options.flags = flags; |
| return MojoReadData(consumer_, &options, nullptr, num_bytes); |
| } |
| |
| MojoResult BeginReadData(const void** elements, uint32_t* num_bytes) { |
| return MojoBeginReadData(consumer_, nullptr, elements, num_bytes); |
| } |
| |
| MojoResult EndReadData(uint32_t num_bytes_read) { |
| return MojoEndReadData(consumer_, num_bytes_read, nullptr); |
| } |
| |
| MojoResult BeginWriteData(void** elements, uint32_t* num_bytes) { |
| return MojoBeginWriteData(producer_, nullptr, elements, num_bytes); |
| } |
| |
| MojoResult EndWriteData(uint32_t num_bytes_written) { |
| return MojoEndWriteData(producer_, num_bytes_written, nullptr); |
| } |
| |
| MojoResult CloseProducer() { |
| MojoResult rv = MojoClose(producer_); |
| producer_ = MOJO_HANDLE_INVALID; |
| return rv; |
| } |
| |
| MojoResult CloseConsumer() { |
| MojoResult rv = MojoClose(consumer_); |
| consumer_ = MOJO_HANDLE_INVALID; |
| return rv; |
| } |
| |
| MojoHandle producer_, consumer_; |
| |
| private: |
| DISALLOW_COPY_AND_ASSIGN(DataPipeTest); |
| }; |
| |
| TEST_F(DataPipeTest, Basic) { |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|. |
| 1000 * sizeof(int32_t) // |capacity_num_bytes|. |
| }; |
| |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| |
| // We can write to a data pipe handle immediately. |
| int32_t elements[10] = {}; |
| uint32_t num_bytes = 0; |
| |
| num_bytes = static_cast<uint32_t>(base::size(elements) * sizeof(elements[0])); |
| |
| elements[0] = 123; |
| elements[1] = 456; |
| num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(&elements[0], &num_bytes)); |
| |
| // Now wait for the other side to become readable. |
| MojoHandleSignalsState state; |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &state)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| state.satisfied_signals); |
| |
| elements[0] = -1; |
| elements[1] = -1; |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(&elements[0], &num_bytes)); |
| ASSERT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes); |
| ASSERT_EQ(elements[0], 123); |
| ASSERT_EQ(elements[1], 456); |
| } |
| |
| // Tests creation of data pipes with various (valid) options. |
| TEST_F(DataPipeTest, CreateAndMaybeTransfer) { |
| MojoCreateDataPipeOptions test_options[] = { |
| // Default options. |
| {}, |
| // Trivial element size, non-default capacity. |
| {kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| 1, // |element_num_bytes|. |
| 1000}, // |capacity_num_bytes|. |
| // Nontrivial element size, non-default capacity. |
| {kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| 4, // |element_num_bytes|. |
| 4000}, // |capacity_num_bytes|. |
| // Nontrivial element size, default capacity. |
| {kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| 100, // |element_num_bytes|. |
| 0} // |capacity_num_bytes|. |
| }; |
| for (size_t i = 0; i < base::size(test_options); i++) { |
| MojoHandle producer_handle, consumer_handle; |
| MojoCreateDataPipeOptions* options = i ? &test_options[i] : nullptr; |
| ASSERT_EQ(MOJO_RESULT_OK, |
| MojoCreateDataPipe(options, &producer_handle, &consumer_handle)); |
| ASSERT_EQ(MOJO_RESULT_OK, MojoClose(producer_handle)); |
| ASSERT_EQ(MOJO_RESULT_OK, MojoClose(consumer_handle)); |
| } |
| } |
| |
| TEST_F(DataPipeTest, SimpleReadWrite) { |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|. |
| 1000 * sizeof(int32_t) // |capacity_num_bytes|. |
| }; |
| |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| int32_t elements[10] = {}; |
| uint32_t num_bytes = 0; |
| |
| // Try reading; nothing there yet. |
| num_bytes = static_cast<uint32_t>(base::size(elements) * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_SHOULD_WAIT, ReadData(elements, &num_bytes)); |
| |
| // Query; nothing there yet. |
| num_bytes = 0; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(0u, num_bytes); |
| |
| // Discard; nothing there yet. |
| num_bytes = static_cast<uint32_t>(5u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_SHOULD_WAIT, DiscardData(&num_bytes)); |
| |
| // Read with invalid |num_bytes|. |
| num_bytes = sizeof(elements[0]) + 1; |
| ASSERT_EQ(MOJO_RESULT_INVALID_ARGUMENT, ReadData(elements, &num_bytes)); |
| |
| // Write two elements. |
| elements[0] = 123; |
| elements[1] = 456; |
| num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes)); |
| // It should have written everything (even without "all or none"). |
| ASSERT_EQ(2u * sizeof(elements[0]), num_bytes); |
| |
| // Wait. |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Query. |
| // TODO(vtl): It's theoretically possible (though not with the current |
| // implementation/configured limits) that not all the data has arrived yet. |
| // (The theoretically-correct assertion here is that |num_bytes| is |1 * ...| |
| // or |2 * ...|.) |
| num_bytes = 0; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(2 * sizeof(elements[0]), num_bytes); |
| |
| // Read one element. |
| elements[0] = -1; |
| elements[1] = -1; |
| num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes)); |
| ASSERT_EQ(1u * sizeof(elements[0]), num_bytes); |
| ASSERT_EQ(123, elements[0]); |
| ASSERT_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; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_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])); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, false, true)); |
| ASSERT_EQ(1u * sizeof(elements[0]), num_bytes); |
| ASSERT_EQ(456, elements[0]); |
| ASSERT_EQ(-1, elements[1]); |
| |
| // Query. Still has 1 element remaining. |
| num_bytes = 0; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_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])); |
| ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, |
| ReadData(elements, &num_bytes, true, false)); |
| ASSERT_EQ(-1, elements[0]); |
| ASSERT_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])); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, false, false)); |
| ASSERT_EQ(1u * sizeof(elements[0]), num_bytes); |
| ASSERT_EQ(456, elements[0]); |
| ASSERT_EQ(-1, elements[1]); |
| |
| // Query. |
| num_bytes = 0; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(0u, num_bytes); |
| } |
| |
| // 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.) |
| TEST_F(DataPipeTest, 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_FLAG_NONE, // |flags|. |
| static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|. |
| 2 * sizeof(int32_t) // |capacity_num_bytes|. |
| }; |
| Create(&options); |
| MojoHandleSignalsState hss; |
| |
| // Never readable. Already writable. |
| hss = GetSignalsState(producer_); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Write two elements. |
| int32_t elements[2] = {123, 456}; |
| uint32_t num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true)); |
| ASSERT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes); |
| |
| // Wait for data to become available to the consumer. |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Peek one element. |
| elements[0] = -1; |
| elements[1] = -1; |
| num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true, true)); |
| ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes); |
| ASSERT_EQ(123, elements[0]); |
| ASSERT_EQ(-1, elements[1]); |
| |
| // Read one element. |
| elements[0] = -1; |
| elements[1] = -1; |
| num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true, false)); |
| ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes); |
| ASSERT_EQ(123, elements[0]); |
| ASSERT_EQ(-1, elements[1]); |
| |
| // Try writing, using a two-phase write. |
| void* buffer = nullptr; |
| num_bytes = static_cast<uint32_t>(3u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&buffer, &num_bytes)); |
| EXPECT_TRUE(buffer); |
| ASSERT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(elements[0]))); |
| |
| static_cast<int32_t*>(buffer)[0] = 789; |
| ASSERT_EQ(MOJO_RESULT_OK, |
| EndWriteData(static_cast<uint32_t>(1u * sizeof(elements[0])))); |
| |
| // Read one element, using a two-phase read. |
| const void* read_buffer = nullptr; |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer, &num_bytes)); |
| EXPECT_TRUE(read_buffer); |
| // The two-phase read should be able to read at least one element. |
| ASSERT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(elements[0]))); |
| ASSERT_EQ(456, static_cast<const int32_t*>(read_buffer)[0]); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| EndReadData(static_cast<uint32_t>(1u * sizeof(elements[0])))); |
| |
| // Write one element. |
| elements[0] = 123; |
| num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes)); |
| ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes); |
| |
| // Close the consumer. |
| CloseConsumer(); |
| |
| // It should now be never-writable. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(producer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss)); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals); |
| } |
| |
| TEST_F(DataPipeTest, PeerClosedProducerWaiting) { |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|. |
| 2 * sizeof(int32_t) // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| // Close the consumer. |
| CloseConsumer(); |
| |
| // It should be signaled. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(producer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss)); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals); |
| } |
| |
| TEST_F(DataPipeTest, PeerClosedConsumerWaiting) { |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|. |
| 2 * sizeof(int32_t) // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| // Close the producer. |
| CloseProducer(); |
| |
| // It should be signaled. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss)); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals); |
| } |
| |
| TEST_F(DataPipeTest, BasicConsumerWaiting) { |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|. |
| 1000 * sizeof(int32_t) // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| // Never writable. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_WRITABLE, &hss)); |
| EXPECT_EQ(0u, hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Write two elements. |
| int32_t elements[2] = {123, 456}; |
| uint32_t num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true)); |
| |
| // Wait for readability. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Discard one element. |
| num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, DiscardData(&num_bytes, true)); |
| ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes); |
| |
| // Should still be readable. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Peek one element. |
| elements[0] = -1; |
| elements[1] = -1; |
| num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true, true)); |
| ASSERT_EQ(456, elements[0]); |
| ASSERT_EQ(-1, elements[1]); |
| |
| // Should still be readable. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Read one element. |
| elements[0] = -1; |
| elements[1] = -1; |
| num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true)); |
| ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes); |
| ASSERT_EQ(456, elements[0]); |
| ASSERT_EQ(-1, elements[1]); |
| |
| // Write one element. |
| elements[0] = 789; |
| elements[1] = -1; |
| num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true)); |
| |
| // Waiting should now succeed. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Close the producer. |
| CloseProducer(); |
| |
| // Should still be readable. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_TRUE(hss.satisfied_signals & (MOJO_HANDLE_SIGNAL_READABLE | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfiable_signals); |
| |
| // Wait for the peer closed signal. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_CLOSED, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfiable_signals); |
| |
| // Read one element. |
| elements[0] = -1; |
| elements[1] = -1; |
| num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true)); |
| ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes); |
| ASSERT_EQ(789, elements[0]); |
| ASSERT_EQ(-1, elements[1]); |
| |
| // Should be never-readable. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals); |
| } |
| |
| TEST_F(DataPipeTest, ConsumerNewDataReadable) { |
| const MojoCreateDataPipeOptions create_options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_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(&create_options)); |
| |
| int32_t elements[2] = {123, 456}; |
| uint32_t num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0])); |
| EXPECT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true)); |
| |
| // The consumer handle should appear to be readable and have new data. |
| EXPECT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE)); |
| EXPECT_TRUE(GetSignalsState(consumer_).satisfied_signals & |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE); |
| |
| // Now try to read a minimum of 6 elements. |
| int32_t read_elements[6]; |
| uint32_t num_read_bytes = sizeof(read_elements); |
| MojoReadDataOptions read_options; |
| read_options.struct_size = sizeof(read_options); |
| read_options.flags = MOJO_READ_DATA_FLAG_ALL_OR_NONE; |
| EXPECT_EQ( |
| MOJO_RESULT_OUT_OF_RANGE, |
| MojoReadData(consumer_, &read_options, read_elements, &num_read_bytes)); |
| |
| // The consumer should still appear to be readable but not with new data. |
| EXPECT_TRUE(GetSignalsState(consumer_).satisfied_signals & |
| MOJO_HANDLE_SIGNAL_READABLE); |
| EXPECT_FALSE(GetSignalsState(consumer_).satisfied_signals & |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE); |
| |
| // Write four more elements. |
| EXPECT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true)); |
| EXPECT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true)); |
| |
| // The consumer handle should once again appear to be readable. |
| EXPECT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE)); |
| |
| // Try again to read a minimum of 6 elements. Should succeed this time. |
| EXPECT_EQ(MOJO_RESULT_OK, MojoReadData(consumer_, &read_options, |
| read_elements, &num_read_bytes)); |
| |
| // And now the consumer is unreadable. |
| EXPECT_FALSE(GetSignalsState(consumer_).satisfied_signals & |
| MOJO_HANDLE_SIGNAL_READABLE); |
| EXPECT_FALSE(GetSignalsState(consumer_).satisfied_signals & |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE); |
| } |
| |
| // Test with two-phase APIs and also closing the producer with an active |
| // consumer waiter. |
| TEST_F(DataPipeTest, ConsumerWaitingTwoPhase) { |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|. |
| 1000 * sizeof(int32_t) // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| // Write two elements. |
| int32_t* elements = nullptr; |
| void* buffer = nullptr; |
| // Request room for three (but we'll only write two). |
| uint32_t num_bytes = static_cast<uint32_t>(3u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&buffer, &num_bytes)); |
| EXPECT_TRUE(buffer); |
| EXPECT_GE(num_bytes, static_cast<uint32_t>(3u * sizeof(elements[0]))); |
| elements = static_cast<int32_t*>(buffer); |
| elements[0] = 123; |
| elements[1] = 456; |
| ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(2u * sizeof(elements[0]))); |
| |
| // Wait for readability. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Read one element. |
| // Two should be available, but only read one. |
| const void* read_buffer = nullptr; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer, &num_bytes)); |
| EXPECT_TRUE(read_buffer); |
| ASSERT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes); |
| const int32_t* read_elements = static_cast<const int32_t*>(read_buffer); |
| ASSERT_EQ(123, read_elements[0]); |
| ASSERT_EQ(MOJO_RESULT_OK, EndReadData(1u * sizeof(elements[0]))); |
| |
| // Should still be readable. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Read one element. |
| // Request three, but not in all-or-none mode. |
| read_buffer = nullptr; |
| num_bytes = static_cast<uint32_t>(3u * sizeof(elements[0])); |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer, &num_bytes)); |
| EXPECT_TRUE(read_buffer); |
| ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes); |
| read_elements = static_cast<const int32_t*>(read_buffer); |
| ASSERT_EQ(456, read_elements[0]); |
| ASSERT_EQ(MOJO_RESULT_OK, EndReadData(1u * sizeof(elements[0]))); |
| |
| // Close the producer. |
| CloseProducer(); |
| |
| // Should be never-readable. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals); |
| } |
| |
| // Tests that data pipes aren't writable/readable during two-phase writes/reads. |
| TEST_F(DataPipeTest, BasicTwoPhaseWaiting) { |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|. |
| 1000 * sizeof(int32_t) // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| // It should be writable. |
| hss = GetSignalsState(producer_); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| uint32_t num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t)); |
| void* write_ptr = nullptr; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_ptr, &num_bytes)); |
| EXPECT_TRUE(write_ptr); |
| EXPECT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(int32_t))); |
| |
| // At this point, it shouldn't be writable. |
| hss = GetSignalsState(producer_); |
| ASSERT_EQ(0u, hss.satisfied_signals); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // It shouldn't be readable yet either (we'll wait later). |
| hss = GetSignalsState(consumer_); |
| ASSERT_EQ(0u, hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| static_cast<int32_t*>(write_ptr)[0] = 123; |
| ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(1u * sizeof(int32_t))); |
| |
| // It should immediately be writable again. |
| hss = GetSignalsState(producer_); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // It should become readable. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Start another two-phase write and check that it's readable even in the |
| // middle of it. |
| num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t)); |
| write_ptr = nullptr; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_ptr, &num_bytes)); |
| EXPECT_TRUE(write_ptr); |
| EXPECT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(int32_t))); |
| |
| // It should be readable. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // End the two-phase write without writing anything. |
| ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(0u)); |
| |
| // Start a two-phase read. |
| num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t)); |
| const void* read_ptr = nullptr; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_ptr, &num_bytes)); |
| EXPECT_TRUE(read_ptr); |
| ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(int32_t)), num_bytes); |
| |
| // At this point, it should still be writable. |
| hss = GetSignalsState(producer_); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // But not readable. |
| hss = GetSignalsState(consumer_); |
| ASSERT_EQ(0u, hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // End the two-phase read without reading anything. |
| ASSERT_EQ(MOJO_RESULT_OK, EndReadData(0u)); |
| |
| // It should be readable again. |
| hss = GetSignalsState(consumer_); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| } |
| |
| void Seq(int32_t start, size_t count, int32_t* out) { |
| for (size_t i = 0; i < count; i++) |
| out[i] = start + static_cast<int32_t>(i); |
| } |
| |
| TEST_F(DataPipeTest, AllOrNone) { |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|. |
| 10 * sizeof(int32_t) // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| // Try writing more than the total capacity of the pipe. |
| uint32_t num_bytes = 20u * sizeof(int32_t); |
| int32_t buffer[100]; |
| Seq(0, base::size(buffer), buffer); |
| ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, WriteData(buffer, &num_bytes, true)); |
| |
| // Should still be empty. |
| num_bytes = ~0u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(0u, num_bytes); |
| |
| // Write some data. |
| num_bytes = 5u * sizeof(int32_t); |
| Seq(100, base::size(buffer), buffer); |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(buffer, &num_bytes, true)); |
| ASSERT_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. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_CLOSED | MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Half full. |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(5u * sizeof(int32_t), num_bytes); |
| |
| // Try writing more than the available capacity of the pipe, but less than the |
| // total capacity. |
| num_bytes = 6u * sizeof(int32_t); |
| Seq(200, base::size(buffer), buffer); |
| ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, WriteData(buffer, &num_bytes, true)); |
| |
| // Try reading too much. |
| num_bytes = 11u * sizeof(int32_t); |
| memset(buffer, 0xab, sizeof(buffer)); |
| ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, ReadData(buffer, &num_bytes, true)); |
| int32_t expected_buffer[100]; |
| memset(expected_buffer, 0xab, sizeof(expected_buffer)); |
| ASSERT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer))); |
| |
| // Try discarding too much. |
| num_bytes = 11u * sizeof(int32_t); |
| ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, DiscardData(&num_bytes, true)); |
| |
| // Just a little. |
| num_bytes = 2u * sizeof(int32_t); |
| Seq(300, base::size(buffer), buffer); |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(buffer, &num_bytes, true)); |
| ASSERT_EQ(2u * sizeof(int32_t), num_bytes); |
| |
| // Just right. |
| num_bytes = 3u * sizeof(int32_t); |
| Seq(400, base::size(buffer), buffer); |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(buffer, &num_bytes, true)); |
| ASSERT_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. |
| for (size_t i = 0; i < kMaxPoll; i++) { |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| if (num_bytes >= 10u * sizeof(int32_t)) |
| break; |
| |
| test::Sleep(test::EpsilonDeadline()); |
| } |
| ASSERT_EQ(10u * sizeof(int32_t), num_bytes); |
| |
| // Read half. |
| num_bytes = 5u * sizeof(int32_t); |
| memset(buffer, 0xab, sizeof(buffer)); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(buffer, &num_bytes, true)); |
| ASSERT_EQ(5u * sizeof(int32_t), num_bytes); |
| memset(expected_buffer, 0xab, sizeof(expected_buffer)); |
| Seq(100, 5, expected_buffer); |
| ASSERT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer))); |
| |
| // Try reading too much again. |
| num_bytes = 6u * sizeof(int32_t); |
| memset(buffer, 0xab, sizeof(buffer)); |
| ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, ReadData(buffer, &num_bytes, true)); |
| memset(expected_buffer, 0xab, sizeof(expected_buffer)); |
| ASSERT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer))); |
| |
| // Try discarding too much again. |
| num_bytes = 6u * sizeof(int32_t); |
| ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, DiscardData(&num_bytes, true)); |
| |
| // Discard a little. |
| num_bytes = 2u * sizeof(int32_t); |
| ASSERT_EQ(MOJO_RESULT_OK, DiscardData(&num_bytes, true)); |
| ASSERT_EQ(2u * sizeof(int32_t), num_bytes); |
| |
| // Three left. |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(3u * sizeof(int32_t), num_bytes); |
| |
| // Close the producer, then test producer-closed cases. |
| CloseProducer(); |
| |
| // Wait. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &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)); |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, |
| ReadData(buffer, &num_bytes, true)); |
| memset(expected_buffer, 0xab, sizeof(expected_buffer)); |
| ASSERT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer))); |
| |
| // Try discarding too much; "failed precondition" again. |
| num_bytes = 4u * sizeof(int32_t); |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, DiscardData(&num_bytes, true)); |
| |
| // Read a little. |
| num_bytes = 2u * sizeof(int32_t); |
| memset(buffer, 0xab, sizeof(buffer)); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(buffer, &num_bytes, true)); |
| ASSERT_EQ(2u * sizeof(int32_t), num_bytes); |
| memset(expected_buffer, 0xab, sizeof(expected_buffer)); |
| Seq(400, 2, expected_buffer); |
| ASSERT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer))); |
| |
| // Discard the remaining element. |
| num_bytes = 1u * sizeof(int32_t); |
| ASSERT_EQ(MOJO_RESULT_OK, DiscardData(&num_bytes, true)); |
| ASSERT_EQ(1u * sizeof(int32_t), num_bytes); |
| |
| // Empty again. |
| num_bytes = ~0u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(0u, num_bytes); |
| } |
| |
| // Tests that |ProducerWriteData()| and |ConsumerReadData()| writes and reads, |
| // respectively, as much as possible, even if it may have to "wrap around" the |
| // internal circular buffer. (Note that the two-phase write and read need not do |
| // this.) |
| TEST_F(DataPipeTest, WrapAround) { |
| unsigned char test_data[1000]; |
| for (size_t i = 0; i < base::size(test_data); i++) |
| test_data[i] = static_cast<unsigned char>(i); |
| |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| 1u, // |element_num_bytes|. |
| 100u // |capacity_num_bytes|. |
| }; |
| |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| // Write 20 bytes. |
| uint32_t num_bytes = 20u; |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(&test_data[0], &num_bytes, true)); |
| ASSERT_EQ(20u, num_bytes); |
| |
| // Wait for data. |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_TRUE(hss.satisfied_signals & MOJO_HANDLE_SIGNAL_READABLE); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Read 10 bytes. |
| unsigned char read_buffer[1000] = {0}; |
| num_bytes = 10u; |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(read_buffer, &num_bytes, true)); |
| ASSERT_EQ(10u, num_bytes); |
| ASSERT_EQ(0, memcmp(read_buffer, &test_data[0], 10u)); |
| |
| // Check that a two-phase write can now only write (at most) 80 bytes. (This |
| // checks an implementation detail; this behavior is not guaranteed.) |
| void* write_buffer_ptr = nullptr; |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_buffer_ptr, &num_bytes)); |
| EXPECT_TRUE(write_buffer_ptr); |
| ASSERT_EQ(80u, num_bytes); |
| ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(0)); |
| |
| size_t total_num_bytes = 0; |
| while (total_num_bytes < 90) { |
| // Wait to write. |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(producer_, MOJO_HANDLE_SIGNAL_WRITABLE, &hss)); |
| ASSERT_EQ(hss.satisfied_signals, MOJO_HANDLE_SIGNAL_WRITABLE); |
| ASSERT_EQ(hss.satisfiable_signals, MOJO_HANDLE_SIGNAL_WRITABLE | |
| MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE); |
| |
| // Write as much as we can. |
| num_bytes = 100; |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WriteData(&test_data[20 + total_num_bytes], &num_bytes, false)); |
| total_num_bytes += num_bytes; |
| } |
| |
| ASSERT_EQ(90u, total_num_bytes); |
| |
| num_bytes = 0; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(100u, num_bytes); |
| |
| // Check that a two-phase read can now only read (at most) 90 bytes. (This |
| // checks an implementation detail; this behavior is not guaranteed.) |
| const void* read_buffer_ptr = nullptr; |
| num_bytes = 0; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer_ptr, &num_bytes)); |
| EXPECT_TRUE(read_buffer_ptr); |
| ASSERT_EQ(90u, num_bytes); |
| ASSERT_EQ(MOJO_RESULT_OK, EndReadData(0)); |
| |
| // Read as much as possible. We should read 100 bytes. |
| num_bytes = |
| static_cast<uint32_t>(base::size(read_buffer) * sizeof(read_buffer[0])); |
| memset(read_buffer, 0, num_bytes); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(read_buffer, &num_bytes)); |
| ASSERT_EQ(100u, num_bytes); |
| ASSERT_EQ(0, memcmp(read_buffer, &test_data[10], 100u)); |
| } |
| |
| // Tests the behavior of writing (simple and two-phase), closing the producer, |
| // then reading (simple and two-phase). |
| TEST_F(DataPipeTest, WriteCloseProducerRead) { |
| const char kTestData[] = "hello world"; |
| const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData)); |
| |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| 1u, // |element_num_bytes|. |
| 1000u // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| |
| // Write some data, so we'll have something to read. |
| uint32_t num_bytes = kTestDataSize; |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes, false)); |
| ASSERT_EQ(kTestDataSize, num_bytes); |
| |
| // Write it again, so we'll have something left over. |
| num_bytes = kTestDataSize; |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes, false)); |
| ASSERT_EQ(kTestDataSize, num_bytes); |
| |
| // Start two-phase write. |
| void* write_buffer_ptr = nullptr; |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_buffer_ptr, &num_bytes)); |
| EXPECT_TRUE(write_buffer_ptr); |
| EXPECT_GT(num_bytes, 0u); |
| |
| // TODO(vtl): (See corresponding TODO in TwoPhaseAllOrNone.) |
| for (size_t i = 0; i < kMaxPoll; i++) { |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| if (num_bytes >= 2u * kTestDataSize) |
| break; |
| |
| test::Sleep(test::EpsilonDeadline()); |
| } |
| ASSERT_EQ(2u * kTestDataSize, num_bytes); |
| |
| // Start two-phase read. |
| const void* read_buffer_ptr = nullptr; |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer_ptr, &num_bytes)); |
| EXPECT_TRUE(read_buffer_ptr); |
| ASSERT_EQ(2u * kTestDataSize, num_bytes); |
| |
| // Close the producer. |
| CloseProducer(); |
| |
| // The consumer can finish its two-phase read. |
| ASSERT_EQ(0, memcmp(read_buffer_ptr, kTestData, kTestDataSize)); |
| ASSERT_EQ(MOJO_RESULT_OK, EndReadData(kTestDataSize)); |
| |
| // And start another. |
| read_buffer_ptr = nullptr; |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer_ptr, &num_bytes)); |
| EXPECT_TRUE(read_buffer_ptr); |
| ASSERT_EQ(kTestDataSize, num_bytes); |
| } |
| |
| // Tests the behavior of interrupting a two-phase read and write by closing the |
| // consumer. |
| TEST_F(DataPipeTest, TwoPhaseWriteReadCloseConsumer) { |
| const char kTestData[] = "hello world"; |
| const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData)); |
| |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| 1u, // |element_num_bytes|. |
| 1000u // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| // Write some data, so we'll have something to read. |
| uint32_t num_bytes = kTestDataSize; |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes)); |
| ASSERT_EQ(kTestDataSize, num_bytes); |
| |
| // Start two-phase write. |
| void* write_buffer_ptr = nullptr; |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_buffer_ptr, &num_bytes)); |
| EXPECT_TRUE(write_buffer_ptr); |
| ASSERT_GT(num_bytes, kTestDataSize); |
| |
| // Wait for data. |
| // TODO(vtl): (See corresponding TODO in AllOrNone.) |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Start two-phase read. |
| const void* read_buffer_ptr = nullptr; |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer_ptr, &num_bytes)); |
| EXPECT_TRUE(read_buffer_ptr); |
| ASSERT_EQ(kTestDataSize, num_bytes); |
| |
| // Close the consumer. |
| CloseConsumer(); |
| |
| // Wait for producer to know that the consumer is closed. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(producer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss)); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals); |
| |
| // Actually write some data. (Note: Premature freeing of the buffer would |
| // probably only be detected under ASAN or similar.) |
| memcpy(write_buffer_ptr, kTestData, kTestDataSize); |
| // Note: Even though the consumer has been closed, ending the two-phase |
| // write will report success. |
| ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(kTestDataSize)); |
| |
| // But trying to write should result in failure. |
| num_bytes = kTestDataSize; |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, WriteData(kTestData, &num_bytes)); |
| |
| // As will trying to start another two-phase write. |
| write_buffer_ptr = nullptr; |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, |
| BeginWriteData(&write_buffer_ptr, &num_bytes)); |
| } |
| |
| // Tests the behavior of "interrupting" a two-phase write by closing both the |
| // producer and the consumer. |
| TEST_F(DataPipeTest, TwoPhaseWriteCloseBoth) { |
| const uint32_t kTestDataSize = 15u; |
| |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| 1u, // |element_num_bytes|. |
| 1000u // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| |
| // Start two-phase write. |
| void* write_buffer_ptr = nullptr; |
| uint32_t num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_buffer_ptr, &num_bytes)); |
| EXPECT_TRUE(write_buffer_ptr); |
| ASSERT_GT(num_bytes, kTestDataSize); |
| } |
| |
| // Tests the behavior of writing, closing the producer, and then reading (with |
| // and without data remaining). |
| TEST_F(DataPipeTest, WriteCloseProducerReadNoData) { |
| const char kTestData[] = "hello world"; |
| const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData)); |
| |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| 1u, // |element_num_bytes|. |
| 1000u // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| // Write some data, so we'll have something to read. |
| uint32_t num_bytes = kTestDataSize; |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes)); |
| ASSERT_EQ(kTestDataSize, num_bytes); |
| |
| // Close the producer. |
| CloseProducer(); |
| |
| // Wait. (Note that once the consumer knows that the producer is closed, it |
| // must also know about all the data that was sent.) |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfiable_signals); |
| |
| // Peek that data. |
| char buffer[1000]; |
| num_bytes = static_cast<uint32_t>(sizeof(buffer)); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(buffer, &num_bytes, false, true)); |
| ASSERT_EQ(kTestDataSize, num_bytes); |
| ASSERT_EQ(0, memcmp(buffer, kTestData, kTestDataSize)); |
| |
| // Read that data. |
| memset(buffer, 0, 1000); |
| num_bytes = static_cast<uint32_t>(sizeof(buffer)); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(buffer, &num_bytes)); |
| ASSERT_EQ(kTestDataSize, num_bytes); |
| ASSERT_EQ(0, memcmp(buffer, kTestData, kTestDataSize)); |
| |
| // A second read should fail. |
| num_bytes = static_cast<uint32_t>(sizeof(buffer)); |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, ReadData(buffer, &num_bytes)); |
| |
| // A two-phase read should also fail. |
| const void* read_buffer_ptr = nullptr; |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, |
| BeginReadData(&read_buffer_ptr, &num_bytes)); |
| |
| // Ditto for discard. |
| num_bytes = 10u; |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, DiscardData(&num_bytes)); |
| } |
| |
| // Test that during a two phase read the memory stays valid even if more data |
| // comes in. |
| TEST_F(DataPipeTest, TwoPhaseReadMemoryStable) { |
| const char kTestData[] = "hello world"; |
| const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData)); |
| |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| 1u, // |element_num_bytes|. |
| 1000u // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| // Write some data. |
| uint32_t num_bytes = kTestDataSize; |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes)); |
| ASSERT_EQ(kTestDataSize, num_bytes); |
| |
| // Wait for the data. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Begin a two-phase read. |
| const void* read_buffer_ptr = nullptr; |
| uint32_t read_buffer_size = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer_ptr, &read_buffer_size)); |
| |
| // Write more data. |
| const char kExtraData[] = "bye world"; |
| const uint32_t kExtraDataSize = static_cast<uint32_t>(sizeof(kExtraData)); |
| num_bytes = kExtraDataSize; |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(kExtraData, &num_bytes)); |
| ASSERT_EQ(kExtraDataSize, num_bytes); |
| |
| // Close the producer. |
| CloseProducer(); |
| |
| // Wait. (Note that once the consumer knows that the producer is closed, it |
| // must also have received the extra data). |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfiable_signals); |
| |
| // Read the two phase memory to check it's still valid. |
| ASSERT_EQ(0, memcmp(read_buffer_ptr, kTestData, kTestDataSize)); |
| EndReadData(read_buffer_size); |
| } |
| |
| // Test that two-phase reads/writes behave correctly when given invalid |
| // arguments. |
| TEST_F(DataPipeTest, TwoPhaseMoreInvalidArguments) { |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|. |
| 10 * sizeof(int32_t) // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| // No data. |
| uint32_t num_bytes = 1000u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(0u, num_bytes); |
| |
| // Try "ending" a two-phase write when one isn't active. |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, |
| EndWriteData(1u * sizeof(int32_t))); |
| |
| // Wait a bit, to make sure that if a signal were (incorrectly) sent, it'd |
| // have time to propagate. |
| test::Sleep(test::EpsilonDeadline()); |
| |
| // Still no data. |
| num_bytes = 1000u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(0u, num_bytes); |
| |
| // Try ending a two-phase write with an invalid amount (too much). |
| num_bytes = 0u; |
| void* write_ptr = nullptr; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_ptr, &num_bytes)); |
| ASSERT_EQ(MOJO_RESULT_INVALID_ARGUMENT, |
| EndWriteData(num_bytes + static_cast<uint32_t>(sizeof(int32_t)))); |
| |
| // But the two-phase write still ended. |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, EndWriteData(0u)); |
| |
| // Wait a bit (as above). |
| test::Sleep(test::EpsilonDeadline()); |
| |
| // Still no data. |
| num_bytes = 1000u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(0u, num_bytes); |
| |
| // Try ending a two-phase write with an invalid amount (not a multiple of the |
| // element size). |
| num_bytes = 0u; |
| write_ptr = nullptr; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_ptr, &num_bytes)); |
| EXPECT_GE(num_bytes, 1u); |
| ASSERT_EQ(MOJO_RESULT_INVALID_ARGUMENT, EndWriteData(1u)); |
| |
| // But the two-phase write still ended. |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, EndWriteData(0u)); |
| |
| // Wait a bit (as above). |
| test::Sleep(test::EpsilonDeadline()); |
| |
| // Still no data. |
| num_bytes = 1000u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(0u, num_bytes); |
| |
| // Now write some data, so we'll be able to try reading. |
| int32_t element = 123; |
| num_bytes = 1u * sizeof(int32_t); |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(&element, &num_bytes)); |
| |
| // Wait for data. |
| // TODO(vtl): (See corresponding TODO in AllOrNone.) |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // One element available. |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(1u * sizeof(int32_t), num_bytes); |
| |
| // Try "ending" a two-phase read when one isn't active. |
| ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, EndReadData(1u * sizeof(int32_t))); |
| |
| // Still one element available. |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(1u * sizeof(int32_t), num_bytes); |
| |
| // Try ending a two-phase read with an invalid amount (too much). |
| num_bytes = 0u; |
| const void* read_ptr = nullptr; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_ptr, &num_bytes)); |
| ASSERT_EQ(MOJO_RESULT_INVALID_ARGUMENT, |
| EndReadData(num_bytes + static_cast<uint32_t>(sizeof(int32_t)))); |
| |
| // Still one element available. |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(1u * sizeof(int32_t), num_bytes); |
| |
| // Try ending a two-phase read with an invalid amount (not a multiple of the |
| // element size). |
| num_bytes = 0u; |
| read_ptr = nullptr; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_ptr, &num_bytes)); |
| ASSERT_EQ(1u * sizeof(int32_t), num_bytes); |
| ASSERT_EQ(123, static_cast<const int32_t*>(read_ptr)[0]); |
| ASSERT_EQ(MOJO_RESULT_INVALID_ARGUMENT, EndReadData(1u)); |
| |
| // Still one element available. |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes)); |
| ASSERT_EQ(1u * sizeof(int32_t), num_bytes); |
| } |
| |
| // Test that a producer can be sent over a MP. |
| TEST_F(DataPipeTest, SendProducer) { |
| const char kTestData[] = "hello world"; |
| const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData)); |
| |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| 1u, // |element_num_bytes|. |
| 1000u // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| MojoHandleSignalsState hss; |
| |
| // Write some data. |
| uint32_t num_bytes = kTestDataSize; |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes)); |
| ASSERT_EQ(kTestDataSize, num_bytes); |
| |
| // Wait for the data. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Check the data. |
| const void* read_buffer = nullptr; |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer, &num_bytes)); |
| ASSERT_EQ(0, memcmp(read_buffer, kTestData, kTestDataSize)); |
| EndReadData(num_bytes); |
| |
| // Now send the producer over a MP so that it's serialized. |
| MojoHandle pipe0, pipe1; |
| ASSERT_EQ(MOJO_RESULT_OK, MojoCreateMessagePipe(nullptr, &pipe0, &pipe1)); |
| |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WriteMessageRaw(MessagePipeHandle(pipe0), nullptr, 0, &producer_, 1, |
| MOJO_WRITE_MESSAGE_FLAG_NONE)); |
| producer_ = MOJO_HANDLE_INVALID; |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(pipe1, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadEmptyMessageWithHandles(pipe1, &producer_, 1)); |
| |
| // Write more data. |
| const char kExtraData[] = "bye world"; |
| const uint32_t kExtraDataSize = static_cast<uint32_t>(sizeof(kExtraData)); |
| num_bytes = kExtraDataSize; |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(kExtraData, &num_bytes)); |
| ASSERT_EQ(kExtraDataSize, num_bytes); |
| |
| // Wait for it. |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| hss.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| |
| // Check the second write. |
| num_bytes = 0u; |
| ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer, &num_bytes)); |
| ASSERT_EQ(0, memcmp(read_buffer, kExtraData, kExtraDataSize)); |
| EndReadData(num_bytes); |
| |
| ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe0)); |
| ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe1)); |
| } |
| |
| // Ensures that if a data pipe consumer whose producer has closed is passed over |
| // a message pipe, the deserialized dispatcher is also marked as having a closed |
| // peer. |
| TEST_F(DataPipeTest, ConsumerWithClosedProducerSent) { |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|. |
| 1000 * sizeof(int32_t) // |capacity_num_bytes|. |
| }; |
| |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| |
| // We can write to a data pipe handle immediately. |
| int32_t data = 123; |
| uint32_t num_bytes = sizeof(data); |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(&data, &num_bytes)); |
| ASSERT_EQ(MOJO_RESULT_OK, CloseProducer()); |
| |
| // Now wait for the other side to become readable and to see the peer closed. |
| MojoHandleSignalsState state; |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &state)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| state.satisfied_signals); |
| ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| state.satisfiable_signals); |
| |
| // Now send the consumer over a MP so that it's serialized. |
| MojoHandle pipe0, pipe1; |
| ASSERT_EQ(MOJO_RESULT_OK, MojoCreateMessagePipe(nullptr, &pipe0, &pipe1)); |
| |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WriteMessageRaw(MessagePipeHandle(pipe0), nullptr, 0, &consumer_, 1, |
| MOJO_WRITE_MESSAGE_FLAG_NONE)); |
| consumer_ = MOJO_HANDLE_INVALID; |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(pipe1, MOJO_HANDLE_SIGNAL_READABLE, &state)); |
| ASSERT_EQ(MOJO_RESULT_OK, ReadEmptyMessageWithHandles(pipe1, &consumer_, 1)); |
| |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &state)); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| state.satisfied_signals); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE, |
| state.satisfiable_signals); |
| |
| int32_t read_data; |
| ASSERT_EQ(MOJO_RESULT_OK, ReadData(&read_data, &num_bytes)); |
| ASSERT_EQ(sizeof(read_data), num_bytes); |
| ASSERT_EQ(data, read_data); |
| |
| ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe0)); |
| ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe1)); |
| } |
| |
| bool WriteAllData(MojoHandle producer, |
| const void* elements, |
| uint32_t num_bytes) { |
| for (size_t i = 0; i < kMaxPoll; i++) { |
| // Write as much data as we can. |
| uint32_t write_bytes = num_bytes; |
| MojoResult result = |
| MojoWriteData(producer, elements, &write_bytes, nullptr); |
| if (result == MOJO_RESULT_OK) { |
| num_bytes -= write_bytes; |
| elements = static_cast<const uint8_t*>(elements) + write_bytes; |
| if (num_bytes == 0) |
| return true; |
| } else { |
| EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT, result); |
| } |
| |
| MojoHandleSignalsState hss = MojoHandleSignalsState(); |
| EXPECT_EQ(MOJO_RESULT_OK, test::MojoTestBase::WaitForSignals( |
| producer, MOJO_HANDLE_SIGNAL_WRITABLE, &hss)); |
| EXPECT_TRUE(hss.satisfied_signals & MOJO_HANDLE_SIGNAL_WRITABLE); |
| EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | |
| MOJO_HANDLE_SIGNAL_PEER_REMOTE, |
| hss.satisfiable_signals); |
| } |
| |
| return false; |
| } |
| |
| // If |expect_empty| is true, expect |consumer| to be empty after reading. |
| bool ReadAllData(MojoHandle consumer, |
| void* elements, |
| uint32_t num_bytes, |
| bool expect_empty) { |
| for (size_t i = 0; i < kMaxPoll; i++) { |
| // Read as much data as we can. |
| uint32_t read_bytes = num_bytes; |
| MojoResult result = MojoReadData(consumer, nullptr, elements, &read_bytes); |
| if (result == MOJO_RESULT_OK) { |
| num_bytes -= read_bytes; |
| elements = static_cast<uint8_t*>(elements) + read_bytes; |
| if (num_bytes == 0) { |
| if (expect_empty) { |
| // Expect no more data. |
| test::Sleep(test::TinyDeadline()); |
| MojoReadDataOptions options; |
| options.struct_size = sizeof(options); |
| options.flags = MOJO_READ_DATA_FLAG_QUERY; |
| MojoReadData(consumer, &options, nullptr, &num_bytes); |
| EXPECT_EQ(0u, num_bytes); |
| } |
| return true; |
| } |
| } else { |
| EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT, result); |
| } |
| |
| MojoHandleSignalsState hss = MojoHandleSignalsState(); |
| EXPECT_EQ(MOJO_RESULT_OK, test::MojoTestBase::WaitForSignals( |
| consumer, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| // Peer could have become closed while we're still waiting for data. |
| EXPECT_TRUE(MOJO_HANDLE_SIGNAL_READABLE & hss.satisfied_signals); |
| EXPECT_TRUE(hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_READABLE); |
| EXPECT_TRUE(hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_PEER_CLOSED); |
| } |
| |
| return num_bytes == 0; |
| } |
| |
| #if !defined(OS_IOS) |
| |
| TEST_F(DataPipeTest, Multiprocess) { |
| const uint32_t kTestDataSize = |
| static_cast<uint32_t>(sizeof(kMultiprocessTestData)); |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| 1, // |element_num_bytes|. |
| kMultiprocessCapacity // |capacity_num_bytes|. |
| }; |
| ASSERT_EQ(MOJO_RESULT_OK, Create(&options)); |
| |
| RunTestClient("MultiprocessClient", [&](MojoHandle server_mp) { |
| // Send some data before serialising and sending the data pipe over. |
| // This is the first write so we don't need to use WriteAllData. |
| uint32_t num_bytes = kTestDataSize; |
| ASSERT_EQ(MOJO_RESULT_OK, WriteData(kMultiprocessTestData, &num_bytes, |
| MOJO_WRITE_DATA_FLAG_ALL_OR_NONE)); |
| ASSERT_EQ(kTestDataSize, num_bytes); |
| |
| // Send child process the data pipe. |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WriteMessageRaw(MessagePipeHandle(server_mp), nullptr, 0, |
| &consumer_, 1, MOJO_WRITE_MESSAGE_FLAG_NONE)); |
| |
| // Send a bunch of data of varying sizes. |
| uint8_t buffer[100]; |
| int seq = 0; |
| for (int i = 0; i < kMultiprocessMaxIter; ++i) { |
| for (uint32_t size = 1; size <= kMultiprocessCapacity; size++) { |
| for (unsigned int j = 0; j < size; ++j) |
| buffer[j] = seq + j; |
| EXPECT_TRUE(WriteAllData(producer_, buffer, size)); |
| seq += size; |
| } |
| } |
| |
| // Write the test string in again. |
| ASSERT_TRUE(WriteAllData(producer_, kMultiprocessTestData, kTestDataSize)); |
| |
| // Swap ends. |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WriteMessageRaw(MessagePipeHandle(server_mp), nullptr, 0, |
| &producer_, 1, MOJO_WRITE_MESSAGE_FLAG_NONE)); |
| |
| // Receive the consumer from the other side. |
| producer_ = MOJO_HANDLE_INVALID; |
| MojoHandleSignalsState hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(server_mp, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| ReadEmptyMessageWithHandles(server_mp, &consumer_, 1)); |
| |
| // Read the test string twice. Once for when we sent it, and once for the |
| // other end sending it. |
| for (int i = 0; i < 2; ++i) { |
| EXPECT_TRUE(ReadAllData(consumer_, buffer, kTestDataSize, i == 1)); |
| EXPECT_EQ(0, memcmp(buffer, kMultiprocessTestData, kTestDataSize)); |
| } |
| |
| WriteMessage(server_mp, "quit"); |
| |
| // Don't have to close the consumer here because it will be done for us. |
| }); |
| } |
| |
| DEFINE_TEST_CLIENT_TEST_WITH_PIPE(MultiprocessClient, DataPipeTest, client_mp) { |
| const uint32_t kTestDataSize = |
| static_cast<uint32_t>(sizeof(kMultiprocessTestData)); |
| |
| // Receive the data pipe from the other side. |
| MojoHandle consumer = MOJO_HANDLE_INVALID; |
| MojoHandleSignalsState hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(client_mp, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| ReadEmptyMessageWithHandles(client_mp, &consumer, 1)); |
| |
| // Read the initial string that was sent. |
| int32_t buffer[100]; |
| EXPECT_TRUE(ReadAllData(consumer, buffer, kTestDataSize, false)); |
| EXPECT_EQ(0, memcmp(buffer, kMultiprocessTestData, kTestDataSize)); |
| |
| // Receive the main data and check it is correct. |
| int seq = 0; |
| uint8_t expected_buffer[100]; |
| for (int i = 0; i < kMultiprocessMaxIter; ++i) { |
| for (uint32_t size = 1; size <= kMultiprocessCapacity; ++size) { |
| for (unsigned int j = 0; j < size; ++j) |
| expected_buffer[j] = seq + j; |
| EXPECT_TRUE(ReadAllData(consumer, buffer, size, false)); |
| EXPECT_EQ(0, memcmp(buffer, expected_buffer, size)); |
| |
| seq += size; |
| } |
| } |
| |
| // Swap ends. |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WriteMessageRaw(MessagePipeHandle(client_mp), nullptr, 0, &consumer, |
| 1, MOJO_WRITE_MESSAGE_FLAG_NONE)); |
| |
| // Receive the producer from the other side. |
| MojoHandle producer = MOJO_HANDLE_INVALID; |
| hss = MojoHandleSignalsState(); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(client_mp, MOJO_HANDLE_SIGNAL_READABLE, &hss)); |
| ASSERT_EQ(MOJO_RESULT_OK, |
| ReadEmptyMessageWithHandles(client_mp, &producer, 1)); |
| |
| // Write the test string one more time. |
| EXPECT_TRUE(WriteAllData(producer, kMultiprocessTestData, kTestDataSize)); |
| |
| // We swapped ends, so close the producer. |
| EXPECT_EQ(MOJO_RESULT_OK, MojoClose(producer)); |
| |
| // Wait to receive a "quit" message before exiting. |
| EXPECT_EQ("quit", ReadMessage(client_mp)); |
| } |
| |
| DEFINE_TEST_CLIENT_TEST_WITH_PIPE(WriteAndCloseProducer, DataPipeTest, h) { |
| MojoHandle p; |
| std::string message = ReadMessageWithHandles(h, &p, 1); |
| |
| // Write some data to the producer and close it. |
| uint32_t num_bytes = static_cast<uint32_t>(message.size()); |
| EXPECT_EQ(MOJO_RESULT_OK, |
| MojoWriteData(p, message.data(), &num_bytes, nullptr)); |
| EXPECT_EQ(num_bytes, static_cast<uint32_t>(message.size())); |
| |
| // Close the producer before quitting. |
| EXPECT_EQ(MOJO_RESULT_OK, MojoClose(p)); |
| |
| // Wait for a quit message. |
| EXPECT_EQ("quit", ReadMessage(h)); |
| } |
| |
| DEFINE_TEST_CLIENT_TEST_WITH_PIPE(ReadAndCloseConsumer, DataPipeTest, h) { |
| MojoHandle c; |
| std::string expected_message = ReadMessageWithHandles(h, &c, 1); |
| |
| // Wait for the consumer to become readable. |
| EXPECT_EQ(MOJO_RESULT_OK, WaitForSignals(c, MOJO_HANDLE_SIGNAL_READABLE)); |
| |
| // Drain the consumer and expect to find the given message. |
| uint32_t num_bytes = static_cast<uint32_t>(expected_message.size()); |
| std::vector<char> bytes(expected_message.size()); |
| EXPECT_EQ(MOJO_RESULT_OK, MojoReadData(c, nullptr, bytes.data(), &num_bytes)); |
| EXPECT_EQ(num_bytes, static_cast<uint32_t>(bytes.size())); |
| |
| std::string message(bytes.data(), bytes.size()); |
| EXPECT_EQ(expected_message, message); |
| |
| EXPECT_EQ(MOJO_RESULT_OK, MojoClose(c)); |
| |
| // Wait for a quit message. |
| EXPECT_EQ("quit", ReadMessage(h)); |
| } |
| |
| TEST_F(DataPipeTest, SendConsumerAndCloseProducer) { |
| // Create a new data pipe. |
| MojoHandle p, c; |
| EXPECT_EQ(MOJO_RESULT_OK, MojoCreateDataPipe(nullptr, &p, &c)); |
| |
| RunTestClient("WriteAndCloseProducer", [&](MojoHandle producer_client) { |
| RunTestClient("ReadAndCloseConsumer", [&](MojoHandle consumer_client) { |
| const std::string kMessage = "Hello, world!"; |
| WriteMessageWithHandles(producer_client, kMessage, &p, 1); |
| WriteMessageWithHandles(consumer_client, kMessage, &c, 1); |
| |
| WriteMessage(consumer_client, "quit"); |
| }); |
| |
| WriteMessage(producer_client, "quit"); |
| }); |
| } |
| |
| DEFINE_TEST_CLIENT_TEST_WITH_PIPE(CreateAndWrite, DataPipeTest, h) { |
| const MojoCreateDataPipeOptions options = { |
| kSizeOfOptions, // |struct_size|. |
| MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|. |
| 1, // |element_num_bytes|. |
| kMultiprocessCapacity // |capacity_num_bytes|. |
| }; |
| |
| MojoHandle p, c; |
| ASSERT_EQ(MOJO_RESULT_OK, MojoCreateDataPipe(&options, &p, &c)); |
| |
| const std::string kMessage = "Hello, world!"; |
| WriteMessageWithHandles(h, kMessage, &c, 1); |
| |
| // Write some data to the producer and close it. |
| uint32_t num_bytes = static_cast<uint32_t>(kMessage.size()); |
| EXPECT_EQ(MOJO_RESULT_OK, |
| MojoWriteData(p, kMessage.data(), &num_bytes, nullptr)); |
| EXPECT_EQ(num_bytes, static_cast<uint32_t>(kMessage.size())); |
| EXPECT_EQ(MOJO_RESULT_OK, MojoClose(p)); |
| |
| // Wait for a quit message. |
| EXPECT_EQ("quit", ReadMessage(h)); |
| } |
| |
| TEST_F(DataPipeTest, CreateInChild) { |
| RunTestClient("CreateAndWrite", [&](MojoHandle child) { |
| MojoHandle c; |
| std::string expected_message = ReadMessageWithHandles(child, &c, 1); |
| |
| // Wait for the consumer to become readable. |
| EXPECT_EQ(MOJO_RESULT_OK, WaitForSignals(c, MOJO_HANDLE_SIGNAL_READABLE)); |
| |
| // Drain the consumer and expect to find the given message. |
| uint32_t num_bytes = static_cast<uint32_t>(expected_message.size()); |
| std::vector<char> bytes(expected_message.size()); |
| EXPECT_EQ(MOJO_RESULT_OK, |
| MojoReadData(c, nullptr, bytes.data(), &num_bytes)); |
| EXPECT_EQ(num_bytes, static_cast<uint32_t>(bytes.size())); |
| |
| std::string message(bytes.data(), bytes.size()); |
| EXPECT_EQ(expected_message, message); |
| |
| EXPECT_EQ(MOJO_RESULT_OK, MojoClose(c)); |
| WriteMessage(child, "quit"); |
| }); |
| } |
| |
| DEFINE_TEST_CLIENT_TEST_WITH_PIPE(DataPipeStatusChangeInTransitClient, |
| DataPipeTest, |
| parent) { |
| // This test verifies that peer closure is detectable through various |
| // mechanisms when it races with handle transfer. |
| |
| MojoHandle handles[6]; |
| EXPECT_EQ("o_O", ReadMessageWithHandles(parent, handles, 6)); |
| MojoHandle* producers = &handles[0]; |
| MojoHandle* consumers = &handles[3]; |
| |
| // Wait on producer 0 |
| EXPECT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(producers[0], MOJO_HANDLE_SIGNAL_PEER_CLOSED)); |
| |
| // Wait on consumer 0 |
| EXPECT_EQ(MOJO_RESULT_OK, |
| WaitForSignals(consumers[0], MOJO_HANDLE_SIGNAL_PEER_CLOSED)); |
| |
| base::MessageLoop message_loop; |
| |
| // Wait on producer 1 and consumer 1 using SimpleWatchers. |
| { |
| base::RunLoop run_loop; |
| int count = 0; |
| auto callback = base::Bind( |
| [](base::RunLoop* loop, int* count, MojoResult result) { |
| EXPECT_EQ(MOJO_RESULT_OK, result); |
| if (++*count == 2) |
| loop->Quit(); |
| }, |
| &run_loop, &count); |
| SimpleWatcher producer_watcher(FROM_HERE, |
| SimpleWatcher::ArmingPolicy::AUTOMATIC, |
| base::SequencedTaskRunnerHandle::Get()); |
| SimpleWatcher consumer_watcher(FROM_HERE, |
| SimpleWatcher::ArmingPolicy::AUTOMATIC, |
| base::SequencedTaskRunnerHandle::Get()); |
| 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. |
| MojoResult result; |
| do { |
| uint32_t num_bytes = 0; |
| result = MojoWriteData(producers[2], nullptr, &num_bytes, nullptr); |
| } while (result == MOJO_RESULT_OK); |
| EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, result); |
| |
| // Wait on consumer 2 by polling with MojoReadData. |
| do { |
| char byte; |
| uint32_t num_bytes = 1; |
| result = MojoReadData(consumers[2], nullptr, &byte, &num_bytes); |
| } while (result == MOJO_RESULT_SHOULD_WAIT); |
| EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, result); |
| |
| for (size_t i = 0; i < 6; ++i) |
| CloseHandle(handles[i]); |
| } |
| |
| TEST_F(DataPipeTest, StatusChangeInTransit) { |
| MojoHandle producers[6]; |
| MojoHandle consumers[6]; |
| for (size_t i = 0; i < 6; ++i) |
| CreateDataPipe(&producers[i], &consumers[i], 1); |
| |
| RunTestClient("DataPipeStatusChangeInTransitClient", [&](MojoHandle child) { |
| MojoHandle handles[] = {producers[0], producers[1], producers[2], |
| consumers[3], consumers[4], consumers[5]}; |
| |
| // Send 3 producers and 3 consumers, and let their transfer race with their |
| // peers' closure. |
| WriteMessageWithHandles(child, "o_O", handles, 6); |
| |
| for (size_t i = 0; i < 3; ++i) |
| CloseHandle(consumers[i]); |
| for (size_t i = 3; i < 6; ++i) |
| CloseHandle(producers[i]); |
| }); |
| } |
| |
| #endif // !defined(OS_IOS) |
| |
| } // namespace |
| } // namespace core |
| } // namespace mojo |