src/mojo/system/dispatcher_unittest.cc

   1 // Copyright 2013 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 "mojo/system/dispatcher.h"
   6
   7 #include "base/basictypes.h"
   8 #include "base/memory/ref_counted.h"
   9 #include "base/memory/scoped_vector.h"
  10 #include "base/synchronization/waitable_event.h"
  11 #include "base/threading/simple_thread.h"
  12 #include "mojo/system/waiter.h"
  13 #include "testing/gtest/include/gtest/gtest.h"
  14
  15 namespace mojo {
  16 namespace system {
  17 namespace {
  18
  19 // Trivial subclass that makes the constructor public.
  20 class TrivialDispatcher : public Dispatcher {
  21  public:
  22   TrivialDispatcher() {}
  23
  24  private:
  25   friend class base::RefCountedThreadSafe<TrivialDispatcher>;
  26   virtual ~TrivialDispatcher() {}
  27
  28   DISALLOW_COPY_AND_ASSIGN(TrivialDispatcher);
  29 };
  30
  31 TEST(DispatcherTest, Basic) {
  32   scoped_refptr<Dispatcher> d(new TrivialDispatcher());
  33
  34   EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
  35             d->WriteMessage(NULL, 0, NULL, 0, MOJO_WRITE_MESSAGE_FLAG_NONE));
  36   EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
  37             d->ReadMessage(NULL, NULL, NULL, NULL,
  38                            MOJO_WRITE_MESSAGE_FLAG_NONE));
  39   Waiter w;
  40   w.Init();
  41   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
  42             d->AddWaiter(&w, MOJO_WAIT_FLAG_EVERYTHING, 0));
  43   // Okay to remove even if it wasn't added (or was already removed).
  44   d->RemoveWaiter(&w);
  45   d->RemoveWaiter(&w);
  46
  47   EXPECT_EQ(MOJO_RESULT_OK, d->Close());
  48
  49   EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
  50             d->WriteMessage(NULL, 0, NULL, 0, MOJO_WRITE_MESSAGE_FLAG_NONE));
  51   EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
  52             d->ReadMessage(NULL, NULL, NULL, NULL,
  53                            MOJO_WRITE_MESSAGE_FLAG_NONE));
  54   EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
  55             d->AddWaiter(&w, MOJO_WAIT_FLAG_EVERYTHING, 0));
  56   d->RemoveWaiter(&w);
  57 }
  58
  59 class ThreadSafetyStressThread : public base::SimpleThread {
  60  public:
  61   enum DispatcherOp {
  62     CLOSE = 0,
  63     WRITE_MESSAGE,
  64     READ_MESSAGE,
  65     ADD_WAITER,
  66     REMOVE_WAITER,
  67
  68     DISPATCHER_OP_COUNT
  69   };
  70
  71   ThreadSafetyStressThread(base::WaitableEvent* event,
  72                            scoped_refptr<Dispatcher> dispatcher,
  73                            DispatcherOp op)
  74       : base::SimpleThread("thread_safety_stress_thread"),
  75         event_(event),
  76         dispatcher_(dispatcher),
  77         op_(op) {
  78     CHECK_LE(0, op_);
  79     CHECK_LT(op_, DISPATCHER_OP_COUNT);
  80   }
  81
  82   virtual ~ThreadSafetyStressThread() {
  83     Join();
  84   }
  85
  86  private:
  87   virtual void Run() OVERRIDE {
  88     event_->Wait();
  89
  90     waiter_.Init();
  91     switch(op_) {
  92       case CLOSE: {
  93         MojoResult r = dispatcher_->Close();
  94         EXPECT_TRUE(r == MOJO_RESULT_OK || r == MOJO_RESULT_INVALID_ARGUMENT)
  95             << "Result: " << r;
  96         break;
  97       }
  98       case WRITE_MESSAGE:
  99         EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
 100                   dispatcher_->WriteMessage(NULL, 0, NULL, 0,
 101                                             MOJO_WRITE_MESSAGE_FLAG_NONE));
 102         break;
 103       case READ_MESSAGE:
 104         EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
 105                   dispatcher_->ReadMessage(NULL, NULL, NULL, NULL,
 106                                            MOJO_WRITE_MESSAGE_FLAG_NONE));
 107         break;
 108       case ADD_WAITER: {
 109         MojoResult r = dispatcher_->AddWaiter(&waiter_,
 110                                               MOJO_WAIT_FLAG_EVERYTHING, 0);
 111         EXPECT_TRUE(r == MOJO_RESULT_FAILED_PRECONDITION ||
 112                     r == MOJO_RESULT_INVALID_ARGUMENT);
 113         break;
 114       }
 115       case REMOVE_WAITER:
 116         dispatcher_->RemoveWaiter(&waiter_);
 117         break;
 118       default:
 119         NOTREACHED();
 120         break;
 121     }
 122
 123     // Always try to remove the waiter, in case we added it.
 124     dispatcher_->RemoveWaiter(&waiter_);
 125   }
 126
 127   base::WaitableEvent* const event_;
 128   const scoped_refptr<Dispatcher> dispatcher_;
 129   const DispatcherOp op_;
 130
 131   Waiter waiter_;
 132
 133   DISALLOW_COPY_AND_ASSIGN(ThreadSafetyStressThread);
 134 };
 135
 136 TEST(DispatcherTest, ThreadSafetyStress) {
 137   static const size_t kRepeatCount = 20;
 138   static const size_t kNumThreads = 100;
 139
 140   for (size_t i = 0; i < kRepeatCount; i++) {
 141     // Manual reset, not initially signalled.
 142     base::WaitableEvent event(true, false);
 143     scoped_refptr<Dispatcher> d(new TrivialDispatcher());
 144
 145     {
 146       ScopedVector<ThreadSafetyStressThread> threads;
 147       for (size_t j = 0; j < kNumThreads; j++) {
 148         ThreadSafetyStressThread::DispatcherOp op =
 149             static_cast<ThreadSafetyStressThread::DispatcherOp>(
 150                 (i+j) % ThreadSafetyStressThread::DISPATCHER_OP_COUNT);
 151         threads.push_back(new ThreadSafetyStressThread(&event, d, op));
 152         threads.back()->Start();
 153       }
 154       event.Signal();  // Kicks off real work on the threads.
 155     }  // Joins all the threads.
 156
 157     // One of the threads should already have closed the dispatcher.
 158     EXPECT_EQ(MOJO_RESULT_INVALID_ARGUMENT, d->Close());
 159   }
 160 }
 161
 162 TEST(DispatcherTest, ThreadSafetyStressNoClose) {
 163   static const size_t kRepeatCount = 20;
 164   static const size_t kNumThreads = 100;
 165
 166   for (size_t i = 0; i < kRepeatCount; i++) {
 167     // Manual reset, not initially signalled.
 168     base::WaitableEvent event(true, false);
 169     scoped_refptr<Dispatcher> d(new TrivialDispatcher());
 170
 171     {
 172       ScopedVector<ThreadSafetyStressThread> threads;
 173       for (size_t j = 0; j < kNumThreads; j++) {
 174         ThreadSafetyStressThread::DispatcherOp op =
 175             static_cast<ThreadSafetyStressThread::DispatcherOp>(
 176                 (i+j) % (ThreadSafetyStressThread::DISPATCHER_OP_COUNT-1) + 1);
 177         threads.push_back(new ThreadSafetyStressThread(&event, d, op));
 178         threads.back()->Start();
 179       }
 180       event.Signal();  // Kicks off real work on the threads.
 181     }  // Joins all the threads.
 182
 183     EXPECT_EQ(MOJO_RESULT_OK, d->Close());
 184   }
 185 }
 186
 187 }  // namespace
 188 }  // namespace system
 189 }  // namespace mojo