src/third_party/webrtc/base/thread_unittest.cc

   1 /*
   2  *  Copyright 2004 The WebRTC Project Authors. All rights reserved.
   3  *
   4  *  Use of this source code is governed by a BSD-style license
   5  *  that can be found in the LICENSE file in the root of the source
   6  *  tree. An additional intellectual property rights grant can be found
   7  *  in the file PATENTS.  All contributing project authors may
   8  *  be found in the AUTHORS file in the root of the source tree.
   9  */
  10
  11 #include "webrtc/base/asyncinvoker.h"
  12 #include "webrtc/base/asyncudpsocket.h"
  13 #include "webrtc/base/event.h"
  14 #include "webrtc/base/gunit.h"
  15 #include "webrtc/base/physicalsocketserver.h"
  16 #include "webrtc/base/socketaddress.h"
  17 #include "webrtc/base/thread.h"
  18 #include "webrtc/test/testsupport/gtest_disable.h"
  19
  20 #if defined(WEBRTC_WIN)
  21 #include <comdef.h>  // NOLINT
  22 #endif
  23
  24 using namespace rtc;
  25
  26 // Generates a sequence of numbers (collaboratively).
  27 class TestGenerator {
  28  public:
  29   TestGenerator() : last(0), count(0) {}
  30
  31   int Next(int prev) {
  32     int result = prev + last;
  33     last = result;
  34     count += 1;
  35     return result;
  36   }
  37
  38   int last;
  39   int count;
  40 };
  41
  42 struct TestMessage : public MessageData {
  43   explicit TestMessage(int v) : value(v) {}
  44   virtual ~TestMessage() {}
  45
  46   int value;
  47 };
  48
  49 // Receives on a socket and sends by posting messages.
  50 class SocketClient : public TestGenerator, public sigslot::has_slots<> {
  51  public:
  52   SocketClient(AsyncSocket* socket, const SocketAddress& addr,
  53                Thread* post_thread, MessageHandler* phandler)
  54       : socket_(AsyncUDPSocket::Create(socket, addr)),
  55         post_thread_(post_thread),
  56         post_handler_(phandler) {
  57     socket_->SignalReadPacket.connect(this, &SocketClient::OnPacket);
  58   }
  59
  60   ~SocketClient() {
  61     delete socket_;
  62   }
  63
  64   SocketAddress address() const { return socket_->GetLocalAddress(); }
  65
  66   void OnPacket(AsyncPacketSocket* socket, const char* buf, size_t size,
  67                 const SocketAddress& remote_addr,
  68                 const PacketTime& packet_time) {
  69     EXPECT_EQ(size, sizeof(uint32));
  70     uint32 prev = reinterpret_cast<const uint32*>(buf)[0];
  71     uint32 result = Next(prev);
  72
  73     post_thread_->PostDelayed(200, post_handler_, 0, new TestMessage(result));
  74   }
  75
  76  private:
  77   AsyncUDPSocket* socket_;
  78   Thread* post_thread_;
  79   MessageHandler* post_handler_;
  80 };
  81
  82 // Receives messages and sends on a socket.
  83 class MessageClient : public MessageHandler, public TestGenerator {
  84  public:
  85   MessageClient(Thread* pth, Socket* socket)
  86       : socket_(socket) {
  87   }
  88
  89   virtual ~MessageClient() {
  90     delete socket_;
  91   }
  92
  93   virtual void OnMessage(Message *pmsg) {
  94     TestMessage* msg = static_cast<TestMessage*>(pmsg->pdata);
  95     int result = Next(msg->value);
  96     EXPECT_GE(socket_->Send(&result, sizeof(result)), 0);
  97     delete msg;
  98   }
  99
 100  private:
 101   Socket* socket_;
 102 };
 103
 104 class CustomThread : public rtc::Thread {
 105  public:
 106   CustomThread() {}
 107   virtual ~CustomThread() { Stop(); }
 108   bool Start() { return false; }
 109
 110   bool WrapCurrent() {
 111     return Thread::WrapCurrent();
 112   }
 113   void UnwrapCurrent() {
 114     Thread::UnwrapCurrent();
 115   }
 116 };
 117
 118
 119 // A thread that does nothing when it runs and signals an event
 120 // when it is destroyed.
 121 class SignalWhenDestroyedThread : public Thread {
 122  public:
 123   SignalWhenDestroyedThread(Event* event)
 124       : event_(event) {
 125   }
 126
 127   virtual ~SignalWhenDestroyedThread() {
 128     Stop();
 129     event_->Set();
 130   }
 131
 132   virtual void Run() {
 133     // Do nothing.
 134   }
 135
 136  private:
 137   Event* event_;
 138 };
 139
 140 // Function objects to test Thread::Invoke.
 141 struct FunctorA {
 142   int operator()() { return 42; }
 143 };
 144 class FunctorB {
 145  public:
 146   explicit FunctorB(bool* flag) : flag_(flag) {}
 147   void operator()() { if (flag_) *flag_ = true; }
 148  private:
 149   bool* flag_;
 150 };
 151 struct FunctorC {
 152   int operator()() {
 153     Thread::Current()->ProcessMessages(50);
 154     return 24;
 155   }
 156 };
 157
 158 // See: https://code.google.com/p/webrtc/issues/detail?id=2409
 159 TEST(ThreadTest, DISABLED_Main) {
 160   const SocketAddress addr("127.0.0.1", 0);
 161
 162   // Create the messaging client on its own thread.
 163   Thread th1;
 164   Socket* socket = th1.socketserver()->CreateAsyncSocket(addr.family(),
 165                                                          SOCK_DGRAM);
 166   MessageClient msg_client(&th1, socket);
 167
 168   // Create the socket client on its own thread.
 169   Thread th2;
 170   AsyncSocket* asocket =
 171       th2.socketserver()->CreateAsyncSocket(addr.family(), SOCK_DGRAM);
 172   SocketClient sock_client(asocket, addr, &th1, &msg_client);
 173
 174   socket->Connect(sock_client.address());
 175
 176   th1.Start();
 177   th2.Start();
 178
 179   // Get the messages started.
 180   th1.PostDelayed(100, &msg_client, 0, new TestMessage(1));
 181
 182   // Give the clients a little while to run.
 183   // Messages will be processed at 100, 300, 500, 700, 900.
 184   Thread* th_main = Thread::Current();
 185   th_main->ProcessMessages(1000);
 186
 187   // Stop the sending client. Give the receiver a bit longer to run, in case
 188   // it is running on a machine that is under load (e.g. the build machine).
 189   th1.Stop();
 190   th_main->ProcessMessages(200);
 191   th2.Stop();
 192
 193   // Make sure the results were correct
 194   EXPECT_EQ(5, msg_client.count);
 195   EXPECT_EQ(34, msg_client.last);
 196   EXPECT_EQ(5, sock_client.count);
 197   EXPECT_EQ(55, sock_client.last);
 198 }
 199
 200 // Test that setting thread names doesn't cause a malfunction.
 201 // There's no easy way to verify the name was set properly at this time.
 202 TEST(ThreadTest, Names) {
 203   // Default name
 204   Thread *thread;
 205   thread = new Thread();
 206   EXPECT_TRUE(thread->Start());
 207   thread->Stop();
 208   delete thread;
 209   thread = new Thread();
 210   // Name with no object parameter
 211   EXPECT_TRUE(thread->SetName("No object", NULL));
 212   EXPECT_TRUE(thread->Start());
 213   thread->Stop();
 214   delete thread;
 215   // Really long name
 216   thread = new Thread();
 217   EXPECT_TRUE(thread->SetName("Abcdefghijklmnopqrstuvwxyz1234567890", this));
 218   EXPECT_TRUE(thread->Start());
 219   thread->Stop();
 220   delete thread;
 221 }
 222
 223 // Test that setting thread priorities doesn't cause a malfunction.
 224 // There's no easy way to verify the priority was set properly at this time.
 225 TEST(ThreadTest, Priorities) {
 226   Thread *thread;
 227   thread = new Thread();
 228   EXPECT_TRUE(thread->SetPriority(PRIORITY_HIGH));
 229   EXPECT_TRUE(thread->Start());
 230   thread->Stop();
 231   delete thread;
 232   thread = new Thread();
 233   EXPECT_TRUE(thread->SetPriority(PRIORITY_ABOVE_NORMAL));
 234   EXPECT_TRUE(thread->Start());
 235   thread->Stop();
 236   delete thread;
 237
 238   thread = new Thread();
 239   EXPECT_TRUE(thread->Start());
 240 #if defined(WEBRTC_WIN)
 241   EXPECT_TRUE(thread->SetPriority(PRIORITY_ABOVE_NORMAL));
 242 #else
 243   EXPECT_FALSE(thread->SetPriority(PRIORITY_ABOVE_NORMAL));
 244 #endif
 245   thread->Stop();
 246   delete thread;
 247
 248 }
 249
 250 TEST(ThreadTest, Wrap) {
 251   Thread* current_thread = Thread::Current();
 252   current_thread->UnwrapCurrent();
 253   CustomThread* cthread = new CustomThread();
 254   EXPECT_TRUE(cthread->WrapCurrent());
 255   EXPECT_TRUE(cthread->RunningForTest());
 256   EXPECT_FALSE(cthread->IsOwned());
 257   cthread->UnwrapCurrent();
 258   EXPECT_FALSE(cthread->RunningForTest());
 259   delete cthread;
 260   current_thread->WrapCurrent();
 261 }
 262
 263 TEST(ThreadTest, Invoke) {
 264   // Create and start the thread.
 265   Thread thread;
 266   thread.Start();
 267   // Try calling functors.
 268   EXPECT_EQ(42, thread.Invoke<int>(FunctorA()));
 269   bool called = false;
 270   FunctorB f2(&called);
 271   thread.Invoke<void>(f2);
 272   EXPECT_TRUE(called);
 273   // Try calling bare functions.
 274   struct LocalFuncs {
 275     static int Func1() { return 999; }
 276     static void Func2() {}
 277   };
 278   EXPECT_EQ(999, thread.Invoke<int>(&LocalFuncs::Func1));
 279   thread.Invoke<void>(&LocalFuncs::Func2);
 280 }
 281
 282 // Verifies that two threads calling Invoke on each other at the same time does
 283 // not deadlock.
 284 TEST(ThreadTest, TwoThreadsInvokeNoDeadlock) {
 285   AutoThread thread;
 286   Thread* current_thread = Thread::Current();
 287   ASSERT_TRUE(current_thread != NULL);
 288
 289   Thread other_thread;
 290   other_thread.Start();
 291
 292   struct LocalFuncs {
 293     static void Set(bool* out) { *out = true; }
 294     static void InvokeSet(Thread* thread, bool* out) {
 295       thread->Invoke<void>(Bind(&Set, out));
 296     }
 297   };
 298
 299   bool called = false;
 300   other_thread.Invoke<void>(
 301       Bind(&LocalFuncs::InvokeSet, current_thread, &called));
 302
 303   EXPECT_TRUE(called);
 304 }
 305
 306 // Verifies that if thread A invokes a call on thread B and thread C is trying
 307 // to invoke A at the same time, thread A does not handle C's invoke while
 308 // invoking B.
 309 TEST(ThreadTest, ThreeThreadsInvoke) {
 310   AutoThread thread;
 311   Thread* thread_a = Thread::Current();
 312   Thread thread_b, thread_c;
 313   thread_b.Start();
 314   thread_c.Start();
 315
 316   class LockedBool {
 317    public:
 318     explicit LockedBool(bool value) : value_(value) {}
 319
 320     void Set(bool value) {
 321       CritScope lock(&crit_);
 322       value_ = value;
 323     }
 324
 325     bool Get() {
 326       CritScope lock(&crit_);
 327       return value_;
 328     }
 329
 330    private:
 331     CriticalSection crit_;
 332     bool value_ GUARDED_BY(crit_);
 333   };
 334
 335   struct LocalFuncs {
 336     static void Set(LockedBool* out) { out->Set(true); }
 337     static void InvokeSet(Thread* thread, LockedBool* out) {
 338       thread->Invoke<void>(Bind(&Set, out));
 339     }
 340
 341     // Set |out| true and call InvokeSet on |thread|.
 342     static void SetAndInvokeSet(LockedBool* out,
 343                                 Thread* thread,
 344                                 LockedBool* out_inner) {
 345       out->Set(true);
 346       InvokeSet(thread, out_inner);
 347     }
 348
 349     // Asynchronously invoke SetAndInvokeSet on |thread1| and wait until
 350     // |thread1| starts the call.
 351     static void AsyncInvokeSetAndWait(
 352         Thread* thread1, Thread* thread2, LockedBool* out) {
 353       CriticalSection crit;
 354       LockedBool async_invoked(false);
 355
 356       AsyncInvoker invoker;
 357       invoker.AsyncInvoke<void>(
 358           thread1, Bind(&SetAndInvokeSet, &async_invoked, thread2, out));
 359
 360       EXPECT_TRUE_WAIT(async_invoked.Get(), 2000);
 361     }
 362   };
 363
 364   LockedBool thread_a_called(false);
 365
 366   // Start the sequence A --(invoke)--> B --(async invoke)--> C --(invoke)--> A.
 367   // Thread B returns when C receives the call and C should be blocked until A
 368   // starts to process messages.
 369   thread_b.Invoke<void>(Bind(&LocalFuncs::AsyncInvokeSetAndWait,
 370                              &thread_c, thread_a, &thread_a_called));
 371   EXPECT_FALSE(thread_a_called.Get());
 372
 373   EXPECT_TRUE_WAIT(thread_a_called.Get(), 2000);
 374 }
 375
 376 class AsyncInvokeTest : public testing::Test {
 377  public:
 378   void IntCallback(int value) {
 379     EXPECT_EQ(expected_thread_, Thread::Current());
 380     int_value_ = value;
 381   }
 382   void AsyncInvokeIntCallback(AsyncInvoker* invoker, Thread* thread) {
 383     expected_thread_ = thread;
 384     invoker->AsyncInvoke(thread, FunctorC(),
 385                          &AsyncInvokeTest::IntCallback,
 386                          static_cast<AsyncInvokeTest*>(this));
 387     invoke_started_.Set();
 388   }
 389   void SetExpectedThreadForIntCallback(Thread* thread) {
 390     expected_thread_ = thread;
 391   }
 392
 393  protected:
 394   enum { kWaitTimeout = 1000 };
 395   AsyncInvokeTest()
 396       : int_value_(0),
 397         invoke_started_(true, false),
 398         expected_thread_(NULL) {}
 399
 400   int int_value_;
 401   Event invoke_started_;
 402   Thread* expected_thread_;
 403 };
 404
 405 TEST_F(AsyncInvokeTest, FireAndForget) {
 406   AsyncInvoker invoker;
 407   // Create and start the thread.
 408   Thread thread;
 409   thread.Start();
 410   // Try calling functor.
 411   bool called = false;
 412   invoker.AsyncInvoke<void>(&thread, FunctorB(&called));
 413   EXPECT_TRUE_WAIT(called, kWaitTimeout);
 414 }
 415
 416 TEST_F(AsyncInvokeTest, WithCallback) {
 417   AsyncInvoker invoker;
 418   // Create and start the thread.
 419   Thread thread;
 420   thread.Start();
 421   // Try calling functor.
 422   SetExpectedThreadForIntCallback(Thread::Current());
 423   invoker.AsyncInvoke(&thread, FunctorA(),
 424                       &AsyncInvokeTest::IntCallback,
 425                       static_cast<AsyncInvokeTest*>(this));
 426   EXPECT_EQ_WAIT(42, int_value_, kWaitTimeout);
 427 }
 428
 429 TEST_F(AsyncInvokeTest, CancelInvoker) {
 430   // Create and start the thread.
 431   Thread thread;
 432   thread.Start();
 433   // Try destroying invoker during call.
 434   {
 435     AsyncInvoker invoker;
 436     invoker.AsyncInvoke(&thread, FunctorC(),
 437                         &AsyncInvokeTest::IntCallback,
 438                         static_cast<AsyncInvokeTest*>(this));
 439   }
 440   // With invoker gone, callback should be cancelled.
 441   Thread::Current()->ProcessMessages(kWaitTimeout);
 442   EXPECT_EQ(0, int_value_);
 443 }
 444
 445 TEST_F(AsyncInvokeTest, CancelCallingThread) {
 446   AsyncInvoker invoker;
 447   { // Create and start the thread.
 448     Thread thread;
 449     thread.Start();
 450     // Try calling functor.
 451     thread.Invoke<void>(Bind(&AsyncInvokeTest::AsyncInvokeIntCallback,
 452                              static_cast<AsyncInvokeTest*>(this),
 453                              &invoker, Thread::Current()));
 454     // Wait for the call to begin.
 455     ASSERT_TRUE(invoke_started_.Wait(kWaitTimeout));
 456   }
 457   // Calling thread is gone. Return message shouldn't happen.
 458   Thread::Current()->ProcessMessages(kWaitTimeout);
 459   EXPECT_EQ(0, int_value_);
 460 }
 461
 462 TEST_F(AsyncInvokeTest, KillInvokerBeforeExecute) {
 463   Thread thread;
 464   thread.Start();
 465   {
 466     AsyncInvoker invoker;
 467     // Try calling functor.
 468     thread.Invoke<void>(Bind(&AsyncInvokeTest::AsyncInvokeIntCallback,
 469                              static_cast<AsyncInvokeTest*>(this),
 470                              &invoker, Thread::Current()));
 471     // Wait for the call to begin.
 472     ASSERT_TRUE(invoke_started_.Wait(kWaitTimeout));
 473   }
 474   // Invoker is destroyed. Function should not execute.
 475   Thread::Current()->ProcessMessages(kWaitTimeout);
 476   EXPECT_EQ(0, int_value_);
 477 }
 478
 479 TEST_F(AsyncInvokeTest, Flush) {
 480   AsyncInvoker invoker;
 481   bool flag1 = false;
 482   bool flag2 = false;
 483   // Queue two async calls to the current thread.
 484   invoker.AsyncInvoke<void>(Thread::Current(),
 485                             FunctorB(&flag1));
 486   invoker.AsyncInvoke<void>(Thread::Current(),
 487                             FunctorB(&flag2));
 488   // Because we haven't pumped messages, these should not have run yet.
 489   EXPECT_FALSE(flag1);
 490   EXPECT_FALSE(flag2);
 491   // Force them to run now.
 492   invoker.Flush(Thread::Current());
 493   EXPECT_TRUE(flag1);
 494   EXPECT_TRUE(flag2);
 495 }
 496
 497 TEST_F(AsyncInvokeTest, FlushWithIds) {
 498   AsyncInvoker invoker;
 499   bool flag1 = false;
 500   bool flag2 = false;
 501   // Queue two async calls to the current thread, one with a message id.
 502   invoker.AsyncInvoke<void>(Thread::Current(),
 503                             FunctorB(&flag1),
 504                             5);
 505   invoker.AsyncInvoke<void>(Thread::Current(),
 506                             FunctorB(&flag2));
 507   // Because we haven't pumped messages, these should not have run yet.
 508   EXPECT_FALSE(flag1);
 509   EXPECT_FALSE(flag2);
 510   // Execute pending calls with id == 5.
 511   invoker.Flush(Thread::Current(), 5);
 512   EXPECT_TRUE(flag1);
 513   EXPECT_FALSE(flag2);
 514   flag1 = false;
 515   // Execute all pending calls. The id == 5 call should not execute again.
 516   invoker.Flush(Thread::Current());
 517   EXPECT_FALSE(flag1);
 518   EXPECT_TRUE(flag2);
 519 }
 520
 521
 522 #if defined(WEBRTC_WIN)
 523 class ComThreadTest : public testing::Test, public MessageHandler {
 524  public:
 525   ComThreadTest() : done_(false) {}
 526  protected:
 527   virtual void OnMessage(Message* message) {
 528     HRESULT hr = CoInitializeEx(NULL, COINIT_MULTITHREADED);
 529     // S_FALSE means the thread was already inited for a multithread apartment.
 530     EXPECT_EQ(S_FALSE, hr);
 531     if (SUCCEEDED(hr)) {
 532       CoUninitialize();
 533     }
 534     done_ = true;
 535   }
 536   bool done_;
 537 };
 538
 539 TEST_F(ComThreadTest, ComInited) {
 540   Thread* thread = new ComThread();
 541   EXPECT_TRUE(thread->Start());
 542   thread->Post(this, 0);
 543   EXPECT_TRUE_WAIT(done_, 1000);
 544   delete thread;
 545 }
 546 #endif