Mojo: add mojo_message_pipe_perftests

This is analogous to ipc_perftests. It measures roundtrip
time with various message sizes.

TEST=message_pipe_perftest.cc
R=viettrungluu@chromium.org
BUG=none

Review URL: https://codereview.chromium.org/501983003

Cr-Commit-Position: refs/heads/master@{#292024}

mojo/BUILD.gn

@@ -34,6 +34,7 @@ group("tests") {
     "//mojo/public/js/bindings/tests:mojo_js_unittests",
     "//mojo/services/public/cpp/surfaces/tests:mojo_surfaces_lib_unittests",
     "//mojo/shell:mojo_shell_tests",
+    "//mojo/system:mojo_message_pipe_perftests",
     "//mojo/system:mojo_system_unittests",
   ]
 

mojo/mojo_base.gyp

@@ -22,6 +22,7 @@
         'mojo_js_bindings',
         'mojo_js_unittests',
         'mojo_message_generator',
+        'mojo_message_pipe_perftests',
         'mojo_public_application_unittests',
         'mojo_public_test_utils',
         'mojo_public_bindings_unittests',
@@ -207,6 +208,8 @@
         'system/local_data_pipe_unittest.cc',
         'system/memory_unittest.cc',
         'system/message_pipe_dispatcher_unittest.cc',
+        'system/message_pipe_test_utils.h',
+        'system/message_pipe_test_utils.cc',
         'system/message_pipe_unittest.cc',
         'system/multiprocess_message_pipe_unittest.cc',
         'system/options_validation_unittest.cc',
@@ -224,6 +227,27 @@
         'system/waiter_unittest.cc',
       ],
     },
+    {
+      # GN version: //mojo/system:mojo_message_pipe_perftests
+      'target_name': 'mojo_message_pipe_perftests',
+      'type': 'executable',
+      'dependencies': [
+        '../base/base.gyp:base',
+        '../base/base.gyp:test_support_base',
+        '../base/base.gyp:test_support_perf',
+        '../testing/gtest.gyp:gtest',
+        'mojo_common_test_support',
+        'mojo_environment_chromium',
+        'mojo_system_impl',
+      ],
+      'sources': [
+        'system/message_pipe_perftest.cc',
+        'system/message_pipe_test_utils.h',
+        'system/message_pipe_test_utils.cc',
+        'system/test_utils.cc',
+        'system/test_utils.h',
+      ],
+    },
     {
       # GN version: //mojo/common/test:test_support_impl
       'target_name': 'mojo_test_support_impl',

mojo/system/BUILD.gn

@@ -103,6 +103,8 @@ test("mojo_system_unittests") {
     "local_data_pipe_unittest.cc",
     "memory_unittest.cc",
     "message_pipe_dispatcher_unittest.cc",
+    "message_pipe_test_utils.h",
+    "message_pipe_test_utils.cc",
     "message_pipe_unittest.cc",
     "multiprocess_message_pipe_unittest.cc",
     "options_validation_unittest.cc",
@@ -120,3 +122,23 @@ test("mojo_system_unittests") {
     "waiter_unittest.cc",
   ]
 }
+
+# GYP version: mojo/mojo_base.gyp:mojo_message_pipe_perftests
+test("mojo_message_pipe_perftests") {
+  deps = [
+    ":system",
+    "//base",
+    "//base/test:test_support",
+    "//base/test:test_support_perf",
+    "//mojo/common/test:test_support",
+    "//testing/gtest",
+  ]
+
+  sources = [
+    "message_pipe_perftest.cc",
+    "message_pipe_test_utils.h",
+    "message_pipe_test_utils.cc",
+    "test_utils.cc",
+    "test_utils.h",
+  ]
+}

mojo/system/message_pipe_perftest.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 <stdint.h>
+#include <stdio.h>
+#include <string.h>
+
+#include <string>
+#include <vector>
+
+#include "base/bind.h"
+#include "base/location.h"
+#include "base/logging.h"
+#include "base/macros.h"
+#include "base/pickle.h"
+#include "base/strings/stringprintf.h"
+#include "base/test/perf_time_logger.h"
+#include "base/time/time.h"
+#include "mojo/common/test/test_utils.h"
+#include "mojo/embedder/scoped_platform_handle.h"
+#include "mojo/system/channel.h"
+#include "mojo/system/local_message_pipe_endpoint.h"
+#include "mojo/system/message_pipe.h"
+#include "mojo/system/message_pipe_test_utils.h"
+#include "mojo/system/proxy_message_pipe_endpoint.h"
+#include "mojo/system/raw_channel.h"
+#include "mojo/system/test_utils.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace mojo {
+namespace system {
+namespace {
+
+class MultiprocessMessagePipePerfTest
+    : public test::MultiprocessMessagePipeTestBase {
+ public:
+  MultiprocessMessagePipePerfTest() : message_count_(0), message_size_(0) {}
+
+  void SetUpMeasurement(int message_count, size_t message_size) {
+    message_count_ = message_count;
+    message_size_ = message_size;
+    payload_ = Pickle();
+    payload_.WriteString(std::string(message_size, '*'));
+    read_buffer_.resize(message_size * 2);
+  }
+
+ protected:
+  void WriteWaitThenRead(scoped_refptr<MessagePipe> mp) {
+    CHECK_EQ(mp->WriteMessage(0,
+                              UserPointer<const void>(payload_.data()),
+                              static_cast<uint32_t>(payload_.size()),
+                              NULL,
+                              MOJO_WRITE_MESSAGE_FLAG_NONE),
+             MOJO_RESULT_OK);
+    HandleSignalsState hss;
+    CHECK_EQ(test::WaitIfNecessary(mp, MOJO_HANDLE_SIGNAL_READABLE, &hss),
+             MOJO_RESULT_OK);
+    uint32_t read_buffer_size = static_cast<uint32_t>(read_buffer_.size());
+    CHECK_EQ(mp->ReadMessage(0,
+                             UserPointer<void>(&read_buffer_[0]),
+                             MakeUserPointer(&read_buffer_size),
+                             NULL,
+                             NULL,
+                             MOJO_READ_MESSAGE_FLAG_NONE),
+             MOJO_RESULT_OK);
+    CHECK_EQ(read_buffer_size, static_cast<uint32_t>(payload_.size()));
+  }
+
+  void SendQuitMessage(scoped_refptr<MessagePipe> mp) {
+    CHECK_EQ(mp->WriteMessage(0,
+                              UserPointer<const void>(""),
+                              0,
+                              NULL,
+                              MOJO_WRITE_MESSAGE_FLAG_NONE),
+             MOJO_RESULT_OK);
+  }
+
+  void Measure(scoped_refptr<MessagePipe> mp) {
+    // Have one ping-pong to ensure channel being established.
+    WriteWaitThenRead(mp);
+
+    std::string test_name =
+        base::StringPrintf("IPC_Perf_%dx_%u",
+                           message_count_,
+                           static_cast<unsigned>(message_size_));
+    base::PerfTimeLogger logger(test_name.c_str());
+
+    for (int i = 0; i < message_count_; ++i)
+      WriteWaitThenRead(mp);
+
+    logger.Done();
+  }
+
+ private:
+  int message_count_;
+  size_t message_size_;
+  Pickle payload_;
+  std::string read_buffer_;
+  scoped_ptr<base::PerfTimeLogger> perf_logger_;
+};
+
+// For each message received, sends a reply message with the same contents
+// repeated twice, until the other end is closed or it receives "quitquitquit"
+// (which it doesn't reply to). It'll return the number of messages received,
+// not including any "quitquitquit" message, modulo 100.
+MOJO_MULTIPROCESS_TEST_CHILD_MAIN(PingPongClient) {
+  embedder::SimplePlatformSupport platform_support;
+  test::ChannelThread channel_thread(&platform_support);
+  embedder::ScopedPlatformHandle client_platform_handle =
+      mojo::test::MultiprocessTestHelper::client_platform_handle.Pass();
+  CHECK(client_platform_handle.is_valid());
+  scoped_refptr<MessagePipe> mp(new MessagePipe(
+      scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint()),
+      scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint())));
+  channel_thread.Start(client_platform_handle.Pass(), mp);
+
+  std::string buffer(1000000, '\0');
+  int rv = 0;
+  while (true) {
+    // Wait for our end of the message pipe to be readable.
+    HandleSignalsState hss;
+    MojoResult result =
+        test::WaitIfNecessary(mp, MOJO_HANDLE_SIGNAL_READABLE, &hss);
+    if (result != MOJO_RESULT_OK) {
+      rv = result;
+      break;
+    }
+
+    uint32_t read_size = static_cast<uint32_t>(buffer.size());
+    CHECK_EQ(mp->ReadMessage(0,
+                             UserPointer<void>(&buffer[0]),
+                             MakeUserPointer(&read_size),
+                             NULL,
+                             NULL,
+                             MOJO_READ_MESSAGE_FLAG_NONE),
+             MOJO_RESULT_OK);
+
+    // Empty message indicates quitting
+    if (0 == read_size)
+      break;
+
+    CHECK_EQ(mp->WriteMessage(0,
+                              UserPointer<const void>(&buffer[0]),
+                              static_cast<uint32_t>(read_size),
+                              NULL,
+                              MOJO_WRITE_MESSAGE_FLAG_NONE),
+             MOJO_RESULT_OK);
+  }
+
+  mp->Close(0);
+  return rv;
+}
+
+// Repeatedly sends messages as previous one got replied by the child.
+// Waits for the child to close its end before quitting once specified
+// number of messages has been sent.
+TEST_F(MultiprocessMessagePipePerfTest, PingPong) {
+  helper()->StartChild("PingPongClient");
+
+  scoped_refptr<MessagePipe> mp(new MessagePipe(
+      scoped_ptr<MessagePipeEndpoint>(new LocalMessagePipeEndpoint()),
+      scoped_ptr<MessagePipeEndpoint>(new ProxyMessagePipeEndpoint())));
+  Init(mp);
+
+  // This values are set to align with one at ipc_pertests.cc for comparison.
+  const size_t kMsgSize[5] = {12, 144, 1728, 20736, 248832};
+  const int kMessageCount[5] = {50000, 50000, 50000, 12000, 1000};
+
+  for (size_t i = 0; i < 5; i++) {
+    SetUpMeasurement(kMessageCount[i], kMsgSize[i]);
+    Measure(mp);
+  }
+
+  SendQuitMessage(mp);
+  mp->Close(0);
+  EXPECT_EQ(0, helper()->WaitForChildShutdown());
+}
+
+}  // namespace
+}  // namespace system
+}  // namespace mojo

mojo/system/message_pipe_test_utils.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/system/message_pipe_test_utils.h"
+
+#include "base/bind.h"
+#include "base/threading/platform_thread.h"  // For |Sleep()|.
+#include "mojo/system/waiter.h"
+
+namespace mojo {
+namespace system {
+namespace test {
+
+MojoResult WaitIfNecessary(scoped_refptr<MessagePipe> mp,
+                           MojoHandleSignals signals,
+                           HandleSignalsState* signals_state) {
+  Waiter waiter;
+  waiter.Init();
+
+  MojoResult add_result = mp->AddWaiter(0, &waiter, signals, 0, signals_state);
+  if (add_result != MOJO_RESULT_OK) {
+    return (add_result == MOJO_RESULT_ALREADY_EXISTS) ? MOJO_RESULT_OK
+                                                      : add_result;
+  }
+
+  MojoResult wait_result = waiter.Wait(MOJO_DEADLINE_INDEFINITE, NULL);
+  mp->RemoveWaiter(0, &waiter, signals_state);
+  return wait_result;
+}
+
+ChannelThread::ChannelThread(embedder::PlatformSupport* platform_support)
+    : platform_support_(platform_support),
+      test_io_thread_(test::TestIOThread::kManualStart) {
+}
+
+ChannelThread::~ChannelThread() {
+  Stop();
+}
+
+void ChannelThread::Start(embedder::ScopedPlatformHandle platform_handle,
+                          scoped_refptr<MessagePipe> message_pipe) {
+  test_io_thread_.Start();
+  test_io_thread_.PostTaskAndWait(
+      FROM_HERE,
+      base::Bind(&ChannelThread::InitChannelOnIOThread,
+                 base::Unretained(this),
+                 base::Passed(&platform_handle),
+                 message_pipe));
+}
+
+void ChannelThread::Stop() {
+  if (channel_.get()) {
+    // Hack to flush write buffers before quitting.
+    // TODO(vtl): Remove this once |Channel| has a
+    // |FlushWriteBufferAndShutdown()| (or whatever).
+    while (!channel_->IsWriteBufferEmpty())
+      base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(20));
+
+    test_io_thread_.PostTaskAndWait(
+        FROM_HERE,
+        base::Bind(&ChannelThread::ShutdownChannelOnIOThread,
+                   base::Unretained(this)));
+  }
+  test_io_thread_.Stop();
+}
+
+void ChannelThread::InitChannelOnIOThread(
+    embedder::ScopedPlatformHandle platform_handle,
+    scoped_refptr<MessagePipe> message_pipe) {
+  CHECK_EQ(base::MessageLoop::current(), test_io_thread_.message_loop());
+  CHECK(platform_handle.is_valid());
+
+  // Create and initialize |Channel|.
+  channel_ = new Channel(platform_support_);
+  CHECK(channel_->Init(RawChannel::Create(platform_handle.Pass())));
+
+  // Attach the message pipe endpoint.
+  // Note: On the "server" (parent process) side, we need not attach the
+  // message pipe endpoint immediately. However, on the "client" (child
+  // process) side, this *must* be done here -- otherwise, the |Channel| may
+  // receive/process messages (which it can do as soon as it's hooked up to
+  // the IO thread message loop, and that message loop runs) before the
+  // message pipe endpoint is attached.
+  CHECK_EQ(channel_->AttachMessagePipeEndpoint(message_pipe, 1),
+           Channel::kBootstrapEndpointId);
+  CHECK(channel_->RunMessagePipeEndpoint(Channel::kBootstrapEndpointId,
+                                         Channel::kBootstrapEndpointId));
+}
+
+void ChannelThread::ShutdownChannelOnIOThread() {
+  CHECK(channel_.get());
+  channel_->Shutdown();
+  channel_ = NULL;
+}
+
+MultiprocessMessagePipeTestBase::MultiprocessMessagePipeTestBase()
+    : channel_thread_(&platform_support_) {
+}
+
+MultiprocessMessagePipeTestBase::~MultiprocessMessagePipeTestBase() {
+}
+
+void MultiprocessMessagePipeTestBase::Init(scoped_refptr<MessagePipe> mp) {
+  channel_thread_.Start(helper_.server_platform_handle.Pass(), mp);
+}
+
+}  // namespace test
+}  // namespace system
+}  // namespace mojo

mojo/system/message_pipe_test_utils.h

+// 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.
+
+#ifndef MOJO_SYSTEM_MESSAGE_PIPE_TEST_UTILS_H_
+#define MOJO_SYSTEM_MESSAGE_PIPE_TEST_UTILS_H_
+
+#include "mojo/common/test/multiprocess_test_helper.h"
+#include "mojo/embedder/simple_platform_support.h"
+#include "mojo/system/channel.h"
+#include "mojo/system/message_pipe.h"
+#include "mojo/system/test_utils.h"
+
+namespace mojo {
+namespace system {
+namespace test {
+
+MojoResult WaitIfNecessary(scoped_refptr<MessagePipe> mp,
+                           MojoHandleSignals signals,
+                           HandleSignalsState* signals_state);
+
+class ChannelThread {
+ public:
+  explicit ChannelThread(embedder::PlatformSupport* platform_support);
+  ~ChannelThread();
+
+  void Start(embedder::ScopedPlatformHandle platform_handle,
+             scoped_refptr<MessagePipe> message_pipe);
+  void Stop();
+
+ private:
+  void InitChannelOnIOThread(embedder::ScopedPlatformHandle platform_handle,
+                             scoped_refptr<MessagePipe> message_pipe);
+  void ShutdownChannelOnIOThread();
+
+  embedder::PlatformSupport* const platform_support_;
+  TestIOThread test_io_thread_;
+  scoped_refptr<Channel> channel_;
+
+  DISALLOW_COPY_AND_ASSIGN(ChannelThread);
+};
+
+class MultiprocessMessagePipeTestBase : public testing::Test {
+ public:
+  MultiprocessMessagePipeTestBase();
+  virtual ~MultiprocessMessagePipeTestBase();
+
+ protected:
+  void Init(scoped_refptr<MessagePipe> mp);
+
+  embedder::PlatformSupport* platform_support() { return &platform_support_; }
+  mojo::test::MultiprocessTestHelper* helper() { return &helper_; }
+
+ private:
+  embedder::SimplePlatformSupport platform_support_;
+  ChannelThread channel_thread_;
+  mojo::test::MultiprocessTestHelper helper_;
+
+  DISALLOW_COPY_AND_ASSIGN(MultiprocessMessagePipeTestBase);
+};
+
+}  // namespace test
+}  // namespace system
+}  // namespace mojo
+
+#endif  // MOJO_SYSTEM_MESSAGE_PIPE_TEST_HELPER_H_