Add a SharedMemSynchronizer class.

media/audio/cross_process_notification_unittest.cc

+// Copyright (c) 2012 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 "base/compiler_specific.h"
+#include "base/logging.h"
+#include "base/shared_memory.h"
+#include "base/stl_util.h"
+#include "base/test/multiprocess_test.h"
+#include "base/threading/platform_thread.h"
+#include "media/audio/cross_process_notification.h"
+#include "testing/gtest/include/gtest/gtest.h"
+#include "testing/multiprocess_func_list.h"
+
+#if defined(OS_POSIX)
+#include <utility>  // NOLINT
+#endif
+
+namespace {
+
+// Initializes (ctor) and deletes (dtor) two vectors of pairs of
+// CrossProcessNotification instances.
+class NotificationsOwner {
+ public:
+  // Attempts to create up to |number_of_pairs| number of pairs. Call size()
+  // after construction to find out how many pairs were actually created.
+  explicit NotificationsOwner(size_t number_of_pairs) {
+    CreateMultiplePairs(number_of_pairs);
+  }
+  ~NotificationsOwner() {
+    STLDeleteElements(&a_);
+    STLDeleteElements(&b_);
+  }
+
+  size_t size() const {
+    DCHECK_EQ(a_.size(), b_.size());
+    return a_.size();
+  }
+
+  const CrossProcessNotification::Notifications& a() { return a_; }
+  const CrossProcessNotification::Notifications& b() { return b_; }
+
+ private:
+  void CreateMultiplePairs(size_t count) {
+    a_.resize(count);
+    b_.resize(count);
+    size_t i = 0;
+    for (; i < count; ++i) {
+      a_[i] = new CrossProcessNotification();
+      b_[i] = new CrossProcessNotification();
+      if (!CrossProcessNotification::InitializePair(a_[i], b_[i])) {
+        LOG(WARNING) << "InitializePair failed at " << i;
+        delete a_[i];
+        delete b_[i];
+        break;
+      }
+    }
+    a_.resize(i);
+    b_.resize(i);
+  }
+
+  CrossProcessNotification::Notifications a_;
+  CrossProcessNotification::Notifications b_;
+};
+
+// A simple thread that we'll run two instances of.  Both threads get a pointer
+// to the same |shared_data| and use a CrossProcessNotification to control when
+// each thread can read/write.
+class SingleNotifierWorker : public base::PlatformThread::Delegate {
+ public:
+  SingleNotifierWorker(size_t* shared_data, size_t repeats,
+                       CrossProcessNotification* notifier)
+      : shared_data_(shared_data), repeats_(repeats),
+        notifier_(notifier) {
+  }
+  virtual ~SingleNotifierWorker() {}
+
+  virtual void ThreadMain() OVERRIDE {
+    for (size_t i = 0; i < repeats_; ++i) {
+      notifier_->Wait();
+      ++(*shared_data_);
+      notifier_->Signal();
+    }
+  }
+
+ private:
+  size_t* shared_data_;
+  size_t repeats_;
+  CrossProcessNotification* notifier_;
+  DISALLOW_COPY_AND_ASSIGN(SingleNotifierWorker);
+};
+
+// Similar to SingleNotifierWorker, except each instance of this class will
+// have >1 instances of CrossProcessNotification to Wait/Signal and an equal
+// amount of |shared_data| that the notifiers control access to.
+class MultiNotifierWorker : public base::PlatformThread::Delegate {
+ public:
+  MultiNotifierWorker(size_t* shared_data, size_t repeats,
+      const CrossProcessNotification::Notifications* notifiers)
+      : shared_data_(shared_data), repeats_(repeats),
+        notifiers_(notifiers) {
+  }
+  virtual ~MultiNotifierWorker() {}
+
+  virtual void ThreadMain() OVERRIDE {
+    CrossProcessNotification::WaitForMultiple waiter(notifiers_);
+    for (size_t i = 0; i < repeats_; ++i) {
+      int signaled = waiter.Wait();
+      ++shared_data_[signaled];
+      (*notifiers_)[signaled]->Signal();
+    }
+  }
+
+ private:
+  size_t* shared_data_;
+  size_t repeats_;
+  const CrossProcessNotification::Notifications* notifiers_;
+  size_t count_;
+  DISALLOW_COPY_AND_ASSIGN(MultiNotifierWorker);
+};
+
+// A fixed array of bool flags.  Each flag uses 1 bit.  Use sizeof(FlagArray)
+// to determine how much memory you need.  The number of flags will therefore
+// be sizeof(FlagArray) * 8.
+// We use 'struct' to signify that this structures represents compiler
+// independent structured data.  I.e. you must be able to map this class
+// to a piece of shared memory of size sizeof(FlagArray) and be able to
+// use the class.  No vtables etc.
+// TODO(tommi): Move this to its own header when we start using it for signaling
+// audio devices.  As is, it's just here for perf comparison against the
+// "multiple notifiers" approach.
+struct FlagArray {
+ public:
+  FlagArray() : flags_() {}
+
+  bool is_set(size_t index) const {
+    return (flags_[index >> 5] & (1 << (index & 31))) ? true : false;
+  }
+
+  void set(size_t index) {
+    flags_[index >> 5] |= (1U << (static_cast<uint32>(index) & 31));
+  }
+
+  void clear(size_t index) {
+    flags_[index >> 5] &= ~(1U << (static_cast<uint32>(index) & 31));
+  }
+
+  // Returns the number of flags that can be set/checked.
+  size_t size() const { return sizeof(flags_) * 8; }
+
+ private:
+  // 256 * 32 = 8192 flags in 1KB.
+  uint32 flags_[256];
+  DISALLOW_COPY_AND_ASSIGN(FlagArray);
+};
+
+class MultiNotifierWorkerFlagArray : public base::PlatformThread::Delegate {
+ public:
+  MultiNotifierWorkerFlagArray(size_t count, FlagArray* signals,
+                               size_t* shared_data, size_t repeats,
+                               CrossProcessNotification* notifier)
+      : count_(count), signals_(signals), shared_data_(shared_data),
+        repeats_(repeats), notifier_(notifier) {
+  }
+  virtual ~MultiNotifierWorkerFlagArray() {}
+
+  virtual void ThreadMain() OVERRIDE {
+    for (size_t i = 0; i < repeats_; ++i) {
+      notifier_->Wait();
+      for (size_t s = 0; s < count_; ++s) {
+        if (signals_->is_set(s)) {
+          ++shared_data_[s];
+          // We don't clear the flag here but simply leave it signaled because
+          // we want the other thread to also increment this variable.
+        }
+      }
+      notifier_->Signal();
+    }
+  }
+
+ private:
+  size_t count_;
+  FlagArray* signals_;
+  size_t* shared_data_;
+  size_t repeats_;
+  CrossProcessNotification* notifier_;
+  DISALLOW_COPY_AND_ASSIGN(MultiNotifierWorkerFlagArray);
+};
+
+}  // end namespace
+
+TEST(CrossProcessNotification, FlagArray) {
+  FlagArray flags;
+  EXPECT_GT(flags.size(), 1000U);
+  for (size_t i = 0; i < flags.size(); ++i) {
+    EXPECT_FALSE(flags.is_set(i));
+    flags.set(i);
+    EXPECT_TRUE(flags.is_set(i));
+    flags.clear(i);
+    EXPECT_FALSE(flags.is_set(i));
+  }
+}
+
+// Initializes two notifiers, signals the each one and make sure the others
+// wait is satisfied.
+TEST(CrossProcessNotification, Basic) {
+  CrossProcessNotification a, b;
+  ASSERT_TRUE(CrossProcessNotification::InitializePair(&a, &b));
+  EXPECT_TRUE(a.IsValid());
+  EXPECT_TRUE(b.IsValid());
+
+  a.Signal();
+  b.Wait();
+
+  b.Signal();
+  a.Wait();
+}
+
+// Spins two worker threads, each with their own CrossProcessNotification
+// that they use to read and write from a shared memory buffer.
+TEST(CrossProcessNotification, TwoThreads) {
+  CrossProcessNotification a, b;
+  ASSERT_TRUE(CrossProcessNotification::InitializePair(&a, &b));
+
+  size_t data = 0;
+  const size_t kRepeats = 10000;
+  SingleNotifierWorker worker1(&data, kRepeats, &a);
+  SingleNotifierWorker worker2(&data, kRepeats, &b);
+  base::PlatformThreadHandle thread1, thread2;
+  base::PlatformThread::Create(0, &worker1, &thread1);
+  base::PlatformThread::Create(0, &worker2, &thread2);
+
+  // Start the first thread.  They should ping pong a few times and take turns
+  // incrementing the shared variable and never step on each other's toes.
+  a.Signal();
+
+  base::PlatformThread::Join(thread1);
+  base::PlatformThread::Join(thread2);
+
+  EXPECT_EQ(kRepeats * 2, data);
+}
+
+// Uses a pair of threads to access up to 1000 pieces of synchronized shared
+// data. On regular dev machines, the number of notifiers should be 1000, but on
+// mac and linux bots, the number will be smaller due to the RLIMIT_NOFILE
+// limit. Specifically, linux will have this limit at 1024 which means for this
+// test that the max number of notifiers will be in the range 500-512. On Mac
+// the limit is 256, so |count| will be ~120.  Oh, and raising the limit via
+// setrlimit() won't work.
+TEST(CrossProcessNotification, ThousandNotifiersTwoThreads) {
+  const size_t kCount = 1000;
+  NotificationsOwner pairs(kCount);
+  size_t data[kCount] = {0};
+  // We use a multiple of the count so that the division in the check below
+  // will be nice and round.
+  size_t repeats = pairs.size() * 1;
+
+  MultiNotifierWorker worker_1(&data[0], repeats, &pairs.a());
+  MultiNotifierWorker worker_2(&data[0], repeats, &pairs.b());
+  base::PlatformThreadHandle thread_1, thread_2;
+  base::PlatformThread::Create(0, &worker_1, &thread_1);
+  base::PlatformThread::Create(0, &worker_2, &thread_2);
+
+  for (size_t i = 0; i < pairs.size(); ++i)
+    pairs.a()[i]->Signal();
+
+  base::PlatformThread::Join(thread_1);
+  base::PlatformThread::Join(thread_2);
+
+  size_t expected_total = pairs.size() * 2;
+  size_t total = 0;
+  for (size_t i = 0; i < pairs.size(); ++i) {
+    // The CrossProcessNotification::WaitForMultiple class should have ensured
+    // that all notifiers had the same quality of service.
+    EXPECT_EQ(expected_total / pairs.size(), data[i]);
+    total += data[i];
+  }
+  EXPECT_EQ(expected_total, total);
+}
+
+// Functionally equivalent (as far as the shared data goes) to the
+// ThousandNotifiersTwoThreads test but uses a single pair of notifiers +
+// FlagArray for the 1000 signals. This approach is significantly faster.
+TEST(CrossProcessNotification, TwoNotifiersTwoThreads1000Signals) {
+  CrossProcessNotification a, b;
+  ASSERT_TRUE(CrossProcessNotification::InitializePair(&a, &b));
+
+  const size_t kCount = 1000;
+  FlagArray signals;
+  ASSERT_GE(signals.size(), kCount);
+  size_t data[kCount] = {0};
+
+  // Since this algorithm checks all events each time the notifier is
+  // signaled, |repeat| doesn't mean the same thing here as it does in
+  // ThousandNotifiersTwoThreads.  1 repeat here is the same as kCount
+  // repeats in ThousandNotifiersTwoThreads.
+  size_t repeats = 1;
+  MultiNotifierWorkerFlagArray worker1(kCount, &signals, &data[0], repeats, &a);
+  MultiNotifierWorkerFlagArray worker2(kCount, &signals, &data[0], repeats, &b);
+  base::PlatformThreadHandle thread1, thread2;
+  base::PlatformThread::Create(0, &worker1, &thread1);
+  base::PlatformThread::Create(0, &worker2, &thread2);
+
+  for (size_t i = 0; i < kCount; ++i)
+    signals.set(i);
+  a.Signal();
+
+  base::PlatformThread::Join(thread1);
+  base::PlatformThread::Join(thread2);
+
+  size_t expected_total = kCount * 2;
+  size_t total = 0;
+  for (size_t i = 0; i < kCount; ++i) {
+    // Since for each signal, we process all signaled events, the shared data
+    // variables should all be equal.
+    EXPECT_EQ(expected_total / kCount, data[i]);
+    total += data[i];
+  }
+  EXPECT_EQ(expected_total, total);
+}
+
+// Test the maximum number of notifiers without spinning further wait
+// threads on Windows. This test assumes we can always create 64 pairs and
+// bails if we can't.
+TEST(CrossProcessNotification, MultipleWaits64) {
+  const size_t kCount = 64;
+  NotificationsOwner pairs(kCount);
+  ASSERT_TRUE(pairs.size() == kCount);
+
+  CrossProcessNotification::WaitForMultiple waiter(&pairs.b());
+  for (size_t i = 0; i < kCount; ++i) {
+    pairs.a()[i]->Signal();
+    int index = waiter.Wait();
+    EXPECT_EQ(i, static_cast<size_t>(index));
+  }
+}
+
+// Tests waiting for more notifiers than the OS supports on one thread.
+// The test will create at most 1000 pairs, but on mac/linux bots the actual
+// number will be lower.  See comment about the RLIMIT_NOFILE limit above for
+// more details.
+TEST(CrossProcessNotification, MultipleWaits1000) {
+  // A 1000 notifiers requires 16 threads on Windows, including the current
+  // one, to perform the wait operation.
+  const size_t kCount = 1000;
+  NotificationsOwner pairs(kCount);
+
+  for (size_t i = 0; i < pairs.size(); ++i) {
+    pairs.a()[i]->Signal();
+    // To disable the load distribution algorithm and force the extra worker
+    // thread(s) to catch the signaled event, we define the |waiter| inside
+    // the loop.
+    CrossProcessNotification::WaitForMultiple waiter(&pairs.b());
+    int index = waiter.Wait();
+    EXPECT_EQ(i, static_cast<size_t>(index));
+  }
+}
+
+class CrossProcessNotificationMultiProcessTest : public base::MultiProcessTest {
+ public:
+  static const char kSharedMemName[];
+  static const size_t kSharedMemSize = 1024;
+
+ protected:
+  virtual void SetUp() OVERRIDE {
+    base::MultiProcessTest::SetUp();
+  }
+
+  virtual void TearDown() OVERRIDE {
+    base::MultiProcessTest::TearDown();
+  }
+};
+
+// static
+const char CrossProcessNotificationMultiProcessTest::kSharedMemName[] =
+    "CrossProcessNotificationMultiProcessTest";
+
+namespace {
+// A very crude IPC mechanism that we use to set up the spawned child process
+// and the parent process.
+struct CrudeIpc {
+  uint8 ready;
+  CrossProcessNotification::IPCHandle handle_1;
+  CrossProcessNotification::IPCHandle handle_2;
+};
+}  // end namespace
+
+// The main routine of the child process.  Waits for the parent process
+// to copy handles over to the child and then uses a CrossProcessNotification to
+// wait and signal to the parent process.
+MULTIPROCESS_TEST_MAIN(CrossProcessNotificationChildMain) {
+  base::SharedMemory mem;
+  bool ok = mem.CreateNamed(
+      CrossProcessNotificationMultiProcessTest::kSharedMemName,
+      true,
+      CrossProcessNotificationMultiProcessTest::kSharedMemSize);
+  DCHECK(ok);
+  if (!ok) {
+    LOG(ERROR) << "Failed to open shared memory segment.";
+    return -1;
+  }
+
+  mem.Map(CrossProcessNotificationMultiProcessTest::kSharedMemSize);
+  CrudeIpc* ipc = reinterpret_cast<CrudeIpc*>(mem.memory());
+
+  while (!ipc->ready)
+    base::PlatformThread::Sleep(10);
+
+  CrossProcessNotification notifier(ipc->handle_1, ipc->handle_2);
+  notifier.Wait();
+  notifier.Signal();
+
+  return 0;
+}
+
+// Spawns a new process and hands a CrossProcessNotification instance to the
+// new process.  Once that's done, it waits for the child process to signal
+// it's end and quits.
+TEST_F(CrossProcessNotificationMultiProcessTest, Basic) {
+  base::SharedMemory mem;
+  mem.Delete(kSharedMemName);  // In case a previous run was unsuccessful.
+  bool ok = mem.CreateNamed(kSharedMemName, false, kSharedMemSize);
+  ASSERT_TRUE(ok);
+
+  ASSERT_TRUE(mem.Map(kSharedMemSize));
+
+  CrossProcessNotification a, b;
+  ASSERT_TRUE(CrossProcessNotification::InitializePair(&a, &b));
+  EXPECT_TRUE(a.IsValid());
+  EXPECT_TRUE(b.IsValid());
+
+  CrudeIpc* ipc = reinterpret_cast<CrudeIpc*>(mem.memory());
+  ipc->ready = false;
+
+#if defined(OS_POSIX)
+  const int kPosixChildSocket = 20;
+  EXPECT_TRUE(b.ShareToProcess(NULL, &ipc->handle_1, &ipc->handle_2));
+  base::FileHandleMappingVector fd_mapping_vec;
+  fd_mapping_vec.push_back(std::pair<int, int>(ipc->handle_1.fd,
+                                               kPosixChildSocket));
+  ipc->handle_1.fd = kPosixChildSocket;
+  base::ProcessHandle process = SpawnChild("CrossProcessNotificationChildMain",
+                                           fd_mapping_vec, false);
+#else
+  base::ProcessHandle process = SpawnChild("CrossProcessNotificationChildMain",
+                                           false);
+  EXPECT_TRUE(b.ShareToProcess(process, &ipc->handle_1, &ipc->handle_2));
+#endif
+
+  ipc->ready = true;
+
+  a.Signal();
+  a.Wait();
+
+  int exit_code = -1;
+  base::WaitForExitCode(process, &exit_code);
+  EXPECT_EQ(0, exit_code);
+}