Mojo: Add a threaded perf test for (local) message pipes.

mojo/public/tests/system/core_perftest.cc

Index: mojo/public/tests/system/core_perftest.cc
diff --git a/mojo/public/tests/system/core_perftest.cc b/mojo/public/tests/system/core_perftest.cc
index bfb8ff36a568adc3b6d2496ae9a99e2f00034bc6..6874375bdadb0fc046094894ab34318ef28263de 100644
--- a/mojo/public/tests/system/core_perftest.cc
+++ b/mojo/public/tests/system/core_perftest.cc
@@ -7,13 +7,112 @@
 #include "mojo/public/system/core.h"
 
 #include <assert.h>
+#include <stddef.h>
+#include <stdint.h>
+#include <stdio.h>
 
 #include "mojo/public/system/macros.h"
+#include "mojo/public/tests/test_support.h"
 #include "mojo/public/tests/test_utils.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
+// TODO(vtl): (here and below) crbug.com/342893
+#if !defined(WIN32)
+#include <time.h>
+#include "mojo/public/utility/thread.h"
+#endif  // !defined(WIN32)
+
 namespace {
 
+#if !defined(WIN32)
+class MessagePipeWriterThread : public mojo::Thread {
+ public:
+  MessagePipeWriterThread(MojoHandle handle, uint32_t num_bytes)
+      : handle_(handle),
+        num_bytes_(num_bytes),
+        num_writes_(0) {}
+  virtual ~MessagePipeWriterThread() {}
+
+  virtual void Run() MOJO_OVERRIDE {
+    char buffer[10000];
+    assert(num_bytes_ <= sizeof(buffer));
+
+    // TODO(vtl): Should I throttle somehow?
+    for (;;) {
+      MojoResult result = MojoWriteMessage(handle_, buffer, num_bytes_, NULL, 0,
+                                           MOJO_WRITE_MESSAGE_FLAG_NONE);
+      if (result == MOJO_RESULT_OK) {
+        num_writes_++;
+        continue;
+      }
+
+      // We failed to write.
+      // Either |handle_| or its peer was closed.
+      assert(result == MOJO_RESULT_INVALID_ARGUMENT ||
+             result == MOJO_RESULT_FAILED_PRECONDITION);
+      break;
+    }
+  }
+
+  // Use only after joining the thread.
+  int64_t num_writes() const { return num_writes_; }
+
+ private:
+  const MojoHandle handle_;
+  const uint32_t num_bytes_;
+  int64_t num_writes_;
+
+  MOJO_DISALLOW_COPY_AND_ASSIGN(MessagePipeWriterThread);
+};
+
+class MessagePipeReaderThread : public mojo::Thread {
+ public:
+  explicit MessagePipeReaderThread(MojoHandle handle)
+      : handle_(handle),
+        num_reads_(0) {
+  }
+  virtual ~MessagePipeReaderThread() {}
+
+  virtual void Run() MOJO_OVERRIDE {
+    char buffer[10000];
+
+    for (;;) {
+      uint32_t num_bytes = static_cast<uint32_t>(sizeof(buffer));
+      MojoResult result = MojoReadMessage(handle_, buffer, &num_bytes, NULL,
+                                          NULL, MOJO_READ_MESSAGE_FLAG_NONE);
+      if (result == MOJO_RESULT_OK) {
+        num_reads_++;
+        continue;
+      }
+
+      if (result == MOJO_RESULT_SHOULD_WAIT) {
+        result = MojoWait(handle_, MOJO_WAIT_FLAG_READABLE,
+                          MOJO_DEADLINE_INDEFINITE);
+        if (result == MOJO_RESULT_OK) {
+          // Go to the top of the loop to read again.
+          continue;
+        }
+      }
+
+      // We failed to read and possibly failed to wait.
+      // Either |handle_| or its peer was closed.
+      assert(result == MOJO_RESULT_INVALID_ARGUMENT ||
+             result == MOJO_RESULT_FAILED_PRECONDITION);
+      break;
+    }
+  }
+
+  // Use only after joining the thread.
+  int64_t num_reads() const { return num_reads_; }
+
+ private:
+  const MojoHandle handle_;
+  int64_t num_reads_;
+
+  MOJO_DISALLOW_COPY_AND_ASSIGN(MessagePipeReaderThread);
+};
+#endif  // !defined(WIN32)
+
 class CorePerftest : public testing::Test {
  public:
   CorePerftest() : buffer_(NULL), num_bytes_(0) {}
@@ -60,6 +159,85 @@ class CorePerftest : public testing::Test {
   }
 
  protected:
+#if !defined(WIN32)
+  void DoMessagePipeThreadedTest(unsigned num_writers,
+                                 unsigned num_readers,
+                                 uint32_t num_bytes) {
+    static const int64_t kPerftestTimeMicroseconds = 3 * 1000000;
+
+    assert(num_writers > 0);
+    assert(num_readers > 0);
+
+    MojoResult result MOJO_ALLOW_UNUSED;
+    result = MojoCreateMessagePipe(&h0_, &h1_);
+    assert(result == MOJO_RESULT_OK);
+
+    std::vector<MessagePipeWriterThread*> writers;
+    for (unsigned i = 0; i < num_writers; i++)
+      writers.push_back(new MessagePipeWriterThread(h0_, num_bytes));
+
+    std::vector<MessagePipeReaderThread*> readers;
+    for (unsigned i = 0; i < num_readers; i++)
+      readers.push_back(new MessagePipeReaderThread(h1_));
+
+    // Start time here, just before we fire off the threads.
+    const MojoTimeTicks start_time = MojoGetTimeTicksNow();
+
+    // Interleave the starts.
+    for (unsigned i = 0; i < num_writers || i < num_readers; i++) {
+      if (i < num_writers)
+        writers[i]->Start();
+      if (i < num_readers)
+        readers[i]->Start();
+    }
+
+    Sleep(kPerftestTimeMicroseconds);
+
+    // Close both handles to make writers and readers stop immediately.
+    result = MojoClose(h0_);
+    assert(result == MOJO_RESULT_OK);
+    result = MojoClose(h1_);
+    assert(result == MOJO_RESULT_OK);
+
+    // Join everything.
+    for (unsigned i = 0; i < num_writers; i++)
+      writers[i]->Join();
+    for (unsigned i = 0; i < num_readers; i++)
+      readers[i]->Join();
+
+    // Stop time here.
+    MojoTimeTicks end_time = MojoGetTimeTicksNow();
+
+    // Add up write and read counts, and destroy the threads.
+    int64_t num_writes = 0;
+    for (unsigned i = 0; i < num_writers; i++) {
+      num_writes += writers[i]->num_writes();
+      delete writers[i];
+    }
+    writers.clear();
+    int64_t num_reads = 0;
+    for (unsigned i = 0; i < num_readers; i++) {
+      num_reads += readers[i]->num_reads();
+      delete readers[i];
+    }
+    readers.clear();
+
+    char test_name[200];
+    sprintf(test_name, "MessagePipe_Threaded_Writes_%uw_%ur_%ubytes",
+            num_writers, num_readers, static_cast<unsigned>(num_bytes));
+    mojo::test::LogPerfResult(test_name,
+                              1000000.0 * static_cast<double>(num_writes) /
+                                  (end_time - start_time),
+                              "writes/second");
+    sprintf(test_name, "MessagePipe_Threaded_Reads_%uw_%ur_%ubytes",
+            num_writers, num_readers, static_cast<unsigned>(num_bytes));
+    mojo::test::LogPerfResult(test_name,
+                              1000000.0 * static_cast<double>(num_reads) /
+                                  (end_time - start_time),
+                                  "reads/second");
+  }
+#endif  // !defined(WIN32)
+
   MojoHandle h0_;
   MojoHandle h1_;
 
@@ -67,6 +245,18 @@ class CorePerftest : public testing::Test {
   uint32_t num_bytes_;
 
  private:
+#if !defined(WIN32)
+  void Sleep(int64_t microseconds) {
+    struct timespec req = {
+      static_cast<time_t>(microseconds / 1000000),  // Seconds.
+      static_cast<long>(microseconds % 1000000) * 1000L  // Nanoseconds.
+    };
+    int rv MOJO_ALLOW_UNUSED;
+    rv = nanosleep(&req, NULL);
+    assert(rv == 0);
+  }
+#endif  // !defined(WIN32)
+
   MOJO_DISALLOW_COPY_AND_ASSIGN(CorePerftest);
 };
 
@@ -122,4 +312,26 @@ TEST_F(CorePerftest, MessagePipe_EmptyRead) {
   assert(result == MOJO_RESULT_OK);
 }
 
+#if !defined(WIN32)
+TEST_F(CorePerftest, MessagePipe_Threaded) {
+  DoMessagePipeThreadedTest(1u, 1u, 100u);
+  DoMessagePipeThreadedTest(2u, 2u, 100u);
+  DoMessagePipeThreadedTest(3u, 3u, 100u);
+  DoMessagePipeThreadedTest(10u, 10u, 100u);
+  DoMessagePipeThreadedTest(10u, 1u, 100u);
+  DoMessagePipeThreadedTest(1u, 10u, 100u);
+
+  // For comparison of overhead:
+  DoMessagePipeThreadedTest(1u, 1u, 10u);
+  // 100 was done above.
+  DoMessagePipeThreadedTest(1u, 1u, 1000u);
+  DoMessagePipeThreadedTest(1u, 1u, 10000u);
+
+  DoMessagePipeThreadedTest(3u, 3u, 10u);
+  // 100 was done above.
+  DoMessagePipeThreadedTest(3u, 3u, 1000u);
+  DoMessagePipeThreadedTest(3u, 3u, 10000u);
+}
+#endif  // !defined(WIN32)
+
 }  // namespace