src/mojo/system/multiprocess_message_pipe_unittest.cc

   1 // Copyright 2013 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include <stdint.h>
   6
   7 #include <string>
   8
   9 #include "base/basictypes.h"
  10 #include "base/bind.h"
  11 #include "base/location.h"
  12 #include "base/logging.h"
  13 #include "base/threading/platform_thread.h"  // For |Sleep()|.
  14 #include "mojo/common/test/multiprocess_test_helper.h"
  15 #include "mojo/embedder/scoped_platform_handle.h"
  16 #include "mojo/system/channel.h"
  17 #include "mojo/system/local_message_pipe_endpoint.h"
  18 #include "mojo/system/message_pipe.h"
  19 #include "mojo/system/proxy_message_pipe_endpoint.h"
  20 #include "mojo/system/test_utils.h"
  21 #include "mojo/system/waiter.h"
  22 #include "testing/gtest/include/gtest/gtest.h"
  23
  24 namespace mojo {
  25 namespace system {
  26 namespace {
  27
  28 class ChannelThread {
  29  public:
  30   ChannelThread() : test_io_thread_(test::TestIOThread::kManualStart) {}
  31   ~ChannelThread() {
  32     Stop();
  33   }
  34
  35   void Start(embedder::ScopedPlatformHandle platform_handle,
  36              scoped_refptr<MessagePipe> message_pipe) {
  37     test_io_thread_.Start();
  38     test_io_thread_.PostTaskAndWait(
  39         FROM_HERE,
  40         base::Bind(&ChannelThread::InitChannelOnIOThread,
  41                    base::Unretained(this), base::Passed(&platform_handle),
  42                    message_pipe));
  43   }
  44
  45   void Stop() {
  46     if (channel_) {
  47       // Hack to flush write buffers before quitting.
  48       // TODO(vtl): Remove this once |Channel| has a
  49       // |FlushWriteBufferAndShutdown()| (or whatever).
  50       while (!channel_->IsWriteBufferEmpty())
  51         base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(20));
  52
  53       test_io_thread_.PostTaskAndWait(
  54           FROM_HERE,
  55           base::Bind(&ChannelThread::ShutdownChannelOnIOThread,
  56                      base::Unretained(this)));
  57     }
  58     test_io_thread_.Stop();
  59   }
  60
  61  private:
  62   void InitChannelOnIOThread(embedder::ScopedPlatformHandle platform_handle,
  63                              scoped_refptr<MessagePipe> message_pipe) {
  64     CHECK_EQ(base::MessageLoop::current(), test_io_thread_.message_loop());
  65     CHECK(platform_handle.is_valid());
  66
  67     // Create and initialize |Channel|.
  68     channel_ = new Channel();
  69     CHECK(channel_->Init(platform_handle.Pass()));
  70
  71     // Attach the message pipe endpoint.
  72     // Note: On the "server" (parent process) side, we need not attach the
  73     // message pipe endpoint immediately. However, on the "client" (child
  74     // process) side, this *must* be done here -- otherwise, the |Channel| may
  75     // receive/process messages (which it can do as soon as it's hooked up to
  76     // the IO thread message loop, and that message loop runs) before the
  77     // message pipe endpoint is attached.
  78     CHECK_EQ(channel_->AttachMessagePipeEndpoint(message_pipe, 1),
  79              Channel::kBootstrapEndpointId);
  80     channel_->RunMessagePipeEndpoint(Channel::kBootstrapEndpointId,
  81                                      Channel::kBootstrapEndpointId);
  82   }
  83
  84   void ShutdownChannelOnIOThread() {
  85     CHECK(channel_.get());
  86     channel_->Shutdown();
  87     channel_ = NULL;
  88   }
  89
  90   test::TestIOThread test_io_thread_;
  91   scoped_refptr<Channel> channel_;
  92
  93   DISALLOW_COPY_AND_ASSIGN(ChannelThread);
  94 };
  95
  96 class MultiprocessMessagePipeTest : public testing::Test {
  97  public:
  98   MultiprocessMessagePipeTest() {}
  99   virtual ~MultiprocessMessagePipeTest() {}
 100
 101  protected:
 102   void Init(scoped_refptr<MessagePipe> mp) {
 103     channel_thread_.Start(helper_.server_platform_handle.Pass(), mp);
 104   }
 105
 106   mojo::test::MultiprocessTestHelper* helper() { return &helper_; }
 107
 108  private:
 109   ChannelThread channel_thread_;
 110   mojo::test::MultiprocessTestHelper helper_;
 111
 112   DISALLOW_COPY_AND_ASSIGN(MultiprocessMessagePipeTest);
 113 };
 114
 115 MojoResult WaitIfNecessary(scoped_refptr<MessagePipe> mp, MojoWaitFlags flags) {
 116   Waiter waiter;
 117   waiter.Init();
 118
 119   MojoResult add_result = mp->AddWaiter(0, &waiter, flags, MOJO_RESULT_OK);
 120   if (add_result != MOJO_RESULT_OK) {
 121     return (add_result == MOJO_RESULT_ALREADY_EXISTS) ? MOJO_RESULT_OK :
 122                                                         add_result;
 123   }
 124
 125   MojoResult wait_result = waiter.Wait(MOJO_DEADLINE_INDEFINITE);
 126   mp->RemoveWaiter(0, &waiter);
 127   return wait_result;
 128 }
 129
 130 // For each message received, sends a reply message with the same contents
 131 // repeated twice, until the other end is closed or it receives "quitquitquit"
 132 // (which it doesn't reply to). It'll return the number of messages received,
 133 // not including any "quitquitquit" message, modulo 100.
 134 MOJO_MULTIPROCESS_TEST_CHILD_MAIN(EchoEcho) {
 135   ChannelThread channel_thread;
 136   embedder::ScopedPlatformHandle client_platform_handle =
 137       mojo::test::MultiprocessTestHelper::client_platform_handle.Pass();
 138   CHECK(client_platform_handle.is_valid());
 139   scoped_refptr<MessagePipe> mp(new MessagePipe(
 140       scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint()),
 141       scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint())));
 142   channel_thread.Start(client_platform_handle.Pass(), mp);
 143
 144   const std::string quitquitquit("quitquitquit");
 145   int rv = 0;
 146   for (;; rv = (rv + 1) % 100) {
 147     // Wait for our end of the message pipe to be readable.
 148     MojoResult result = WaitIfNecessary(mp, MOJO_WAIT_FLAG_READABLE);
 149     if (result != MOJO_RESULT_OK) {
 150       // It was closed, probably.
 151       CHECK_EQ(result, MOJO_RESULT_FAILED_PRECONDITION);
 152       break;
 153     }
 154
 155     std::string read_buffer(1000, '\0');
 156     uint32_t read_buffer_size = static_cast<uint32_t>(read_buffer.size());
 157     CHECK_EQ(mp->ReadMessage(0,
 158                              &read_buffer[0], &read_buffer_size,
 159                              NULL, NULL,
 160                              MOJO_READ_MESSAGE_FLAG_NONE),
 161              MOJO_RESULT_OK);
 162     read_buffer.resize(read_buffer_size);
 163     VLOG(2) << "Child got: " << read_buffer;
 164
 165     if (read_buffer == quitquitquit) {
 166       VLOG(2) << "Child quitting.";
 167       break;
 168     }
 169
 170     std::string write_buffer = read_buffer + read_buffer;
 171     CHECK_EQ(mp->WriteMessage(0,
 172                               write_buffer.data(),
 173                               static_cast<uint32_t>(write_buffer.size()),
 174                               NULL,
 175                               MOJO_WRITE_MESSAGE_FLAG_NONE),
 176              MOJO_RESULT_OK);
 177   }
 178
 179
 180   mp->Close(0);
 181   return rv;
 182 }
 183
 184 // Sends "hello" to child, and expects "hellohello" back.
 185 TEST_F(MultiprocessMessagePipeTest, Basic) {
 186   helper()->StartChild("EchoEcho");
 187
 188   scoped_refptr<MessagePipe> mp(new MessagePipe(
 189       scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint()),
 190       scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint())));
 191   Init(mp);
 192
 193   std::string hello("hello");
 194   EXPECT_EQ(MOJO_RESULT_OK,
 195             mp->WriteMessage(0,
 196                              hello.data(), static_cast<uint32_t>(hello.size()),
 197                              NULL,
 198                              MOJO_WRITE_MESSAGE_FLAG_NONE));
 199
 200   EXPECT_EQ(MOJO_RESULT_OK, WaitIfNecessary(mp, MOJO_WAIT_FLAG_READABLE));
 201
 202   std::string read_buffer(1000, '\0');
 203   uint32_t read_buffer_size = static_cast<uint32_t>(read_buffer.size());
 204   CHECK_EQ(mp->ReadMessage(0,
 205                            &read_buffer[0], &read_buffer_size,
 206                            NULL, NULL,
 207                            MOJO_READ_MESSAGE_FLAG_NONE),
 208            MOJO_RESULT_OK);
 209   read_buffer.resize(read_buffer_size);
 210   VLOG(2) << "Parent got: " << read_buffer;
 211   EXPECT_EQ(hello + hello, read_buffer);
 212
 213   mp->Close(0);
 214
 215   // We sent one message.
 216   EXPECT_EQ(1 % 100, helper()->WaitForChildShutdown());
 217 }
 218
 219 // Sends a bunch of messages to the child. Expects them "repeated" back. Waits
 220 // for the child to close its end before quitting.
 221 TEST_F(MultiprocessMessagePipeTest, QueueMessages) {
 222   helper()->StartChild("EchoEcho");
 223
 224   scoped_refptr<MessagePipe> mp(new MessagePipe(
 225       scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint()),
 226       scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint())));
 227   Init(mp);
 228
 229   static const size_t kNumMessages = 1001;
 230   for (size_t i = 0; i < kNumMessages; i++) {
 231     std::string write_buffer(i, 'A' + (i % 26));
 232     EXPECT_EQ(MOJO_RESULT_OK,
 233               mp->WriteMessage(0,
 234                                write_buffer.data(),
 235                                static_cast<uint32_t>(write_buffer.size()),
 236                                NULL,
 237                                MOJO_WRITE_MESSAGE_FLAG_NONE));
 238   }
 239
 240   const std::string quitquitquit("quitquitquit");
 241   EXPECT_EQ(MOJO_RESULT_OK,
 242             mp->WriteMessage(0,
 243                              quitquitquit.data(),
 244                              static_cast<uint32_t>(quitquitquit.size()),
 245                              NULL,
 246                              MOJO_WRITE_MESSAGE_FLAG_NONE));
 247
 248   for (size_t i = 0; i < kNumMessages; i++) {
 249     EXPECT_EQ(MOJO_RESULT_OK, WaitIfNecessary(mp, MOJO_WAIT_FLAG_READABLE));
 250
 251     std::string read_buffer(kNumMessages * 2, '\0');
 252     uint32_t read_buffer_size = static_cast<uint32_t>(read_buffer.size());
 253     CHECK_EQ(mp->ReadMessage(0,
 254                              &read_buffer[0], &read_buffer_size,
 255                              NULL, NULL,
 256                              MOJO_READ_MESSAGE_FLAG_NONE),
 257              MOJO_RESULT_OK);
 258     read_buffer.resize(read_buffer_size);
 259
 260     EXPECT_EQ(std::string(i * 2, 'A' + (i % 26)), read_buffer);
 261   }
 262
 263   // Wait for it to become readable, which should fail (since we sent
 264   // "quitquitquit").
 265   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
 266             WaitIfNecessary(mp, MOJO_WAIT_FLAG_READABLE));
 267
 268   mp->Close(0);
 269
 270   EXPECT_EQ(static_cast<int>(kNumMessages % 100),
 271             helper()->WaitForChildShutdown());
 272 }
 273
 274 }  // namespace
 275 }  // namespace system
 276 }  // namespace mojo