// Copyright 2013 The Chromium Authors. All rights reserved. // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. #include <stdint.h> #include <string.h> #include <algorithm> #include <memory> #include <string> #include <vector> #include "base/memory/ptr_util.h" #include "base/memory/ref_counted.h" #include "build/build_config.h" #include "mojo/core/test/mojo_test_base.h" #include "mojo/core/test_utils.h" #include "mojo/public/c/system/core.h" #include "mojo/public/c/system/types.h" #include "mojo/public/cpp/system/message_pipe.h" namespace mojo { namespace core { namespace { const MojoHandleSignals kAllSignals = MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED | MOJO_HANDLE_SIGNAL_PEER_REMOTE | MOJO_HANDLE_SIGNAL_QUOTA_EXCEEDED; static const char kHelloWorld[] = "hello world"; class MessagePipeTest : public test::MojoTestBase { public: MessagePipeTest() { CHECK_EQ(MOJO_RESULT_OK, MojoCreateMessagePipe(nullptr, &pipe0_, &pipe1_)); } ~MessagePipeTest() override { if (pipe0_ != MOJO_HANDLE_INVALID) CHECK_EQ(MOJO_RESULT_OK, MojoClose(pipe0_)); if (pipe1_ != MOJO_HANDLE_INVALID) CHECK_EQ(MOJO_RESULT_OK, MojoClose(pipe1_)); } MojoResult WriteMessage(MojoHandle message_pipe_handle, const void* bytes, uint32_t num_bytes) { return mojo::WriteMessageRaw(MessagePipeHandle(message_pipe_handle), bytes, num_bytes, nullptr, 0, MOJO_WRITE_MESSAGE_FLAG_NONE); } MojoResult ReadMessage(MojoHandle message_pipe_handle, void* bytes, uint32_t* num_bytes, bool may_discard = false) { MojoMessageHandle message_handle; MojoResult rv = MojoReadMessage(message_pipe_handle, nullptr, &message_handle); if (rv != MOJO_RESULT_OK) return rv; const uint32_t expected_num_bytes = *num_bytes; void* buffer; rv = MojoGetMessageData(message_handle, nullptr, &buffer, num_bytes, nullptr, nullptr); if (rv == MOJO_RESULT_RESOURCE_EXHAUSTED) { CHECK(may_discard); } else if (*num_bytes) { CHECK_EQ(MOJO_RESULT_OK, rv); CHECK_GE(expected_num_bytes, *num_bytes); CHECK(bytes); memcpy(bytes, buffer, *num_bytes); } CHECK_EQ(MOJO_RESULT_OK, MojoDestroyMessage(message_handle)); return rv; } MojoHandle pipe0_, pipe1_; private: DISALLOW_COPY_AND_ASSIGN(MessagePipeTest); }; using FuseMessagePipeTest = test::MojoTestBase; TEST_F(MessagePipeTest, WriteData) { ASSERT_EQ(MOJO_RESULT_OK, WriteMessage(pipe0_, kHelloWorld, sizeof(kHelloWorld))); } // Tests: // - only default flags // - reading messages from a port // - when there are no/one/two messages available for that port // - with buffer size 0 (and null buffer) -- should get size // - with too-small buffer -- should get size // - also verify that buffers aren't modified when/where they shouldn't be // - writing messages to a port // - in the obvious scenarios (as above) // - to a port that's been closed // - writing a message to a port, closing the other (would be the source) port, // and reading it TEST_F(MessagePipeTest, Basic) { int32_t buffer[2]; const uint32_t kBufferSize = static_cast<uint32_t>(sizeof(buffer)); uint32_t buffer_size; // Nothing to read yet on port 0. buffer[0] = 123; buffer[1] = 456; buffer_size = kBufferSize; ASSERT_EQ(MOJO_RESULT_SHOULD_WAIT, ReadMessage(pipe0_, buffer, &buffer_size)); ASSERT_EQ(kBufferSize, buffer_size); ASSERT_EQ(123, buffer[0]); ASSERT_EQ(456, buffer[1]); // Ditto for port 1. buffer[0] = 123; buffer[1] = 456; buffer_size = kBufferSize; ASSERT_EQ(MOJO_RESULT_SHOULD_WAIT, ReadMessage(pipe1_, buffer, &buffer_size)); // Write from port 1 (to port 0). buffer[0] = 789012345; buffer[1] = 0; ASSERT_EQ(MOJO_RESULT_OK, WriteMessage(pipe1_, buffer, sizeof(buffer[0]))); MojoHandleSignalsState state; ASSERT_EQ(MOJO_RESULT_OK, WaitForSignals(pipe0_, MOJO_HANDLE_SIGNAL_READABLE, &state)); // Read from port 0. buffer[0] = 123; buffer[1] = 456; buffer_size = kBufferSize; ASSERT_EQ(MOJO_RESULT_OK, ReadMessage(pipe0_, buffer, &buffer_size)); ASSERT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size); ASSERT_EQ(789012345, buffer[0]); ASSERT_EQ(456, buffer[1]); // Read again from port 0 -- it should be empty. buffer_size = kBufferSize; ASSERT_EQ(MOJO_RESULT_SHOULD_WAIT, ReadMessage(pipe0_, buffer, &buffer_size)); // Write two messages from port 0 (to port 1). buffer[0] = 123456789; buffer[1] = 0; ASSERT_EQ(MOJO_RESULT_OK, WriteMessage(pipe0_, buffer, sizeof(buffer[0]))); buffer[0] = 234567890; buffer[1] = 0; ASSERT_EQ(MOJO_RESULT_OK, WriteMessage(pipe0_, buffer, sizeof(buffer[0]))); ASSERT_EQ(MOJO_RESULT_OK, WaitForSignals(pipe1_, MOJO_HANDLE_SIGNAL_READABLE, &state)); // Read from port 1. buffer[0] = 123; buffer[1] = 456; buffer_size = kBufferSize; ASSERT_EQ(MOJO_RESULT_OK, ReadMessage(pipe1_, buffer, &buffer_size)); ASSERT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size); ASSERT_EQ(123456789, buffer[0]); ASSERT_EQ(456, buffer[1]); ASSERT_EQ(MOJO_RESULT_OK, WaitForSignals(pipe1_, MOJO_HANDLE_SIGNAL_READABLE, &state)); // Read again from port 1. buffer[0] = 123; buffer[1] = 456; buffer_size = kBufferSize; ASSERT_EQ(MOJO_RESULT_OK, ReadMessage(pipe1_, buffer, &buffer_size)); ASSERT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size); ASSERT_EQ(234567890, buffer[0]); ASSERT_EQ(456, buffer[1]); // Read again from port 1 -- it should be empty. buffer_size = kBufferSize; ASSERT_EQ(MOJO_RESULT_SHOULD_WAIT, ReadMessage(pipe1_, buffer, &buffer_size)); // Write from port 0 (to port 1). buffer[0] = 345678901; buffer[1] = 0; ASSERT_EQ(MOJO_RESULT_OK, WriteMessage(pipe0_, buffer, sizeof(buffer[0]))); // Close port 0. MojoClose(pipe0_); pipe0_ = MOJO_HANDLE_INVALID; ASSERT_EQ(MOJO_RESULT_OK, WaitForSignals(pipe1_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &state)); // Try to write from port 1 (to port 0). buffer[0] = 456789012; buffer[1] = 0; ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, WriteMessage(pipe1_, buffer, sizeof(buffer[0]))); // Read from port 1; should still get message (even though port 0 was closed). buffer[0] = 123; buffer[1] = 456; buffer_size = kBufferSize; ASSERT_EQ(MOJO_RESULT_OK, ReadMessage(pipe1_, buffer, &buffer_size)); ASSERT_EQ(static_cast<uint32_t>(sizeof(buffer[0])), buffer_size); ASSERT_EQ(345678901, buffer[0]); ASSERT_EQ(456, buffer[1]); // Read again from port 1 -- it should be empty (and port 0 is closed). buffer_size = kBufferSize; ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, ReadMessage(pipe1_, buffer, &buffer_size)); } TEST_F(MessagePipeTest, CloseWithQueuedIncomingMessages) { int32_t buffer[1]; const uint32_t kBufferSize = static_cast<uint32_t>(sizeof(buffer)); uint32_t buffer_size; // Write some messages from port 1 (to port 0). for (int32_t i = 0; i < 5; i++) { buffer[0] = i; ASSERT_EQ(MOJO_RESULT_OK, WriteMessage(pipe1_, buffer, kBufferSize)); } MojoHandleSignalsState state; ASSERT_EQ(MOJO_RESULT_OK, WaitForSignals(pipe0_, MOJO_HANDLE_SIGNAL_READABLE, &state)); // Port 0 shouldn't be empty. buffer_size = kBufferSize; ASSERT_EQ(MOJO_RESULT_OK, ReadMessage(pipe0_, buffer, &buffer_size)); ASSERT_EQ(kBufferSize, buffer_size); // Close port 0 first, which should have outstanding (incoming) messages. MojoClose(pipe0_); MojoClose(pipe1_); pipe0_ = pipe1_ = MOJO_HANDLE_INVALID; } TEST_F(MessagePipeTest, BasicWaiting) { MojoHandleSignalsState hss; int32_t buffer[1]; const uint32_t kBufferSize = static_cast<uint32_t>(sizeof(buffer)); uint32_t buffer_size; // Always writable (until the other port is closed). Not yet readable. Peer // not closed. hss = GetSignalsState(pipe0_); ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals); ASSERT_EQ(kAllSignals, hss.satisfiable_signals); hss = MojoHandleSignalsState(); // Write from port 0 (to port 1), to make port 1 readable. buffer[0] = 123456789; ASSERT_EQ(MOJO_RESULT_OK, WriteMessage(pipe0_, buffer, kBufferSize)); // Port 1 should already be readable now. ASSERT_EQ(MOJO_RESULT_OK, WaitForSignals(pipe1_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals); ASSERT_EQ(kAllSignals, hss.satisfiable_signals); // ... and still writable. hss = MojoHandleSignalsState(); ASSERT_EQ(MOJO_RESULT_OK, WaitForSignals(pipe1_, MOJO_HANDLE_SIGNAL_WRITABLE, &hss)); ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals); ASSERT_EQ(kAllSignals, hss.satisfiable_signals); // Close port 0. MojoClose(pipe0_); pipe0_ = MOJO_HANDLE_INVALID; // Port 1 should be signaled with peer closed. hss = MojoHandleSignalsState(); ASSERT_EQ(MOJO_RESULT_OK, WaitForSignals(pipe1_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss)); ASSERT_TRUE(hss.satisfied_signals & MOJO_HANDLE_SIGNAL_PEER_CLOSED); ASSERT_TRUE(hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_PEER_CLOSED); // Port 1 should not be writable now or ever again. hss = MojoHandleSignalsState(); ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, WaitForSignals(pipe1_, MOJO_HANDLE_SIGNAL_WRITABLE, &hss)); ASSERT_FALSE(hss.satisfied_signals & MOJO_HANDLE_SIGNAL_WRITABLE); ASSERT_FALSE(hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_WRITABLE); // But it should still be readable. hss = MojoHandleSignalsState(); ASSERT_EQ(MOJO_RESULT_OK, WaitForSignals(pipe1_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); ASSERT_TRUE(hss.satisfied_signals & MOJO_HANDLE_SIGNAL_READABLE); ASSERT_TRUE(hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_READABLE); // Read from port 1. buffer[0] = 0; buffer_size = kBufferSize; ASSERT_EQ(MOJO_RESULT_OK, ReadMessage(pipe1_, buffer, &buffer_size)); ASSERT_EQ(123456789, buffer[0]); // Now port 1 should no longer be readable. hss = MojoHandleSignalsState(); ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, WaitForSignals(pipe1_, MOJO_HANDLE_SIGNAL_READABLE, &hss)); ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals); ASSERT_FALSE(hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_READABLE); ASSERT_FALSE(hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_WRITABLE); } #if !defined(OS_IOS) const size_t kPingPongHandlesPerIteration = 30; const size_t kPingPongIterations = 500; DEFINE_TEST_CLIENT_TEST_WITH_PIPE(HandlePingPong, MessagePipeTest, h) { // Waits for a handle to become readable and writes it back to the sender. for (size_t i = 0; i < kPingPongIterations; i++) { MojoHandle handles[kPingPongHandlesPerIteration]; ReadMessageWithHandles(h, handles, kPingPongHandlesPerIteration); WriteMessageWithHandles(h, "", handles, kPingPongHandlesPerIteration); } EXPECT_EQ(MOJO_RESULT_OK, WaitForSignals(h, MOJO_HANDLE_SIGNAL_READABLE)); char msg[4]; uint32_t num_bytes = 4; EXPECT_EQ(MOJO_RESULT_OK, ReadMessage(h, msg, &num_bytes)); } // This test is flaky: http://crbug.com/585784 TEST_F(MessagePipeTest, DISABLED_DataPipeConsumerHandlePingPong) { MojoHandle p, c[kPingPongHandlesPerIteration]; for (size_t i = 0; i < kPingPongHandlesPerIteration; ++i) { EXPECT_EQ(MOJO_RESULT_OK, MojoCreateDataPipe(nullptr, &p, &c[i])); MojoClose(p); } RunTestClient("HandlePingPong", [&](MojoHandle h) { for (size_t i = 0; i < kPingPongIterations; i++) { WriteMessageWithHandles(h, "", c, kPingPongHandlesPerIteration); ReadMessageWithHandles(h, c, kPingPongHandlesPerIteration); } WriteMessage(h, "quit", 4); }); for (size_t i = 0; i < kPingPongHandlesPerIteration; ++i) MojoClose(c[i]); } // This test is flaky: http://crbug.com/585784 TEST_F(MessagePipeTest, DISABLED_DataPipeProducerHandlePingPong) { MojoHandle p[kPingPongHandlesPerIteration], c; for (size_t i = 0; i < kPingPongHandlesPerIteration; ++i) { EXPECT_EQ(MOJO_RESULT_OK, MojoCreateDataPipe(nullptr, &p[i], &c)); MojoClose(c); } RunTestClient("HandlePingPong", [&](MojoHandle h) { for (size_t i = 0; i < kPingPongIterations; i++) { WriteMessageWithHandles(h, "", p, kPingPongHandlesPerIteration); ReadMessageWithHandles(h, p, kPingPongHandlesPerIteration); } WriteMessage(h, "quit", 4); }); for (size_t i = 0; i < kPingPongHandlesPerIteration; ++i) MojoClose(p[i]); } TEST_F(MessagePipeTest, SharedBufferHandlePingPong) { MojoHandle buffers[kPingPongHandlesPerIteration]; for (size_t i = 0; i < kPingPongHandlesPerIteration; ++i) EXPECT_EQ(MOJO_RESULT_OK, MojoCreateSharedBuffer(1, nullptr, &buffers[i])); RunTestClient("HandlePingPong", [&](MojoHandle h) { for (size_t i = 0; i < kPingPongIterations; i++) { WriteMessageWithHandles(h, "", buffers, kPingPongHandlesPerIteration); ReadMessageWithHandles(h, buffers, kPingPongHandlesPerIteration); } WriteMessage(h, "quit", 4); }); for (size_t i = 0; i < kPingPongHandlesPerIteration; ++i) MojoClose(buffers[i]); } #endif // !defined(OS_IOS) TEST_F(FuseMessagePipeTest, Basic) { // Test that we can fuse pipes and they still work. MojoHandle a, b, c, d; CreateMessagePipe(&a, &b); CreateMessagePipe(&c, &d); EXPECT_EQ(MOJO_RESULT_OK, MojoFuseMessagePipes(b, c, nullptr)); // Handles b and c should be closed. EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(b)); EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(c)); const std::string kTestMessage1 = "Hello, world!"; const std::string kTestMessage2 = "Goodbye, world!"; WriteMessage(a, kTestMessage1); EXPECT_EQ(kTestMessage1, ReadMessage(d)); WriteMessage(d, kTestMessage2); EXPECT_EQ(kTestMessage2, ReadMessage(a)); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(a)); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(d)); } TEST_F(FuseMessagePipeTest, FuseAfterPeerWrite) { // Test that messages written before fusion are eventually delivered. MojoHandle a, b, c, d; CreateMessagePipe(&a, &b); CreateMessagePipe(&c, &d); const std::string kTestMessage1 = "Hello, world!"; const std::string kTestMessage2 = "Goodbye, world!"; WriteMessage(a, kTestMessage1); WriteMessage(d, kTestMessage2); EXPECT_EQ(MOJO_RESULT_OK, MojoFuseMessagePipes(b, c, nullptr)); // Handles b and c should be closed. EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(b)); EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(c)); EXPECT_EQ(kTestMessage1, ReadMessage(d)); EXPECT_EQ(kTestMessage2, ReadMessage(a)); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(a)); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(d)); } TEST_F(FuseMessagePipeTest, NoFuseAfterWrite) { // Test that a pipe endpoint which has been written to cannot be fused. MojoHandle a, b, c, d; CreateMessagePipe(&a, &b); CreateMessagePipe(&c, &d); WriteMessage(b, "shouldn't have done that!"); EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, MojoFuseMessagePipes(b, c, nullptr)); // Handles b and c should be closed. EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(b)); EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(c)); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(a)); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(d)); } TEST_F(FuseMessagePipeTest, NoFuseSelf) { // Test that a pipe's own endpoints can't be fused together. MojoHandle a, b; CreateMessagePipe(&a, &b); EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, MojoFuseMessagePipes(a, b, nullptr)); // Handles a and b should be closed. EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(a)); EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(b)); } TEST_F(FuseMessagePipeTest, FuseInvalidArguments) { MojoHandle a, b, c, d; CreateMessagePipe(&a, &b); CreateMessagePipe(&c, &d); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(b)); // Can't fuse an invalid handle. EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoFuseMessagePipes(b, c, nullptr)); // Handle c should be closed. EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(c)); // Can't fuse a non-message pipe handle. MojoHandle e, f; CreateDataPipe(&e, &f, 16); EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoFuseMessagePipes(e, d, nullptr)); // Handles d and e should be closed. EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(d)); EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(e)); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(a)); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(f)); } TEST_F(FuseMessagePipeTest, FuseAfterPeerClosure) { // Test that peer closure prior to fusion can still be detected after fusion. MojoHandle a, b, c, d; CreateMessagePipe(&a, &b); CreateMessagePipe(&c, &d); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(a)); EXPECT_EQ(MOJO_RESULT_OK, MojoFuseMessagePipes(b, c, nullptr)); // Handles b and c should be closed. EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(b)); EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(c)); EXPECT_EQ(MOJO_RESULT_OK, WaitForSignals(d, MOJO_HANDLE_SIGNAL_PEER_CLOSED)); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(d)); } TEST_F(FuseMessagePipeTest, FuseAfterPeerWriteAndClosure) { // Test that peer write and closure prior to fusion still results in the // both message arrival and awareness of peer closure. MojoHandle a, b, c, d; CreateMessagePipe(&a, &b); CreateMessagePipe(&c, &d); const std::string kTestMessage = "ayyy lmao"; WriteMessage(a, kTestMessage); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(a)); EXPECT_EQ(MOJO_RESULT_OK, MojoFuseMessagePipes(b, c, nullptr)); // Handles b and c should be closed. EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(b)); EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, MojoClose(c)); EXPECT_EQ(kTestMessage, ReadMessage(d)); EXPECT_EQ(MOJO_RESULT_OK, WaitForSignals(d, MOJO_HANDLE_SIGNAL_PEER_CLOSED)); EXPECT_EQ(MOJO_RESULT_OK, MojoClose(d)); } TEST_F(MessagePipeTest, ClosePipesStressTest) { // Stress test to exercise https://crbug.com/665869. const size_t kNumPipes = 100000; for (size_t i = 0; i < kNumPipes; ++i) { MojoHandle a, b; CreateMessagePipe(&a, &b); MojoClose(a); MojoClose(b); } } } // namespace } // namespace core } // namespace mojo