src/third_party/libjingle/source/talk/base/thread_unittest.cc

   1 /*
   2  * libjingle
   3  * Copyright 2004--2011, Google Inc.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions are met:
   7  *
   8  *  1. Redistributions of source code must retain the above copyright notice,
   9  *     this list of conditions and the following disclaimer.
  10  *  2. Redistributions in binary form must reproduce the above copyright notice,
  11  *     this list of conditions and the following disclaimer in the documentation
  12  *     and/or other materials provided with the distribution.
  13  *  3. The name of the author may not be used to endorse or promote products
  14  *     derived from this software without specific prior written permission.
  15  *
  16  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED
  17  * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
  18  * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
  19  * EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  20  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
  21  * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
  22  * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
  23  * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
  24  * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
  25  * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  26  */
  27
  28 #include "talk/base/asyncinvoker.h"
  29 #include "talk/base/asyncudpsocket.h"
  30 #include "talk/base/event.h"
  31 #include "talk/base/gunit.h"
  32 #include "talk/base/physicalsocketserver.h"
  33 #include "talk/base/socketaddress.h"
  34 #include "talk/base/thread.h"
  35
  36 #ifdef WIN32
  37 #include <comdef.h>  // NOLINT
  38 #endif
  39
  40 using namespace talk_base;
  41
  42 const int MAX = 65536;
  43
  44 // Generates a sequence of numbers (collaboratively).
  45 class TestGenerator {
  46  public:
  47   TestGenerator() : last(0), count(0) {}
  48
  49   int Next(int prev) {
  50     int result = prev + last;
  51     last = result;
  52     count += 1;
  53     return result;
  54   }
  55
  56   int last;
  57   int count;
  58 };
  59
  60 struct TestMessage : public MessageData {
  61   explicit TestMessage(int v) : value(v) {}
  62   virtual ~TestMessage() {}
  63
  64   int value;
  65 };
  66
  67 // Receives on a socket and sends by posting messages.
  68 class SocketClient : public TestGenerator, public sigslot::has_slots<> {
  69  public:
  70   SocketClient(AsyncSocket* socket, const SocketAddress& addr,
  71                Thread* post_thread, MessageHandler* phandler)
  72       : socket_(AsyncUDPSocket::Create(socket, addr)),
  73         post_thread_(post_thread),
  74         post_handler_(phandler) {
  75     socket_->SignalReadPacket.connect(this, &SocketClient::OnPacket);
  76   }
  77
  78   ~SocketClient() {
  79     delete socket_;
  80   }
  81
  82   SocketAddress address() const { return socket_->GetLocalAddress(); }
  83
  84   void OnPacket(AsyncPacketSocket* socket, const char* buf, size_t size,
  85                 const SocketAddress& remote_addr,
  86                 const PacketTime& packet_time) {
  87     EXPECT_EQ(size, sizeof(uint32));
  88     uint32 prev = reinterpret_cast<const uint32*>(buf)[0];
  89     uint32 result = Next(prev);
  90
  91     //socket_->set_readable(last < MAX);
  92     post_thread_->PostDelayed(200, post_handler_, 0, new TestMessage(result));
  93   }
  94
  95  private:
  96   AsyncUDPSocket* socket_;
  97   Thread* post_thread_;
  98   MessageHandler* post_handler_;
  99 };
 100
 101 // Receives messages and sends on a socket.
 102 class MessageClient : public MessageHandler, public TestGenerator {
 103  public:
 104   MessageClient(Thread* pth, Socket* socket)
 105       : thread_(pth), socket_(socket) {
 106   }
 107
 108   virtual ~MessageClient() {
 109     delete socket_;
 110   }
 111
 112   virtual void OnMessage(Message *pmsg) {
 113     TestMessage* msg = static_cast<TestMessage*>(pmsg->pdata);
 114     int result = Next(msg->value);
 115     EXPECT_GE(socket_->Send(&result, sizeof(result)), 0);
 116     delete msg;
 117   }
 118
 119  private:
 120   Thread* thread_;
 121   Socket* socket_;
 122 };
 123
 124 class CustomThread : public talk_base::Thread {
 125  public:
 126   CustomThread() {}
 127   virtual ~CustomThread() { Stop(); }
 128   bool Start() { return false; }
 129 };
 130
 131
 132 // A thread that does nothing when it runs and signals an event
 133 // when it is destroyed.
 134 class SignalWhenDestroyedThread : public Thread {
 135  public:
 136   SignalWhenDestroyedThread(Event* event)
 137       : event_(event) {
 138   }
 139
 140   virtual ~SignalWhenDestroyedThread() {
 141     Stop();
 142     event_->Set();
 143   }
 144
 145   virtual void Run() {
 146     // Do nothing.
 147   }
 148
 149  private:
 150   Event* event_;
 151 };
 152
 153 // Function objects to test Thread::Invoke.
 154 struct FunctorA {
 155   int operator()() { return 42; }
 156 };
 157 class FunctorB {
 158  public:
 159   explicit FunctorB(bool* flag) : flag_(flag) {}
 160   void operator()() { if (flag_) *flag_ = true; }
 161  private:
 162   bool* flag_;
 163 };
 164 struct FunctorC {
 165   int operator()() {
 166     Thread::Current()->ProcessMessages(50);
 167     return 24;
 168   }
 169 };
 170
 171 // See: https://code.google.com/p/webrtc/issues/detail?id=2409
 172 TEST(ThreadTest, DISABLED_Main) {
 173   const SocketAddress addr("127.0.0.1", 0);
 174
 175   // Create the messaging client on its own thread.
 176   Thread th1;
 177   Socket* socket = th1.socketserver()->CreateAsyncSocket(addr.family(),
 178                                                          SOCK_DGRAM);
 179   MessageClient msg_client(&th1, socket);
 180
 181   // Create the socket client on its own thread.
 182   Thread th2;
 183   AsyncSocket* asocket =
 184       th2.socketserver()->CreateAsyncSocket(addr.family(), SOCK_DGRAM);
 185   SocketClient sock_client(asocket, addr, &th1, &msg_client);
 186
 187   socket->Connect(sock_client.address());
 188
 189   th1.Start();
 190   th2.Start();
 191
 192   // Get the messages started.
 193   th1.PostDelayed(100, &msg_client, 0, new TestMessage(1));
 194
 195   // Give the clients a little while to run.
 196   // Messages will be processed at 100, 300, 500, 700, 900.
 197   Thread* th_main = Thread::Current();
 198   th_main->ProcessMessages(1000);
 199
 200   // Stop the sending client. Give the receiver a bit longer to run, in case
 201   // it is running on a machine that is under load (e.g. the build machine).
 202   th1.Stop();
 203   th_main->ProcessMessages(200);
 204   th2.Stop();
 205
 206   // Make sure the results were correct
 207   EXPECT_EQ(5, msg_client.count);
 208   EXPECT_EQ(34, msg_client.last);
 209   EXPECT_EQ(5, sock_client.count);
 210   EXPECT_EQ(55, sock_client.last);
 211 }
 212
 213 // Test that setting thread names doesn't cause a malfunction.
 214 // There's no easy way to verify the name was set properly at this time.
 215 TEST(ThreadTest, Names) {
 216   // Default name
 217   Thread *thread;
 218   thread = new Thread();
 219   EXPECT_TRUE(thread->Start());
 220   thread->Stop();
 221   delete thread;
 222   thread = new Thread();
 223   // Name with no object parameter
 224   EXPECT_TRUE(thread->SetName("No object", NULL));
 225   EXPECT_TRUE(thread->Start());
 226   thread->Stop();
 227   delete thread;
 228   // Really long name
 229   thread = new Thread();
 230   EXPECT_TRUE(thread->SetName("Abcdefghijklmnopqrstuvwxyz1234567890", this));
 231   EXPECT_TRUE(thread->Start());
 232   thread->Stop();
 233   delete thread;
 234 }
 235
 236 // Test that setting thread priorities doesn't cause a malfunction.
 237 // There's no easy way to verify the priority was set properly at this time.
 238 TEST(ThreadTest, Priorities) {
 239   Thread *thread;
 240   thread = new Thread();
 241   EXPECT_TRUE(thread->SetPriority(PRIORITY_HIGH));
 242   EXPECT_TRUE(thread->Start());
 243   thread->Stop();
 244   delete thread;
 245   thread = new Thread();
 246   EXPECT_TRUE(thread->SetPriority(PRIORITY_ABOVE_NORMAL));
 247   EXPECT_TRUE(thread->Start());
 248   thread->Stop();
 249   delete thread;
 250
 251   thread = new Thread();
 252   EXPECT_TRUE(thread->Start());
 253 #ifdef WIN32
 254   EXPECT_TRUE(thread->SetPriority(PRIORITY_ABOVE_NORMAL));
 255 #else
 256   EXPECT_FALSE(thread->SetPriority(PRIORITY_ABOVE_NORMAL));
 257 #endif
 258   thread->Stop();
 259   delete thread;
 260
 261 }
 262
 263 TEST(ThreadTest, Wrap) {
 264   Thread* current_thread = Thread::Current();
 265   current_thread->UnwrapCurrent();
 266   CustomThread* cthread = new CustomThread();
 267   EXPECT_TRUE(cthread->WrapCurrent());
 268   EXPECT_TRUE(cthread->started());
 269   EXPECT_FALSE(cthread->IsOwned());
 270   cthread->UnwrapCurrent();
 271   EXPECT_FALSE(cthread->started());
 272   delete cthread;
 273   current_thread->WrapCurrent();
 274 }
 275
 276 // Test that calling Release on a thread causes it to self-destruct when
 277 // it's finished running
 278 TEST(ThreadTest, Release) {
 279   scoped_ptr<Event> event(new Event(true, false));
 280   // Ensure the event is initialized.
 281   event->Reset();
 282
 283   Thread* thread = new SignalWhenDestroyedThread(event.get());
 284   thread->Start();
 285   thread->Release();
 286
 287   // The event should get signaled when the thread completes, which should
 288   // be nearly instantaneous, since it doesn't do anything.  For safety,
 289   // give it 3 seconds in case the machine is under load.
 290   bool signaled = event->Wait(3000);
 291   EXPECT_TRUE(signaled);
 292 }
 293
 294 TEST(ThreadTest, Invoke) {
 295   // Create and start the thread.
 296   Thread thread;
 297   thread.Start();
 298   // Try calling functors.
 299   EXPECT_EQ(42, thread.Invoke<int>(FunctorA()));
 300   bool called = false;
 301   FunctorB f2(&called);
 302   thread.Invoke<void>(f2);
 303   EXPECT_TRUE(called);
 304   // Try calling bare functions.
 305   struct LocalFuncs {
 306     static int Func1() { return 999; }
 307     static void Func2() {}
 308   };
 309   EXPECT_EQ(999, thread.Invoke<int>(&LocalFuncs::Func1));
 310   thread.Invoke<void>(&LocalFuncs::Func2);
 311 }
 312
 313 class AsyncInvokeTest : public testing::Test {
 314  public:
 315   void IntCallback(int value) {
 316     EXPECT_EQ(expected_thread_, Thread::Current());
 317     int_value_ = value;
 318   }
 319   void AsyncInvokeIntCallback(AsyncInvoker* invoker, Thread* thread) {
 320     expected_thread_ = thread;
 321     invoker->AsyncInvoke(thread, FunctorC(),
 322                          &AsyncInvokeTest::IntCallback,
 323                          static_cast<AsyncInvokeTest*>(this));
 324     invoke_started_.Set();
 325   }
 326   void SetExpectedThreadForIntCallback(Thread* thread) {
 327     expected_thread_ = thread;
 328   }
 329
 330  protected:
 331   enum { kWaitTimeout = 1000 };
 332   AsyncInvokeTest()
 333       : int_value_(0),
 334         invoke_started_(true, false),
 335         expected_thread_(NULL) {}
 336
 337   int int_value_;
 338   Event invoke_started_;
 339   Thread* expected_thread_;
 340 };
 341
 342 TEST_F(AsyncInvokeTest, FireAndForget) {
 343   AsyncInvoker invoker;
 344   // Create and start the thread.
 345   Thread thread;
 346   thread.Start();
 347   // Try calling functor.
 348   bool called = false;
 349   invoker.AsyncInvoke<void>(&thread, FunctorB(&called));
 350   EXPECT_TRUE_WAIT(called, kWaitTimeout);
 351 }
 352
 353 TEST_F(AsyncInvokeTest, WithCallback) {
 354   AsyncInvoker invoker;
 355   // Create and start the thread.
 356   Thread thread;
 357   thread.Start();
 358   // Try calling functor.
 359   SetExpectedThreadForIntCallback(Thread::Current());
 360   invoker.AsyncInvoke(&thread, FunctorA(),
 361                       &AsyncInvokeTest::IntCallback,
 362                       static_cast<AsyncInvokeTest*>(this));
 363   EXPECT_EQ_WAIT(42, int_value_, kWaitTimeout);
 364 }
 365
 366 TEST_F(AsyncInvokeTest, CancelInvoker) {
 367   // Create and start the thread.
 368   Thread thread;
 369   thread.Start();
 370   // Try destroying invoker during call.
 371   {
 372     AsyncInvoker invoker;
 373     invoker.AsyncInvoke(&thread, FunctorC(),
 374                         &AsyncInvokeTest::IntCallback,
 375                         static_cast<AsyncInvokeTest*>(this));
 376   }
 377   // With invoker gone, callback should be cancelled.
 378   Thread::Current()->ProcessMessages(kWaitTimeout);
 379   EXPECT_EQ(0, int_value_);
 380 }
 381
 382 TEST_F(AsyncInvokeTest, CancelCallingThread) {
 383   AsyncInvoker invoker;
 384   { // Create and start the thread.
 385     Thread thread;
 386     thread.Start();
 387     // Try calling functor.
 388     thread.Invoke<void>(Bind(&AsyncInvokeTest::AsyncInvokeIntCallback,
 389                              static_cast<AsyncInvokeTest*>(this),
 390                              &invoker, Thread::Current()));
 391     // Wait for the call to begin.
 392     ASSERT_TRUE(invoke_started_.Wait(kWaitTimeout));
 393   }
 394   // Calling thread is gone. Return message shouldn't happen.
 395   Thread::Current()->ProcessMessages(kWaitTimeout);
 396   EXPECT_EQ(0, int_value_);
 397 }
 398
 399 TEST_F(AsyncInvokeTest, KillInvokerBeforeExecute) {
 400   Thread thread;
 401   thread.Start();
 402   {
 403     AsyncInvoker invoker;
 404     // Try calling functor.
 405     thread.Invoke<void>(Bind(&AsyncInvokeTest::AsyncInvokeIntCallback,
 406                              static_cast<AsyncInvokeTest*>(this),
 407                              &invoker, Thread::Current()));
 408     // Wait for the call to begin.
 409     ASSERT_TRUE(invoke_started_.Wait(kWaitTimeout));
 410   }
 411   // Invoker is destroyed. Function should not execute.
 412   Thread::Current()->ProcessMessages(kWaitTimeout);
 413   EXPECT_EQ(0, int_value_);
 414 }
 415
 416 TEST_F(AsyncInvokeTest, Flush) {
 417   AsyncInvoker invoker;
 418   bool flag1 = false;
 419   bool flag2 = false;
 420   // Queue two async calls to the current thread.
 421   invoker.AsyncInvoke<void>(Thread::Current(),
 422                             FunctorB(&flag1));
 423   invoker.AsyncInvoke<void>(Thread::Current(),
 424                             FunctorB(&flag2));
 425   // Because we haven't pumped messages, these should not have run yet.
 426   EXPECT_FALSE(flag1);
 427   EXPECT_FALSE(flag2);
 428   // Force them to run now.
 429   invoker.Flush(Thread::Current());
 430   EXPECT_TRUE(flag1);
 431   EXPECT_TRUE(flag2);
 432 }
 433
 434 TEST_F(AsyncInvokeTest, FlushWithIds) {
 435   AsyncInvoker invoker;
 436   bool flag1 = false;
 437   bool flag2 = false;
 438   // Queue two async calls to the current thread, one with a message id.
 439   invoker.AsyncInvoke<void>(Thread::Current(),
 440                             FunctorB(&flag1),
 441                             5);
 442   invoker.AsyncInvoke<void>(Thread::Current(),
 443                             FunctorB(&flag2));
 444   // Because we haven't pumped messages, these should not have run yet.
 445   EXPECT_FALSE(flag1);
 446   EXPECT_FALSE(flag2);
 447   // Execute pending calls with id == 5.
 448   invoker.Flush(Thread::Current(), 5);
 449   EXPECT_TRUE(flag1);
 450   EXPECT_FALSE(flag2);
 451   flag1 = false;
 452   // Execute all pending calls. The id == 5 call should not execute again.
 453   invoker.Flush(Thread::Current());
 454   EXPECT_FALSE(flag1);
 455   EXPECT_TRUE(flag2);
 456 }
 457
 458
 459 #ifdef WIN32
 460 class ComThreadTest : public testing::Test, public MessageHandler {
 461  public:
 462   ComThreadTest() : done_(false) {}
 463  protected:
 464   virtual void OnMessage(Message* message) {
 465     HRESULT hr = CoInitializeEx(NULL, COINIT_MULTITHREADED);
 466     // S_FALSE means the thread was already inited for a multithread apartment.
 467     EXPECT_EQ(S_FALSE, hr);
 468     if (SUCCEEDED(hr)) {
 469       CoUninitialize();
 470     }
 471     done_ = true;
 472   }
 473   bool done_;
 474 };
 475
 476 TEST_F(ComThreadTest, ComInited) {
 477   Thread* thread = new ComThread();
 478   EXPECT_TRUE(thread->Start());
 479   thread->Post(this, 0);
 480   EXPECT_TRUE_WAIT(done_, 1000);
 481   delete thread;
 482 }
 483 #endif