Mojo: add mojo_message_pipe_perftests

mojo/system/message_pipe_perftest.cc

+// Copyright 2013 The Chromium Authors. All rights reserved.

[viettrungluu, 2014/08/25 20:32:45]

nit: 2014

[Hajime Morrita, 2014/08/26 18:16:26]

Done.

+// 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 "build/build_config.h"  // TODO(vtl): Remove this.

[viettrungluu, 2014/08/25 20:32:45]

I don't think you need this include.

[Hajime Morrita, 2014/08/26 18:16:26]

Done.

+#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) {

[viettrungluu, 2014/08/25 20:32:46]

nit: "Setup" -> "SetUp" (the verb is always "set up", as two words).

[Hajime Morrita, 2014/08/26 18:16:26]

Done.

+    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()),
+                              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, 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(PerfClient) {
+  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;
+}
+
+// Sends a bunch of messages to the child. Expects them "repeated" back. Waits

[viettrungluu, 2014/08/25 20:32:45]

This description isn't entirely accurate. This test is really a latency test,
since it waits for the reply before sending the next one.

(Also, in contrast to ipc_perf_tests, this test necessitates a thread-hop on the
receive side, whereas ipc_perf_tests does everything on the I/O thread. This
isn't entirely unfair, the Mojo design doesn't currently really let you do
anything on the I/O thread, but it kind of pits the native IPC's best case
(handle everything on the I/O -- or test main -- thread) versus the Mojo normal
case. It'd be interesting to have a version of the native IPC where the
sender/receiver are on a different thread.)

[Hajime Morrita, 2014/08/26 18:16:26]

That's true. Updated the comment. I'll add ChannelProxy-based benchmark next.

+// for the child to close its end before quitting.
+TEST_F(MultiprocessMessagePipePerfTest, Performance) {

[viettrungluu, 2014/08/25 20:32:45]

Possibly you should have a slightly more descriptive name than "Performance".
Maybe it should be "PingPong" or something like that.

[Hajime Morrita, 2014/08/26 18:16:26]

Done.

+  helper()->StartChild("PerfClient");
+
+  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