media/audio/cross_process_notification_unittest.cc

   1 // Copyright (c) 2012 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 "base/compiler_specific.h"
   6 #include "base/logging.h"
   7 #include "base/shared_memory.h"
   8 #include "base/stl_util.h"
   9 #include "base/test/multiprocess_test.h"
  10 #include "base/threading/platform_thread.h"
  11 #include "media/audio/cross_process_notification.h"
  12 #include "testing/gtest/include/gtest/gtest.h"
  13 #include "testing/multiprocess_func_list.h"
  14
  15 #if defined(OS_POSIX)
  16 #include <utility>  // NOLINT
  17 #endif
  18
  19 namespace {
  20
  21 // Initializes (ctor) and deletes (dtor) two vectors of pairs of
  22 // CrossProcessNotification instances.
  23 class NotificationsOwner {
  24  public:
  25   // Attempts to create up to |number_of_pairs| number of pairs. Call size()
  26   // after construction to find out how many pairs were actually created.
  27   explicit NotificationsOwner(size_t number_of_pairs) {
  28     CreateMultiplePairs(number_of_pairs);
  29   }
  30   ~NotificationsOwner() {
  31     STLDeleteElements(&a_);
  32     STLDeleteElements(&b_);
  33   }
  34
  35   size_t size() const {
  36     DCHECK_EQ(a_.size(), b_.size());
  37     return a_.size();
  38   }
  39
  40   const CrossProcessNotification::Notifications& a() { return a_; }
  41   const CrossProcessNotification::Notifications& b() { return b_; }
  42
  43  private:
  44   void CreateMultiplePairs(size_t count) {
  45     a_.resize(count);
  46     b_.resize(count);
  47     size_t i = 0;
  48     for (; i < count; ++i) {
  49       a_[i] = new CrossProcessNotification();
  50       b_[i] = new CrossProcessNotification();
  51       if (!CrossProcessNotification::InitializePair(a_[i], b_[i])) {
  52         LOG(WARNING) << "InitializePair failed at " << i;
  53         delete a_[i];
  54         delete b_[i];
  55         break;
  56       }
  57     }
  58     a_.resize(i);
  59     b_.resize(i);
  60   }
  61
  62   CrossProcessNotification::Notifications a_;
  63   CrossProcessNotification::Notifications b_;
  64 };
  65
  66 // A simple thread that we'll run two instances of.  Both threads get a pointer
  67 // to the same |shared_data| and use a CrossProcessNotification to control when
  68 // each thread can read/write.
  69 class SingleNotifierWorker : public base::PlatformThread::Delegate {
  70  public:
  71   SingleNotifierWorker(size_t* shared_data, size_t repeats,
  72                        CrossProcessNotification* notifier)
  73       : shared_data_(shared_data), repeats_(repeats),
  74         notifier_(notifier) {
  75   }
  76   virtual ~SingleNotifierWorker() {}
  77
  78   virtual void ThreadMain() OVERRIDE {
  79     for (size_t i = 0; i < repeats_; ++i) {
  80       notifier_->Wait();
  81       ++(*shared_data_);
  82       notifier_->Signal();
  83     }
  84   }
  85
  86  private:
  87   size_t* shared_data_;
  88   size_t repeats_;
  89   CrossProcessNotification* notifier_;
  90   DISALLOW_COPY_AND_ASSIGN(SingleNotifierWorker);
  91 };
  92
  93 // Similar to SingleNotifierWorker, except each instance of this class will
  94 // have >1 instances of CrossProcessNotification to Wait/Signal and an equal
  95 // amount of |shared_data| that the notifiers control access to.
  96 class MultiNotifierWorker : public base::PlatformThread::Delegate {
  97  public:
  98   MultiNotifierWorker(size_t* shared_data, size_t repeats,
  99       const CrossProcessNotification::Notifications* notifiers)
 100       : shared_data_(shared_data), repeats_(repeats),
 101         notifiers_(notifiers) {
 102   }
 103   virtual ~MultiNotifierWorker() {}
 104
 105   virtual void ThreadMain() OVERRIDE {
 106     CrossProcessNotification::WaitForMultiple waiter(notifiers_);
 107     for (size_t i = 0; i < repeats_; ++i) {
 108       int signaled = waiter.Wait();
 109       ++shared_data_[signaled];
 110       (*notifiers_)[signaled]->Signal();
 111     }
 112   }
 113
 114  private:
 115   size_t* shared_data_;
 116   size_t repeats_;
 117   const CrossProcessNotification::Notifications* notifiers_;
 118   size_t count_;
 119   DISALLOW_COPY_AND_ASSIGN(MultiNotifierWorker);
 120 };
 121
 122 // A fixed array of bool flags.  Each flag uses 1 bit.  Use sizeof(FlagArray)
 123 // to determine how much memory you need.  The number of flags will therefore
 124 // be sizeof(FlagArray) * 8.
 125 // We use 'struct' to signify that this structures represents compiler
 126 // independent structured data.  I.e. you must be able to map this class
 127 // to a piece of shared memory of size sizeof(FlagArray) and be able to
 128 // use the class.  No vtables etc.
 129 // TODO(tommi): Move this to its own header when we start using it for signaling
 130 // audio devices.  As is, it's just here for perf comparison against the
 131 // "multiple notifiers" approach.
 132 struct FlagArray {
 133  public:
 134   FlagArray() : flags_() {}
 135
 136   bool is_set(size_t index) const {
 137     return (flags_[index >> 5] & (1 << (index & 31))) ? true : false;
 138   }
 139
 140   void set(size_t index) {
 141     flags_[index >> 5] |= (1U << (static_cast<uint32>(index) & 31));
 142   }
 143
 144   void clear(size_t index) {
 145     flags_[index >> 5] &= ~(1U << (static_cast<uint32>(index) & 31));
 146   }
 147
 148   // Returns the number of flags that can be set/checked.
 149   size_t size() const { return sizeof(flags_) * 8; }
 150
 151  private:
 152   // 256 * 32 = 8192 flags in 1KB.
 153   uint32 flags_[256];
 154   DISALLOW_COPY_AND_ASSIGN(FlagArray);
 155 };
 156
 157 class MultiNotifierWorkerFlagArray : public base::PlatformThread::Delegate {
 158  public:
 159   MultiNotifierWorkerFlagArray(size_t count, FlagArray* signals,
 160                                size_t* shared_data, size_t repeats,
 161                                CrossProcessNotification* notifier)
 162       : count_(count), signals_(signals), shared_data_(shared_data),
 163         repeats_(repeats), notifier_(notifier) {
 164   }
 165   virtual ~MultiNotifierWorkerFlagArray() {}
 166
 167   virtual void ThreadMain() OVERRIDE {
 168     for (size_t i = 0; i < repeats_; ++i) {
 169       notifier_->Wait();
 170       for (size_t s = 0; s < count_; ++s) {
 171         if (signals_->is_set(s)) {
 172           ++shared_data_[s];
 173           // We don't clear the flag here but simply leave it signaled because
 174           // we want the other thread to also increment this variable.
 175         }
 176       }
 177       notifier_->Signal();
 178     }
 179   }
 180
 181  private:
 182   size_t count_;
 183   FlagArray* signals_;
 184   size_t* shared_data_;
 185   size_t repeats_;
 186   CrossProcessNotification* notifier_;
 187   DISALLOW_COPY_AND_ASSIGN(MultiNotifierWorkerFlagArray);
 188 };
 189
 190 }  // end namespace
 191
 192 TEST(CrossProcessNotification, FlagArray) {
 193   FlagArray flags;
 194   EXPECT_GT(flags.size(), 1000U);
 195   for (size_t i = 0; i < flags.size(); ++i) {
 196     EXPECT_FALSE(flags.is_set(i));
 197     flags.set(i);
 198     EXPECT_TRUE(flags.is_set(i));
 199     flags.clear(i);
 200     EXPECT_FALSE(flags.is_set(i));
 201   }
 202 }
 203
 204 // Initializes two notifiers, signals the each one and make sure the others
 205 // wait is satisfied.
 206 TEST(CrossProcessNotification, Basic) {
 207   CrossProcessNotification a, b;
 208   ASSERT_TRUE(CrossProcessNotification::InitializePair(&a, &b));
 209   EXPECT_TRUE(a.IsValid());
 210   EXPECT_TRUE(b.IsValid());
 211
 212   a.Signal();
 213   b.Wait();
 214
 215   b.Signal();
 216   a.Wait();
 217 }
 218
 219 // Spins two worker threads, each with their own CrossProcessNotification
 220 // that they use to read and write from a shared memory buffer.
 221 // Disabled as it trips of the TSAN bot (false positive since TSAN doesn't
 222 // recognize sockets as being a synchronization primitive).
 223 TEST(CrossProcessNotification, DISABLED_TwoThreads) {
 224   CrossProcessNotification a, b;
 225   ASSERT_TRUE(CrossProcessNotification::InitializePair(&a, &b));
 226
 227   size_t data = 0;
 228   const size_t kRepeats = 10000;
 229   SingleNotifierWorker worker1(&data, kRepeats, &a);
 230   SingleNotifierWorker worker2(&data, kRepeats, &b);
 231   base::PlatformThreadHandle thread1, thread2;
 232   base::PlatformThread::Create(0, &worker1, &thread1);
 233   base::PlatformThread::Create(0, &worker2, &thread2);
 234
 235   // Start the first thread.  They should ping pong a few times and take turns
 236   // incrementing the shared variable and never step on each other's toes.
 237   a.Signal();
 238
 239   base::PlatformThread::Join(thread1);
 240   base::PlatformThread::Join(thread2);
 241
 242   EXPECT_EQ(kRepeats * 2, data);
 243 }
 244
 245 // Uses a pair of threads to access up to 1000 pieces of synchronized shared
 246 // data. On regular dev machines, the number of notifiers should be 1000, but on
 247 // mac and linux bots, the number will be smaller due to the RLIMIT_NOFILE
 248 // limit. Specifically, linux will have this limit at 1024 which means for this
 249 // test that the max number of notifiers will be in the range 500-512. On Mac
 250 // the limit is 256, so |count| will be ~120.  Oh, and raising the limit via
 251 // setrlimit() won't work.
 252 // DISABLED since the distribution won't be accurate when run on valgrind.
 253 TEST(CrossProcessNotification, DISABLED_ThousandNotifiersTwoThreads) {
 254   const size_t kCount = 1000;
 255   NotificationsOwner pairs(kCount);
 256   size_t data[kCount] = {0};
 257   // We use a multiple of the count so that the division in the check below
 258   // will be nice and round.
 259   size_t repeats = pairs.size() * 1;
 260
 261   MultiNotifierWorker worker_1(&data[0], repeats, &pairs.a());
 262   MultiNotifierWorker worker_2(&data[0], repeats, &pairs.b());
 263   base::PlatformThreadHandle thread_1, thread_2;
 264   base::PlatformThread::Create(0, &worker_1, &thread_1);
 265   base::PlatformThread::Create(0, &worker_2, &thread_2);
 266
 267   for (size_t i = 0; i < pairs.size(); ++i)
 268     pairs.a()[i]->Signal();
 269
 270   base::PlatformThread::Join(thread_1);
 271   base::PlatformThread::Join(thread_2);
 272
 273   size_t expected_total = pairs.size() * 2;
 274   size_t total = 0;
 275   for (size_t i = 0; i < pairs.size(); ++i) {
 276     // The CrossProcessNotification::WaitForMultiple class should have ensured
 277     // that all notifiers had the same quality of service.
 278     EXPECT_EQ(expected_total / pairs.size(), data[i]);
 279     total += data[i];
 280   }
 281   EXPECT_EQ(expected_total, total);
 282 }
 283
 284 // Functionally equivalent (as far as the shared data goes) to the
 285 // ThousandNotifiersTwoThreads test but uses a single pair of notifiers +
 286 // FlagArray for the 1000 signals. This approach is significantly faster.
 287 // Disabled as it trips of the TSAN bot - "Possible data race during write of
 288 // size 4" (the flag array).
 289 TEST(CrossProcessNotification, DISABLED_TwoNotifiersTwoThreads1000Signals) {
 290   CrossProcessNotification a, b;
 291   ASSERT_TRUE(CrossProcessNotification::InitializePair(&a, &b));
 292
 293   const size_t kCount = 1000;
 294   FlagArray signals;
 295   ASSERT_GE(signals.size(), kCount);
 296   size_t data[kCount] = {0};
 297
 298   // Since this algorithm checks all events each time the notifier is
 299   // signaled, |repeat| doesn't mean the same thing here as it does in
 300   // ThousandNotifiersTwoThreads.  1 repeat here is the same as kCount
 301   // repeats in ThousandNotifiersTwoThreads.
 302   size_t repeats = 1;
 303   MultiNotifierWorkerFlagArray worker1(kCount, &signals, &data[0], repeats, &a);
 304   MultiNotifierWorkerFlagArray worker2(kCount, &signals, &data[0], repeats, &b);
 305   base::PlatformThreadHandle thread1, thread2;
 306   base::PlatformThread::Create(0, &worker1, &thread1);
 307   base::PlatformThread::Create(0, &worker2, &thread2);
 308
 309   for (size_t i = 0; i < kCount; ++i)
 310     signals.set(i);
 311   a.Signal();
 312
 313   base::PlatformThread::Join(thread1);
 314   base::PlatformThread::Join(thread2);
 315
 316   size_t expected_total = kCount * 2;
 317   size_t total = 0;
 318   for (size_t i = 0; i < kCount; ++i) {
 319     // Since for each signal, we process all signaled events, the shared data
 320     // variables should all be equal.
 321     EXPECT_EQ(expected_total / kCount, data[i]);
 322     total += data[i];
 323   }
 324   EXPECT_EQ(expected_total, total);
 325 }
 326
 327 // Test the maximum number of notifiers without spinning further wait
 328 // threads on Windows. This test assumes we can always create 64 pairs and
 329 // bails if we can't.
 330 TEST(CrossProcessNotification, MultipleWaits64) {
 331   const size_t kCount = 64;
 332   NotificationsOwner pairs(kCount);
 333   ASSERT_TRUE(pairs.size() == kCount);
 334
 335   CrossProcessNotification::WaitForMultiple waiter(&pairs.b());
 336   for (size_t i = 0; i < kCount; ++i) {
 337     pairs.a()[i]->Signal();
 338     int index = waiter.Wait();
 339     EXPECT_EQ(i, static_cast<size_t>(index));
 340   }
 341 }
 342
 343 // Tests waiting for more notifiers than the OS supports on one thread.
 344 // The test will create at most 1000 pairs, but on mac/linux bots the actual
 345 // number will be lower.  See comment about the RLIMIT_NOFILE limit above for
 346 // more details.
 347 // DISABLED since the distribution won't be accurate when run on valgrind.
 348 TEST(CrossProcessNotification, DISABLED_MultipleWaits1000) {
 349   // A 1000 notifiers requires 16 threads on Windows, including the current
 350   // one, to perform the wait operation.
 351   const size_t kCount = 1000;
 352   NotificationsOwner pairs(kCount);
 353
 354   for (size_t i = 0; i < pairs.size(); ++i) {
 355     pairs.a()[i]->Signal();
 356     // To disable the load distribution algorithm and force the extra worker
 357     // thread(s) to catch the signaled event, we define the |waiter| inside
 358     // the loop.
 359     CrossProcessNotification::WaitForMultiple waiter(&pairs.b());
 360     int index = waiter.Wait();
 361     EXPECT_EQ(i, static_cast<size_t>(index));
 362   }
 363 }
 364
 365 class CrossProcessNotificationMultiProcessTest : public base::MultiProcessTest {
 366  public:
 367   static const char kSharedMemName[];
 368   static const size_t kSharedMemSize = 1024;
 369
 370  protected:
 371   virtual void SetUp() OVERRIDE {
 372     base::MultiProcessTest::SetUp();
 373   }
 374
 375   virtual void TearDown() OVERRIDE {
 376     base::MultiProcessTest::TearDown();
 377   }
 378 };
 379
 380 // static
 381 const char CrossProcessNotificationMultiProcessTest::kSharedMemName[] =
 382     "CrossProcessNotificationMultiProcessTest";
 383
 384 namespace {
 385 // A very crude IPC mechanism that we use to set up the spawned child process
 386 // and the parent process.
 387 struct CrudeIpc {
 388   uint8 ready;
 389   CrossProcessNotification::IPCHandle handle_1;
 390   CrossProcessNotification::IPCHandle handle_2;
 391 };
 392 }  // end namespace
 393
 394 // The main routine of the child process.  Waits for the parent process
 395 // to copy handles over to the child and then uses a CrossProcessNotification to
 396 // wait and signal to the parent process.
 397 MULTIPROCESS_TEST_MAIN(CrossProcessNotificationChildMain) {
 398   base::SharedMemory mem;
 399   bool ok = mem.CreateNamed(
 400       CrossProcessNotificationMultiProcessTest::kSharedMemName,
 401       true,
 402       CrossProcessNotificationMultiProcessTest::kSharedMemSize);
 403   DCHECK(ok);
 404   if (!ok) {
 405     LOG(ERROR) << "Failed to open shared memory segment.";
 406     return -1;
 407   }
 408
 409   mem.Map(CrossProcessNotificationMultiProcessTest::kSharedMemSize);
 410   CrudeIpc* ipc = reinterpret_cast<CrudeIpc*>(mem.memory());
 411
 412   while (!ipc->ready)
 413     base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(10));
 414
 415   CrossProcessNotification notifier(ipc->handle_1, ipc->handle_2);
 416   notifier.Wait();
 417   notifier.Signal();
 418
 419   return 0;
 420 }
 421
 422 // Spawns a new process and hands a CrossProcessNotification instance to the
 423 // new process.  Once that's done, it waits for the child process to signal
 424 // it's end and quits.
 425 TEST_F(CrossProcessNotificationMultiProcessTest, Basic) {
 426   base::SharedMemory mem;
 427   mem.Delete(kSharedMemName);  // In case a previous run was unsuccessful.
 428   bool ok = mem.CreateNamed(kSharedMemName, false, kSharedMemSize);
 429   ASSERT_TRUE(ok);
 430
 431   ASSERT_TRUE(mem.Map(kSharedMemSize));
 432
 433   CrossProcessNotification a, b;
 434   ASSERT_TRUE(CrossProcessNotification::InitializePair(&a, &b));
 435   EXPECT_TRUE(a.IsValid());
 436   EXPECT_TRUE(b.IsValid());
 437
 438   CrudeIpc* ipc = reinterpret_cast<CrudeIpc*>(mem.memory());
 439   ipc->ready = false;
 440
 441 #if defined(OS_POSIX)
 442   const int kPosixChildSocket = 20;
 443   EXPECT_TRUE(b.ShareToProcess(
 444         base::kNullProcessHandle, &ipc->handle_1, &ipc->handle_2));
 445   base::FileHandleMappingVector fd_mapping_vec;
 446   fd_mapping_vec.push_back(std::pair<int, int>(ipc->handle_1.fd,
 447                                                kPosixChildSocket));
 448   ipc->handle_1.fd = kPosixChildSocket;
 449   base::ProcessHandle process = SpawnChild("CrossProcessNotificationChildMain",
 450                                            fd_mapping_vec, false);
 451 #else
 452   base::ProcessHandle process = SpawnChild("CrossProcessNotificationChildMain",
 453                                            false);
 454   EXPECT_TRUE(b.ShareToProcess(process, &ipc->handle_1, &ipc->handle_2));
 455 #endif
 456
 457   ipc->ready = true;
 458
 459   a.Signal();
 460   a.Wait();
 461
 462   int exit_code = -1;
 463   base::WaitForExitCode(process, &exit_code);
 464   EXPECT_EQ(0, exit_code);
 465 }