Add multiplexing of message pipes in the new EDK.

mojo/edk/system/raw_channel_unittest.cc

 // Copyright 2014 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 "mojo/edk/system/raw_channel.h"
 
 #include <stdint.h>
 #include <stdio.h>
 
 #include <vector>
 
 #include "base/bind.h"
 #include "base/files/file_path.h"
 #include "base/files/file_util.h"
 #include "base/files/scoped_file.h"
 #include "base/files/scoped_temp_dir.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/memory/scoped_vector.h"
 #include "base/rand_util.h"
 #include "base/synchronization/lock.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/test/test_io_thread.h"
 #include "base/threading/simple_thread.h"
-#include "build/build_config.h"  // TODO(vtl): Remove this.
+#include "build/build_config.h"
+#include "mojo/edk/embedder/embedder.h"
+#include "mojo/edk/embedder/embedder_internal.h"
 #include "mojo/edk/embedder/platform_channel_pair.h"
 #include "mojo/edk/embedder/platform_handle.h"
 #include "mojo/edk/embedder/scoped_platform_handle.h"
 #include "mojo/edk/system/message_in_transit.h"
 #include "mojo/edk/system/test_utils.h"
 #include "mojo/edk/system/transport_data.h"
 #include "mojo/edk/test/test_utils.h"
 #include "mojo/public/cpp/system/macros.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
@@ skipping 29 matching lines @@
   scoped_ptr<MessageInTransit> message(MakeTestMessage(num_bytes));
 
   size_t write_size = 0;
   test::BlockingWrite(handle, message->main_buffer(),
                       message->main_buffer_size(), &write_size);
   return write_size == message->main_buffer_size();
 }
 
 // -----------------------------------------------------------------------------
 
-class RawChannelTest : public test::MojoSystemTest {
+class RawChannelTest : public testing::Test {
  public:
   RawChannelTest() {}
   ~RawChannelTest() override {}
 
   void SetUp() override {
     PlatformChannelPair channel_pair;
     handles[0] = channel_pair.PassServerHandle();
     handles[1] = channel_pair.PassClientHandle();\
   }
 
   void TearDown() override {
     handles[0].reset();
     handles[1].reset();
   }
 
+  void FlushIOThread() {
+    base::WaitableEvent event(false, false);
+    internal::g_io_thread_task_runner->PostTask(
+        FROM_HERE,
+        base::Bind(&base::WaitableEvent::Signal, base::Unretained(&event)));
+    event.Wait();
+  }
+
  protected:
   ScopedPlatformHandle handles[2];
 
  private:
   MOJO_DISALLOW_COPY_AND_ASSIGN(RawChannelTest);
 };
 
 // RawChannelTest.WriteMessage -------------------------------------------------
 
 class WriteOnlyRawChannelDelegate : public RawChannel::Delegate {
@@ skipping 80 matching lines @@
   std::vector<unsigned char> bytes_;
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(TestMessageReaderAndChecker);
 };
 
 // Tests writing (and verifies reading using our own custom reader).
 TEST_F(RawChannelTest, WriteMessage) {
   WriteOnlyRawChannelDelegate delegate;
   RawChannel* rc = RawChannel::Create(handles[0].Pass());
   TestMessageReaderAndChecker checker(handles[1].get());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&InitOnIOThread, rc, base::Unretained(&delegate)));
 
   // Write and read, for a variety of sizes.
   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1) {
     EXPECT_TRUE(rc->WriteMessage(MakeTestMessage(size)));
     EXPECT_TRUE(checker.ReadAndCheckNextMessage(size)) << size;
   }
 
   // Write/queue and read afterwards, for a variety of sizes.
   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1)
     EXPECT_TRUE(rc->WriteMessage(MakeTestMessage(size)));
   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1)
     EXPECT_TRUE(checker.ReadAndCheckNextMessage(size)) << size;
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE, base::Bind(&RawChannel::Shutdown, base::Unretained(rc)));
 }
 
 // RawChannelTest.OnReadMessage ------------------------------------------------
 
 class ReadCheckerRawChannelDelegate : public RawChannel::Delegate {
  public:
   ReadCheckerRawChannelDelegate() : done_event_(false, false), position_(0) {}
   ~ReadCheckerRawChannelDelegate() override {}
 
@@ skipping 48 matching lines @@
   std::vector<uint32_t> expected_sizes_;
   size_t position_;
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(ReadCheckerRawChannelDelegate);
 };
 
 // Tests reading (writing using our own custom writer).
 TEST_F(RawChannelTest, OnReadMessage) {
   ReadCheckerRawChannelDelegate delegate;
   RawChannel* rc = RawChannel::Create(handles[0].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&InitOnIOThread, rc, base::Unretained(&delegate)));
 
   // Write and read, for a variety of sizes.
   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1) {
     delegate.SetExpectedSizes(std::vector<uint32_t>(1, size));
 
     EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), size));
 
     delegate.Wait();
   }
 
   // Set up reader and write as fast as we can.
   // Write/queue and read afterwards, for a variety of sizes.
   std::vector<uint32_t> expected_sizes;
   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1)
     expected_sizes.push_back(size);
   delegate.SetExpectedSizes(expected_sizes);
   for (uint32_t size = 1; size < 5 * 1000 * 1000; size += size / 2 + 1)
     EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), size));
   delegate.Wait();
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE, base::Bind(&RawChannel::Shutdown, base::Unretained(rc)));
 }
 
 // RawChannelTest.WriteMessageAndOnReadMessage ---------------------------------
 
 class RawChannelWriterThread : public base::SimpleThread {
  public:
   RawChannelWriterThread(RawChannel* raw_channel, size_t write_count)
       : base::SimpleThread("raw_channel_writer_thread"),
         raw_channel_(raw_channel),
@@ skipping 53 matching lines @@
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(ReadCountdownRawChannelDelegate);
 };
 
 TEST_F(RawChannelTest, WriteMessageAndOnReadMessage) {
   static const size_t kNumWriterThreads = 10;
   static const size_t kNumWriteMessagesPerThread = 400;
 
   WriteOnlyRawChannelDelegate writer_delegate;
   RawChannel* writer_rc = RawChannel::Create(handles[0].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
     FROM_HERE,
     base::Bind(&InitOnIOThread, writer_rc,
                base::Unretained(&writer_delegate)));
 
   ReadCountdownRawChannelDelegate reader_delegate(kNumWriterThreads *
                                                   kNumWriteMessagesPerThread);
   RawChannel* reader_rc = RawChannel::Create(handles[1].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
     FROM_HERE,
     base::Bind(&InitOnIOThread, reader_rc,
                base::Unretained(&reader_delegate)));
 
   {
     ScopedVector<RawChannelWriterThread> writer_threads;
     for (size_t i = 0; i < kNumWriterThreads; i++) {
       writer_threads.push_back(new RawChannelWriterThread(
           writer_rc, kNumWriteMessagesPerThread));
     }
     for (size_t i = 0; i < writer_threads.size(); i++)
       writer_threads[i]->Start();
   }  // Joins all the writer threads.
 
   // Sleep a bit, to let any extraneous reads be processed. (There shouldn't be
   // any, but we want to know about them.)
   test::Sleep(test::DeadlineFromMilliseconds(100));
 
   // Wait for reading to finish.
   reader_delegate.Wait();
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&RawChannel::Shutdown, base::Unretained(reader_rc)));
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&RawChannel::Shutdown, base::Unretained(writer_rc)));
 }
 
 // RawChannelTest.OnError ------------------------------------------------------
 
 class ErrorRecordingRawChannelDelegate
     : public ReadCountdownRawChannelDelegate {
  public:
   ErrorRecordingRawChannelDelegate(size_t expected_read_count,
@@ skipping 44 matching lines @@
   bool expecting_read_error_;
   bool expecting_write_error_;
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(ErrorRecordingRawChannelDelegate);
 };
 
 // Tests (fatal) errors.
 TEST_F(RawChannelTest, OnError) {
   ErrorRecordingRawChannelDelegate delegate(0, true, true);
   RawChannel* rc = RawChannel::Create(handles[0].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&InitOnIOThread, rc, base::Unretained(&delegate)));
+  FlushIOThread();
 
   // Close the handle of the other end, which should make writing fail.
   handles[1].reset();
 
   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
 
   // We should get a write error.
   delegate.WaitForWriteError();
 
   // We should also get a read error.
   delegate.WaitForReadError();
 
   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(2)));
 
   // Sleep a bit, to make sure we don't get another |OnError()|
   // notification. (If we actually get another one, |OnError()| crashes.)
   test::Sleep(test::DeadlineFromMilliseconds(20));
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE, base::Bind(&RawChannel::Shutdown, base::Unretained(rc)));
 }
 
 // RawChannelTest.ReadUnaffectedByWriteError -----------------------------------
 
 TEST_F(RawChannelTest, ReadUnaffectedByWriteError) {
   const size_t kMessageCount = 5;
 
   // Write a few messages into the other end.
   uint32_t message_size = 1;
   for (size_t i = 0; i < kMessageCount;
        i++, message_size += message_size / 2 + 1)
     EXPECT_TRUE(WriteTestMessageToHandle(handles[1].get(), message_size));
 
   // Close the other end, which should make writing fail.
   handles[1].reset();
 
   // Only start up reading here. The system buffer should still contain the
   // messages that were written.
   ErrorRecordingRawChannelDelegate delegate(kMessageCount, true, true);
   RawChannel* rc = RawChannel::Create(handles[0].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&InitOnIOThread, rc, base::Unretained(&delegate)));
+  FlushIOThread();
 
   EXPECT_FALSE(rc->WriteMessage(MakeTestMessage(1)));
 
   // We should definitely get a write error.
   delegate.WaitForWriteError();
 
   // Wait for reading to finish. A writing failure shouldn't affect reading.
   delegate.Wait();
 
   // And then we should get a read error.
   delegate.WaitForReadError();
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE, base::Bind(&RawChannel::Shutdown, base::Unretained(rc)));
 }
 
 // RawChannelTest.ReadWritePlatformHandles -------------------------------------
 
 class ReadPlatformHandlesCheckerRawChannelDelegate
     : public RawChannel::Delegate {
  public:
   ReadPlatformHandlesCheckerRawChannelDelegate() : done_event_(false, false) {}
   ~ReadPlatformHandlesCheckerRawChannelDelegate() override {}
@@ skipping 48 matching lines @@
 
   MOJO_DISALLOW_COPY_AND_ASSIGN(ReadPlatformHandlesCheckerRawChannelDelegate);
 };
 
 TEST_F(RawChannelTest, ReadWritePlatformHandles) {
   base::ScopedTempDir temp_dir;
   ASSERT_TRUE(temp_dir.CreateUniqueTempDir());
 
   WriteOnlyRawChannelDelegate write_delegate;
   RawChannel* rc_write = RawChannel::Create(handles[0].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&InitOnIOThread, rc_write, base::Unretained(&write_delegate)));
 
   ReadPlatformHandlesCheckerRawChannelDelegate read_delegate;
   RawChannel* rc_read = RawChannel::Create(handles[1].Pass());
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&InitOnIOThread, rc_read, base::Unretained(&read_delegate)));
 
   base::FilePath unused;
   base::ScopedFILE fp1(
       base::CreateAndOpenTemporaryFileInDir(temp_dir.path(), &unused));
   EXPECT_EQ(1u, fwrite("1", 1, 1, fp1.get()));
   base::ScopedFILE fp2(
       base::CreateAndOpenTemporaryFileInDir(temp_dir.path(), &unused));
   EXPECT_EQ(1u, fwrite("2", 1, 1, fp2.get()));
 
   {
     const char kHello[] = "hello";
     ScopedPlatformHandleVectorPtr platform_handles(new PlatformHandleVector());
     platform_handles->push_back(
         test::PlatformHandleFromFILE(fp1.Pass()).release());
     platform_handles->push_back(
         test::PlatformHandleFromFILE(fp2.Pass()).release());
 
     scoped_ptr<MessageInTransit> message(
         new MessageInTransit(MessageInTransit::Type::MESSAGE,
                              sizeof(kHello), kHello));
     message->SetTransportData(make_scoped_ptr(new TransportData(
         platform_handles.Pass(), rc_write->GetSerializedPlatformHandleSize())));
     EXPECT_TRUE(rc_write->WriteMessage(message.Pass()));
   }
 
   read_delegate.Wait();
 
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&RawChannel::Shutdown, base::Unretained(rc_read)));
-  test_io_thread()->PostTaskAndWait(
+  internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&RawChannel::Shutdown, base::Unretained(rc_write)));
 }
 
 }  // namespace
 }  // namespace edk
 }  // namespace mojo