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/asyncudpsocket.h"
  29 #include "talk/base/event.h"
  30 #include "talk/base/gunit.h"
  31 #include "talk/base/physicalsocketserver.h"
  32 #include "talk/base/socketaddress.h"
  33 #include "talk/base/thread.h"
  34
  35 #ifdef WIN32
  36 #include <comdef.h>  // NOLINT
  37 #endif
  38
  39 using namespace talk_base;
  40
  41 const int MAX = 65536;
  42
  43 // Generates a sequence of numbers (collaboratively).
  44 class TestGenerator {
  45  public:
  46   TestGenerator() : last(0), count(0) {}
  47
  48   int Next(int prev) {
  49     int result = prev + last;
  50     last = result;
  51     count += 1;
  52     return result;
  53   }
  54
  55   int last;
  56   int count;
  57 };
  58
  59 struct TestMessage : public MessageData {
  60   explicit TestMessage(int v) : value(v) {}
  61   virtual ~TestMessage() {}
  62
  63   int value;
  64 };
  65
  66 // Receives on a socket and sends by posting messages.
  67 class SocketClient : public TestGenerator, public sigslot::has_slots<> {
  68  public:
  69   SocketClient(AsyncSocket* socket, const SocketAddress& addr,
  70                Thread* post_thread, MessageHandler* phandler)
  71       : socket_(AsyncUDPSocket::Create(socket, addr)),
  72         post_thread_(post_thread),
  73         post_handler_(phandler) {
  74     socket_->SignalReadPacket.connect(this, &SocketClient::OnPacket);
  75   }
  76
  77   ~SocketClient() {
  78     delete socket_;
  79   }
  80
  81   SocketAddress address() const { return socket_->GetLocalAddress(); }
  82
  83   void OnPacket(AsyncPacketSocket* socket, const char* buf, size_t size,
  84                 const SocketAddress& remote_addr) {
  85     EXPECT_EQ(size, sizeof(uint32));
  86     uint32 prev = reinterpret_cast<const uint32*>(buf)[0];
  87     uint32 result = Next(prev);
  88
  89     //socket_->set_readable(last < MAX);
  90     post_thread_->PostDelayed(200, post_handler_, 0, new TestMessage(result));
  91   }
  92
  93  private:
  94   AsyncUDPSocket* socket_;
  95   Thread* post_thread_;
  96   MessageHandler* post_handler_;
  97 };
  98
  99 // Receives messages and sends on a socket.
 100 class MessageClient : public MessageHandler, public TestGenerator {
 101  public:
 102   MessageClient(Thread* pth, Socket* socket)
 103       : thread_(pth), socket_(socket) {
 104   }
 105
 106   virtual ~MessageClient() {
 107     delete socket_;
 108   }
 109
 110   virtual void OnMessage(Message *pmsg) {
 111     TestMessage* msg = static_cast<TestMessage*>(pmsg->pdata);
 112     int result = Next(msg->value);
 113     EXPECT_GE(socket_->Send(&result, sizeof(result)), 0);
 114     delete msg;
 115   }
 116
 117  private:
 118   Thread* thread_;
 119   Socket* socket_;
 120 };
 121
 122 class CustomThread : public talk_base::Thread {
 123  public:
 124   CustomThread() {}
 125   virtual ~CustomThread() { Stop(); }
 126   bool Start() { return false; }
 127 };
 128
 129
 130 // A thread that does nothing when it runs and signals an event
 131 // when it is destroyed.
 132 class SignalWhenDestroyedThread : public Thread {
 133  public:
 134   SignalWhenDestroyedThread(Event* event)
 135       : event_(event) {
 136   }
 137
 138   virtual ~SignalWhenDestroyedThread() {
 139     Stop();
 140     event_->Set();
 141   }
 142
 143   virtual void Run() {
 144     // Do nothing.
 145   }
 146
 147  private:
 148   Event* event_;
 149 };
 150
 151 // Function objects to test Thread::Invoke.
 152 struct Functor1 {
 153   int operator()() { return 42; }
 154 };
 155 class Functor2 {
 156  public:
 157   explicit Functor2(bool* flag) : flag_(flag) {}
 158   void operator()() { if (flag_) *flag_ = true; }
 159  private:
 160   bool* flag_;
 161 };
 162
 163 // See: https://code.google.com/p/webrtc/issues/detail?id=2409
 164 TEST(ThreadTest, DISABLED_Main) {
 165   const SocketAddress addr("127.0.0.1", 0);
 166
 167   // Create the messaging client on its own thread.
 168   Thread th1;
 169   Socket* socket = th1.socketserver()->CreateAsyncSocket(addr.family(),
 170                                                          SOCK_DGRAM);
 171   MessageClient msg_client(&th1, socket);
 172
 173   // Create the socket client on its own thread.
 174   Thread th2;
 175   AsyncSocket* asocket =
 176       th2.socketserver()->CreateAsyncSocket(addr.family(), SOCK_DGRAM);
 177   SocketClient sock_client(asocket, addr, &th1, &msg_client);
 178
 179   socket->Connect(sock_client.address());
 180
 181   th1.Start();
 182   th2.Start();
 183
 184   // Get the messages started.
 185   th1.PostDelayed(100, &msg_client, 0, new TestMessage(1));
 186
 187   // Give the clients a little while to run.
 188   // Messages will be processed at 100, 300, 500, 700, 900.
 189   Thread* th_main = Thread::Current();
 190   th_main->ProcessMessages(1000);
 191
 192   // Stop the sending client. Give the receiver a bit longer to run, in case
 193   // it is running on a machine that is under load (e.g. the build machine).
 194   th1.Stop();
 195   th_main->ProcessMessages(200);
 196   th2.Stop();
 197
 198   // Make sure the results were correct
 199   EXPECT_EQ(5, msg_client.count);
 200   EXPECT_EQ(34, msg_client.last);
 201   EXPECT_EQ(5, sock_client.count);
 202   EXPECT_EQ(55, sock_client.last);
 203 }
 204
 205 // Test that setting thread names doesn't cause a malfunction.
 206 // There's no easy way to verify the name was set properly at this time.
 207 TEST(ThreadTest, Names) {
 208   // Default name
 209   Thread *thread;
 210   thread = new Thread();
 211   EXPECT_TRUE(thread->Start());
 212   thread->Stop();
 213   delete thread;
 214   thread = new Thread();
 215   // Name with no object parameter
 216   EXPECT_TRUE(thread->SetName("No object", NULL));
 217   EXPECT_TRUE(thread->Start());
 218   thread->Stop();
 219   delete thread;
 220   // Really long name
 221   thread = new Thread();
 222   EXPECT_TRUE(thread->SetName("Abcdefghijklmnopqrstuvwxyz1234567890", this));
 223   EXPECT_TRUE(thread->Start());
 224   thread->Stop();
 225   delete thread;
 226 }
 227
 228 // Test that setting thread priorities doesn't cause a malfunction.
 229 // There's no easy way to verify the priority was set properly at this time.
 230 TEST(ThreadTest, Priorities) {
 231   Thread *thread;
 232   thread = new Thread();
 233   EXPECT_TRUE(thread->SetPriority(PRIORITY_HIGH));
 234   EXPECT_TRUE(thread->Start());
 235   thread->Stop();
 236   delete thread;
 237   thread = new Thread();
 238   EXPECT_TRUE(thread->SetPriority(PRIORITY_ABOVE_NORMAL));
 239   EXPECT_TRUE(thread->Start());
 240   thread->Stop();
 241   delete thread;
 242
 243   thread = new Thread();
 244   EXPECT_TRUE(thread->Start());
 245 #ifdef WIN32
 246   EXPECT_TRUE(thread->SetPriority(PRIORITY_ABOVE_NORMAL));
 247 #else
 248   EXPECT_FALSE(thread->SetPriority(PRIORITY_ABOVE_NORMAL));
 249 #endif
 250   thread->Stop();
 251   delete thread;
 252
 253 }
 254
 255 TEST(ThreadTest, Wrap) {
 256   Thread* current_thread = Thread::Current();
 257   current_thread->UnwrapCurrent();
 258   CustomThread* cthread = new CustomThread();
 259   EXPECT_TRUE(cthread->WrapCurrent());
 260   EXPECT_TRUE(cthread->started());
 261   EXPECT_FALSE(cthread->IsOwned());
 262   cthread->UnwrapCurrent();
 263   EXPECT_FALSE(cthread->started());
 264   delete cthread;
 265   current_thread->WrapCurrent();
 266 }
 267
 268 // Test that calling Release on a thread causes it to self-destruct when
 269 // it's finished running
 270 TEST(ThreadTest, Release) {
 271   scoped_ptr<Event> event(new Event(true, false));
 272   // Ensure the event is initialized.
 273   event->Reset();
 274
 275   Thread* thread = new SignalWhenDestroyedThread(event.get());
 276   thread->Start();
 277   thread->Release();
 278
 279   // The event should get signaled when the thread completes, which should
 280   // be nearly instantaneous, since it doesn't do anything.  For safety,
 281   // give it 3 seconds in case the machine is under load.
 282   bool signaled = event->Wait(3000);
 283   EXPECT_TRUE(signaled);
 284 }
 285
 286 TEST(ThreadTest, Invoke) {
 287   // Create and start the thread.
 288   Thread thread;
 289   thread.Start();
 290   // Try calling functors.
 291   EXPECT_EQ(42, thread.Invoke<int>(Functor1()));
 292   bool called = false;
 293   Functor2 f2(&called);
 294   thread.Invoke<void>(f2);
 295   EXPECT_TRUE(called);
 296   // Try calling bare functions.
 297   struct LocalFuncs {
 298     static int Func1() { return 999; }
 299     static void Func2() {}
 300   };
 301   EXPECT_EQ(999, thread.Invoke<int>(&LocalFuncs::Func1));
 302   thread.Invoke<void>(&LocalFuncs::Func2);
 303 }
 304
 305 #ifdef WIN32
 306 class ComThreadTest : public testing::Test, public MessageHandler {
 307  public:
 308   ComThreadTest() : done_(false) {}
 309  protected:
 310   virtual void OnMessage(Message* message) {
 311     HRESULT hr = CoInitializeEx(NULL, COINIT_MULTITHREADED);
 312     // S_FALSE means the thread was already inited for a multithread apartment.
 313     EXPECT_EQ(S_FALSE, hr);
 314     if (SUCCEEDED(hr)) {
 315       CoUninitialize();
 316     }
 317     done_ = true;
 318   }
 319   bool done_;
 320 };
 321
 322 TEST_F(ComThreadTest, ComInited) {
 323   Thread* thread = new ComThread();
 324   EXPECT_TRUE(thread->Start());
 325   thread->Post(this, 0);
 326   EXPECT_TRUE_WAIT(done_, 1000);
 327   delete thread;
 328 }
 329 #endif