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 #include <utility>  // NOLINT
  16
  17 namespace {
  18
  19 // Initializes (ctor) and deletes (dtor) two vectors of pairs of
  20 // CrossProcessNotification instances.
  21 class NotificationsOwner {
  22  public:
  23   // Attempts to create up to |number_of_pairs| number of pairs. Call size()
  24   // after construction to find out how many pairs were actually created.
  25   explicit NotificationsOwner(size_t number_of_pairs) {
  26     CreateMultiplePairs(number_of_pairs);
  27   }
  28   ~NotificationsOwner() {
  29     STLDeleteElements(&a_);
  30     STLDeleteElements(&b_);
  31   }
  32
  33   size_t size() const {
  34     DCHECK_EQ(a_.size(), b_.size());
  35     return a_.size();
  36   }
  37
  38   const CrossProcessNotification::Notifications& a() { return a_; }
  39   const CrossProcessNotification::Notifications& b() { return b_; }
  40
  41  private:
  42   void CreateMultiplePairs(size_t count) {
  43     a_.resize(count);
  44     b_.resize(count);
  45     size_t i = 0;
  46     for (; i < count; ++i) {
  47       a_[i] = new CrossProcessNotification();
  48       b_[i] = new CrossProcessNotification();
  49       if (!CrossProcessNotification::InitializePair(a_[i], b_[i])) {
  50         LOG(WARNING) << "InitializePair failed at " << i;
  51         delete a_[i];
  52         delete b_[i];
  53         break;
  54       }
  55     }
  56     a_.resize(i);
  57     b_.resize(i);
  58   }
  59
  60   CrossProcessNotification::Notifications a_;
  61   CrossProcessNotification::Notifications b_;
  62 };
  63
  64 // A simple thread that we'll run two instances of.  Both threads get a pointer
  65 // to the same |shared_data| and use a CrossProcessNotification to control when
  66 // each thread can read/write.
  67 class SingleNotifierWorker : public base::PlatformThread::Delegate {
  68  public:
  69   SingleNotifierWorker(size_t* shared_data, size_t repeats,
  70                        CrossProcessNotification* notifier)
  71       : shared_data_(shared_data), repeats_(repeats),
  72         notifier_(notifier) {
  73   }
  74   virtual ~SingleNotifierWorker() {}
  75
  76   // base::PlatformThread::Delegate:
  77   virtual void ThreadMain() OVERRIDE {
  78     for (size_t i = 0; i < repeats_; ++i) {
  79       notifier_->Wait();
  80       ++(*shared_data_);
  81       notifier_->Signal();
  82     }
  83   }
  84
  85  private:
  86   size_t* shared_data_;
  87   size_t repeats_;
  88   CrossProcessNotification* notifier_;
  89   DISALLOW_COPY_AND_ASSIGN(SingleNotifierWorker);
  90 };
  91
  92 // Similar to SingleNotifierWorker, except each instance of this class will
  93 // have >1 instances of CrossProcessNotification to Wait/Signal and an equal
  94 // amount of |shared_data| that the notifiers control access to.
  95 class MultiNotifierWorker : public base::PlatformThread::Delegate {
  96  public:
  97   MultiNotifierWorker(size_t* shared_data, size_t repeats,
  98       const CrossProcessNotification::Notifications* notifiers)
  99       : shared_data_(shared_data), repeats_(repeats),
 100         notifiers_(notifiers) {
 101   }
 102   virtual ~MultiNotifierWorker() {}
 103
 104   // base::PlatformThread::Delegate:
 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   DISALLOW_COPY_AND_ASSIGN(MultiNotifierWorker);
 119 };
 120
 121 // A fixed array of bool flags.  Each flag uses 1 bit.  Use sizeof(FlagArray)
 122 // to determine how much memory you need.  The number of flags will therefore
 123 // be sizeof(FlagArray) * 8.
 124 // We use 'struct' to signify that this structures represents compiler
 125 // independent structured data.  I.e. you must be able to map this class
 126 // to a piece of shared memory of size sizeof(FlagArray) and be able to
 127 // use the class.  No vtables etc.
 128 // TODO(tommi): Move this to its own header when we start using it for signaling
 129 // audio devices.  As is, it's just here for perf comparison against the
 130 // "multiple notifiers" approach.
 131 struct FlagArray {
 132  public:
 133   FlagArray() : flags_() {}
 134
 135   bool is_set(size_t index) const {
 136     return (flags_[index >> 5] & (1 << (index & 31)));
 137   }
 138
 139   void set(size_t index) {
 140     flags_[index >> 5] |= (1U << (static_cast<uint32>(index) & 31));
 141   }
 142
 143   void clear(size_t index) {
 144     flags_[index >> 5] &= ~(1U << (static_cast<uint32>(index) & 31));
 145   }
 146
 147   // Returns the number of flags that can be set/checked.
 148   size_t size() const { return sizeof(flags_) * 8; }
 149
 150  private:
 151   // 256 * 32 = 8192 flags in 1KB.
 152   uint32 flags_[256];
 153   DISALLOW_COPY_AND_ASSIGN(FlagArray);
 154 };
 155
 156 class MultiNotifierWorkerFlagArray : public base::PlatformThread::Delegate {
 157  public:
 158   MultiNotifierWorkerFlagArray(size_t count, FlagArray* signals,
 159                                size_t* shared_data, size_t repeats,
 160                                CrossProcessNotification* notifier)
 161       : count_(count), signals_(signals), shared_data_(shared_data),
 162         repeats_(repeats), notifier_(notifier) {
 163   }
 164   virtual ~MultiNotifierWorkerFlagArray() {}
 165
 166   // base::PlatformThread::Delegate:
 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 };
 367
 368 namespace {
 369
 370 // A very crude IPC mechanism that we use to set up the spawned child process
 371 // and the parent process.
 372 struct CrudeIpc {
 373   uint8 ready;
 374   CrossProcessNotification::IPCHandle handle_1;
 375   CrossProcessNotification::IPCHandle handle_2;
 376 };
 377
 378 #if defined(OS_POSIX)
 379 const int kPosixChildSharedMem = 30;
 380 #else
 381 const char kSharedMemName[] = "CrossProcessNotificationMultiProcessTest";
 382 #endif
 383
 384 const size_t kSharedMemSize = 1024;
 385
 386 }  // namespace
 387
 388 // The main routine of the child process.  Waits for the parent process
 389 // to copy handles over to the child and then uses a CrossProcessNotification to
 390 // wait and signal to the parent process.
 391 MULTIPROCESS_TEST_MAIN(CrossProcessNotificationChildMain) {
 392 #if defined(OS_POSIX)
 393   base::SharedMemory mem(
 394       base::SharedMemoryHandle(kPosixChildSharedMem, true /* auto close */),
 395       false);
 396 #else
 397   base::SharedMemory mem;
 398   CHECK(mem.CreateNamed(kSharedMemName, true, kSharedMemSize));
 399 #endif
 400
 401   CHECK(mem.Map(kSharedMemSize));
 402   CrudeIpc* ipc = reinterpret_cast<CrudeIpc*>(mem.memory());
 403
 404   while (!ipc->ready)
 405     base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(10));
 406
 407   CrossProcessNotification notifier(ipc->handle_1, ipc->handle_2);
 408   notifier.Wait();
 409   notifier.Signal();
 410
 411   return 0;
 412 }
 413
 414 // Spawns a new process and hands a CrossProcessNotification instance to the
 415 // new process.  Once that's done, it waits for the child process to signal
 416 // it's end and quits.
 417 TEST_F(CrossProcessNotificationMultiProcessTest, Basic) {
 418   CrossProcessNotification a, b;
 419   ASSERT_TRUE(CrossProcessNotification::InitializePair(&a, &b));
 420   EXPECT_TRUE(a.IsValid());
 421   EXPECT_TRUE(b.IsValid());
 422
 423   base::SharedMemory mem;
 424
 425 #if defined(OS_POSIX)
 426   ASSERT_TRUE(mem.CreateAndMapAnonymous(kSharedMemSize));
 427 #else
 428   mem.Delete(kSharedMemName);  // In case a previous run was unsuccessful.
 429   ASSERT_TRUE(mem.CreateNamed(kSharedMemName, false, kSharedMemSize));
 430   ASSERT_TRUE(mem.Map(kSharedMemSize));
 431 #endif
 432
 433   CrudeIpc* ipc = reinterpret_cast<CrudeIpc*>(mem.memory());
 434   ipc->ready = false;
 435
 436 #if defined(OS_POSIX)
 437   const int kPosixChildSocket = 20;
 438   EXPECT_TRUE(b.ShareToProcess(
 439         base::kNullProcessHandle, &ipc->handle_1, &ipc->handle_2));
 440   base::FileHandleMappingVector fd_mapping_vec;
 441   fd_mapping_vec.push_back(std::make_pair(ipc->handle_1.fd, kPosixChildSocket));
 442   fd_mapping_vec.push_back(
 443       std::make_pair(mem.handle().fd, kPosixChildSharedMem));
 444   ipc->handle_1.fd = kPosixChildSocket;
 445   base::ProcessHandle process = SpawnChild("CrossProcessNotificationChildMain",
 446                                            fd_mapping_vec, false);
 447 #else
 448   base::ProcessHandle process = SpawnChild("CrossProcessNotificationChildMain",
 449                                            false);
 450   EXPECT_TRUE(b.ShareToProcess(process, &ipc->handle_1, &ipc->handle_2));
 451 #endif
 452
 453   ipc->ready = true;
 454
 455   a.Signal();
 456   a.Wait();
 457
 458   int exit_code = -1;
 459   base::WaitForExitCode(process, &exit_code);
 460   EXPECT_EQ(0, exit_code);
 461 }