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

   1 /*
   2  * libjingle
   3  * Copyright 2004 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/thread.h"
  29
  30 #ifndef __has_feature
  31 #define __has_feature(x) 0  // Compatibility with non-clang or LLVM compilers.
  32 #endif  // __has_feature
  33
  34 #if defined(WIN32)
  35 #include <comdef.h>
  36 #elif defined(POSIX)
  37 #include <time.h>
  38 #endif
  39
  40 #include "talk/base/common.h"
  41 #include "talk/base/logging.h"
  42 #include "talk/base/stringutils.h"
  43 #include "talk/base/timeutils.h"
  44
  45 #if !__has_feature(objc_arc) && (defined(OSX) || defined(IOS))
  46 #include "talk/base/maccocoathreadhelper.h"
  47 #include "talk/base/scoped_autorelease_pool.h"
  48 #endif
  49
  50 namespace talk_base {
  51
  52 ThreadManager* ThreadManager::Instance() {
  53   LIBJINGLE_DEFINE_STATIC_LOCAL(ThreadManager, thread_manager, ());
  54   return &thread_manager;
  55 }
  56
  57 // static
  58 Thread* Thread::Current() {
  59   return ThreadManager::Instance()->CurrentThread();
  60 }
  61
  62 #ifdef POSIX
  63 ThreadManager::ThreadManager() {
  64   pthread_key_create(&key_, NULL);
  65 #ifndef NO_MAIN_THREAD_WRAPPING
  66   WrapCurrentThread();
  67 #endif
  68 #if !__has_feature(objc_arc) && (defined(OSX) || defined(IOS))
  69   // Under Automatic Reference Counting (ARC), you cannot use autorelease pools
  70   // directly. Instead, you use @autoreleasepool blocks instead.  Also, we are
  71   // maintaining thread safety using immutability within context of GCD dispatch
  72   // queues in this case.
  73   InitCocoaMultiThreading();
  74 #endif
  75 }
  76
  77 ThreadManager::~ThreadManager() {
  78 #if __has_feature(objc_arc)
  79   @autoreleasepool
  80 #elif defined(OSX) || defined(IOS)
  81   // This is called during exit, at which point apparently no NSAutoreleasePools
  82   // are available; but we might still need them to do cleanup (or we get the
  83   // "no autoreleasepool in place, just leaking" warning when exiting).
  84   ScopedAutoreleasePool pool;
  85 #endif
  86   {
  87     UnwrapCurrentThread();
  88     pthread_key_delete(key_);
  89   }
  90 }
  91
  92 Thread *ThreadManager::CurrentThread() {
  93   return static_cast<Thread *>(pthread_getspecific(key_));
  94 }
  95
  96 void ThreadManager::SetCurrentThread(Thread *thread) {
  97   pthread_setspecific(key_, thread);
  98 }
  99 #endif
 100
 101 #ifdef WIN32
 102 ThreadManager::ThreadManager() {
 103   key_ = TlsAlloc();
 104 #ifndef NO_MAIN_THREAD_WRAPPING
 105   WrapCurrentThread();
 106 #endif
 107 }
 108
 109 ThreadManager::~ThreadManager() {
 110   UnwrapCurrentThread();
 111   TlsFree(key_);
 112 }
 113
 114 Thread *ThreadManager::CurrentThread() {
 115   return static_cast<Thread *>(TlsGetValue(key_));
 116 }
 117
 118 void ThreadManager::SetCurrentThread(Thread *thread) {
 119   TlsSetValue(key_, thread);
 120 }
 121 #endif
 122
 123 Thread *ThreadManager::WrapCurrentThread() {
 124   Thread* result = CurrentThread();
 125   if (NULL == result) {
 126     result = new Thread();
 127     result->WrapCurrentWithThreadManager(this);
 128   }
 129   return result;
 130 }
 131
 132 void ThreadManager::UnwrapCurrentThread() {
 133   Thread* t = CurrentThread();
 134   if (t && !(t->IsOwned())) {
 135     t->UnwrapCurrent();
 136     delete t;
 137   }
 138 }
 139
 140 struct ThreadInit {
 141   Thread* thread;
 142   Runnable* runnable;
 143 };
 144
 145 Thread::Thread(SocketServer* ss)
 146     : MessageQueue(ss),
 147       priority_(PRIORITY_NORMAL),
 148       started_(false),
 149 #if defined(WIN32)
 150       thread_(NULL),
 151       thread_id_(0),
 152 #endif
 153       owned_(true),
 154       delete_self_when_complete_(false) {
 155   SetName("Thread", this);  // default name
 156 }
 157
 158 Thread::~Thread() {
 159   Stop();
 160   if (active_)
 161     Clear(NULL);
 162 }
 163
 164 bool Thread::SleepMs(int milliseconds) {
 165 #ifdef WIN32
 166   ::Sleep(milliseconds);
 167   return true;
 168 #else
 169   // POSIX has both a usleep() and a nanosleep(), but the former is deprecated,
 170   // so we use nanosleep() even though it has greater precision than necessary.
 171   struct timespec ts;
 172   ts.tv_sec = milliseconds / 1000;
 173   ts.tv_nsec = (milliseconds % 1000) * 1000000;
 174   int ret = nanosleep(&ts, NULL);
 175   if (ret != 0) {
 176     LOG_ERR(LS_WARNING) << "nanosleep() returning early";
 177     return false;
 178   }
 179   return true;
 180 #endif
 181 }
 182
 183 bool Thread::SetName(const std::string& name, const void* obj) {
 184   if (started_) return false;
 185   name_ = name;
 186   if (obj) {
 187     char buf[16];
 188     sprintfn(buf, sizeof(buf), " 0x%p", obj);
 189     name_ += buf;
 190   }
 191   return true;
 192 }
 193
 194 bool Thread::SetPriority(ThreadPriority priority) {
 195 #if defined(WIN32)
 196   if (started_) {
 197     BOOL ret = FALSE;
 198     if (priority == PRIORITY_NORMAL) {
 199       ret = ::SetThreadPriority(thread_, THREAD_PRIORITY_NORMAL);
 200     } else if (priority == PRIORITY_HIGH) {
 201       ret = ::SetThreadPriority(thread_, THREAD_PRIORITY_HIGHEST);
 202     } else if (priority == PRIORITY_ABOVE_NORMAL) {
 203       ret = ::SetThreadPriority(thread_, THREAD_PRIORITY_ABOVE_NORMAL);
 204     } else if (priority == PRIORITY_IDLE) {
 205       ret = ::SetThreadPriority(thread_, THREAD_PRIORITY_IDLE);
 206     }
 207     if (!ret) {
 208       return false;
 209     }
 210   }
 211   priority_ = priority;
 212   return true;
 213 #else
 214   // TODO: Implement for Linux/Mac if possible.
 215   if (started_) return false;
 216   priority_ = priority;
 217   return true;
 218 #endif
 219 }
 220
 221 bool Thread::Start(Runnable* runnable) {
 222   ASSERT(owned_);
 223   if (!owned_) return false;
 224   ASSERT(!started_);
 225   if (started_) return false;
 226
 227   Restart();  // reset fStop_ if the thread is being restarted
 228
 229   // Make sure that ThreadManager is created on the main thread before
 230   // we start a new thread.
 231   ThreadManager::Instance();
 232
 233   ThreadInit* init = new ThreadInit;
 234   init->thread = this;
 235   init->runnable = runnable;
 236 #if defined(WIN32)
 237   DWORD flags = 0;
 238   if (priority_ != PRIORITY_NORMAL) {
 239     flags = CREATE_SUSPENDED;
 240   }
 241   thread_ = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)PreRun, init, flags,
 242                          &thread_id_);
 243   if (thread_) {
 244     started_ = true;
 245     if (priority_ != PRIORITY_NORMAL) {
 246       SetPriority(priority_);
 247       ::ResumeThread(thread_);
 248     }
 249   } else {
 250     return false;
 251   }
 252 #elif defined(POSIX)
 253   pthread_attr_t attr;
 254   pthread_attr_init(&attr);
 255   if (priority_ != PRIORITY_NORMAL) {
 256     if (priority_ == PRIORITY_IDLE) {
 257       // There is no POSIX-standard way to set a below-normal priority for an
 258       // individual thread (only whole process), so let's not support it.
 259       LOG(LS_WARNING) << "PRIORITY_IDLE not supported";
 260     } else {
 261       // Set real-time round-robin policy.
 262       if (pthread_attr_setschedpolicy(&attr, SCHED_RR) != 0) {
 263         LOG(LS_ERROR) << "pthread_attr_setschedpolicy";
 264       }
 265       struct sched_param param;
 266       if (pthread_attr_getschedparam(&attr, &param) != 0) {
 267         LOG(LS_ERROR) << "pthread_attr_getschedparam";
 268       } else {
 269         // The numbers here are arbitrary.
 270         if (priority_ == PRIORITY_HIGH) {
 271           param.sched_priority = 6;           // 6 = HIGH
 272         } else {
 273           ASSERT(priority_ == PRIORITY_ABOVE_NORMAL);
 274           param.sched_priority = 4;           // 4 = ABOVE_NORMAL
 275         }
 276         if (pthread_attr_setschedparam(&attr, &param) != 0) {
 277           LOG(LS_ERROR) << "pthread_attr_setschedparam";
 278         }
 279       }
 280     }
 281   }
 282   int error_code = pthread_create(&thread_, &attr, PreRun, init);
 283   if (0 != error_code) {
 284     LOG(LS_ERROR) << "Unable to create pthread, error " << error_code;
 285     return false;
 286   }
 287   started_ = true;
 288 #endif
 289   return true;
 290 }
 291
 292 void Thread::Join() {
 293   if (started_) {
 294     ASSERT(!IsCurrent());
 295 #if defined(WIN32)
 296     WaitForSingleObject(thread_, INFINITE);
 297     CloseHandle(thread_);
 298     thread_ = NULL;
 299     thread_id_ = 0;
 300 #elif defined(POSIX)
 301     void *pv;
 302     pthread_join(thread_, &pv);
 303 #endif
 304     started_ = false;
 305   }
 306 }
 307
 308 #ifdef WIN32
 309 // As seen on MSDN.
 310 // http://msdn.microsoft.com/en-us/library/xcb2z8hs(VS.71).aspx
 311 #define MSDEV_SET_THREAD_NAME  0x406D1388
 312 typedef struct tagTHREADNAME_INFO {
 313   DWORD dwType;
 314   LPCSTR szName;
 315   DWORD dwThreadID;
 316   DWORD dwFlags;
 317 } THREADNAME_INFO;
 318
 319 void SetThreadName(DWORD dwThreadID, LPCSTR szThreadName) {
 320   THREADNAME_INFO info;
 321   info.dwType = 0x1000;
 322   info.szName = szThreadName;
 323   info.dwThreadID = dwThreadID;
 324   info.dwFlags = 0;
 325
 326   __try {
 327     RaiseException(MSDEV_SET_THREAD_NAME, 0, sizeof(info) / sizeof(DWORD),
 328                    reinterpret_cast<ULONG_PTR*>(&info));
 329   }
 330   __except(EXCEPTION_CONTINUE_EXECUTION) {
 331   }
 332 }
 333 #endif  // WIN32
 334
 335 void* Thread::PreRun(void* pv) {
 336   ThreadInit* init = static_cast<ThreadInit*>(pv);
 337   ThreadManager::Instance()->SetCurrentThread(init->thread);
 338 #if defined(WIN32)
 339   SetThreadName(GetCurrentThreadId(), init->thread->name_.c_str());
 340 #elif defined(POSIX)
 341   // TODO: See if naming exists for pthreads.
 342 #endif
 343 #if __has_feature(objc_arc)
 344   @autoreleasepool
 345 #elif defined(OSX) || defined(IOS)
 346   // Make sure the new thread has an autoreleasepool
 347   ScopedAutoreleasePool pool;
 348 #endif
 349   {
 350     if (init->runnable) {
 351       init->runnable->Run(init->thread);
 352     } else {
 353       init->thread->Run();
 354     }
 355     if (init->thread->delete_self_when_complete_) {
 356       init->thread->started_ = false;
 357       delete init->thread;
 358     }
 359     delete init;
 360     return NULL;
 361   }
 362 }
 363
 364 void Thread::Run() {
 365   ProcessMessages(kForever);
 366 }
 367
 368 bool Thread::IsOwned() {
 369   return owned_;
 370 }
 371
 372 void Thread::Stop() {
 373   MessageQueue::Quit();
 374   Join();
 375 }
 376
 377 void Thread::Send(MessageHandler *phandler, uint32 id, MessageData *pdata) {
 378   if (fStop_)
 379     return;
 380
 381   // Sent messages are sent to the MessageHandler directly, in the context
 382   // of "thread", like Win32 SendMessage. If in the right context,
 383   // call the handler directly.
 384
 385   Message msg;
 386   msg.phandler = phandler;
 387   msg.message_id = id;
 388   msg.pdata = pdata;
 389   if (IsCurrent()) {
 390     phandler->OnMessage(&msg);
 391     return;
 392   }
 393
 394   AutoThread thread;
 395   Thread *current_thread = Thread::Current();
 396   ASSERT(current_thread != NULL);  // AutoThread ensures this
 397
 398   bool ready = false;
 399   {
 400     CritScope cs(&crit_);
 401     EnsureActive();
 402     _SendMessage smsg;
 403     smsg.thread = current_thread;
 404     smsg.msg = msg;
 405     smsg.ready = &ready;
 406     sendlist_.push_back(smsg);
 407   }
 408
 409   // Wait for a reply
 410
 411   ss_->WakeUp();
 412
 413   bool waited = false;
 414   crit_.Enter();
 415   while (!ready) {
 416     crit_.Leave();
 417     current_thread->ReceiveSends();
 418     current_thread->socketserver()->Wait(kForever, false);
 419     waited = true;
 420     crit_.Enter();
 421   }
 422   crit_.Leave();
 423
 424   // Our Wait loop above may have consumed some WakeUp events for this
 425   // MessageQueue, that weren't relevant to this Send.  Losing these WakeUps can
 426   // cause problems for some SocketServers.
 427   //
 428   // Concrete example:
 429   // Win32SocketServer on thread A calls Send on thread B.  While processing the
 430   // message, thread B Posts a message to A.  We consume the wakeup for that
 431   // Post while waiting for the Send to complete, which means that when we exit
 432   // this loop, we need to issue another WakeUp, or else the Posted message
 433   // won't be processed in a timely manner.
 434
 435   if (waited) {
 436     current_thread->socketserver()->WakeUp();
 437   }
 438 }
 439
 440 void Thread::ReceiveSends() {
 441   // Receive a sent message. Cleanup scenarios:
 442   // - thread sending exits: We don't allow this, since thread can exit
 443   //   only via Join, so Send must complete.
 444   // - thread receiving exits: Wakeup/set ready in Thread::Clear()
 445   // - object target cleared: Wakeup/set ready in Thread::Clear()
 446   crit_.Enter();
 447   while (!sendlist_.empty()) {
 448     _SendMessage smsg = sendlist_.front();
 449     sendlist_.pop_front();
 450     crit_.Leave();
 451     smsg.msg.phandler->OnMessage(&smsg.msg);
 452     crit_.Enter();
 453     *smsg.ready = true;
 454     smsg.thread->socketserver()->WakeUp();
 455   }
 456   crit_.Leave();
 457 }
 458
 459 void Thread::Clear(MessageHandler *phandler, uint32 id,
 460                    MessageList* removed) {
 461   CritScope cs(&crit_);
 462
 463   // Remove messages on sendlist_ with phandler
 464   // Object target cleared: remove from send list, wakeup/set ready
 465   // if sender not NULL.
 466
 467   std::list<_SendMessage>::iterator iter = sendlist_.begin();
 468   while (iter != sendlist_.end()) {
 469     _SendMessage smsg = *iter;
 470     if (smsg.msg.Match(phandler, id)) {
 471       if (removed) {
 472         removed->push_back(smsg.msg);
 473       } else {
 474         delete smsg.msg.pdata;
 475       }
 476       iter = sendlist_.erase(iter);
 477       *smsg.ready = true;
 478       smsg.thread->socketserver()->WakeUp();
 479       continue;
 480     }
 481     ++iter;
 482   }
 483
 484   MessageQueue::Clear(phandler, id, removed);
 485 }
 486
 487 bool Thread::ProcessMessages(int cmsLoop) {
 488   uint32 msEnd = (kForever == cmsLoop) ? 0 : TimeAfter(cmsLoop);
 489   int cmsNext = cmsLoop;
 490
 491   while (true) {
 492 #if __has_feature(objc_arc)
 493     @autoreleasepool
 494 #elif defined(OSX) || defined(IOS)
 495     // see: http://developer.apple.com/library/mac/#documentation/Cocoa/Reference/Foundation/Classes/NSAutoreleasePool_Class/Reference/Reference.html
 496     // Each thread is supposed to have an autorelease pool. Also for event loops
 497     // like this, autorelease pool needs to be created and drained/released
 498     // for each cycle.
 499     ScopedAutoreleasePool pool;
 500 #endif
 501     {
 502       Message msg;
 503       if (!Get(&msg, cmsNext))
 504         return !IsQuitting();
 505       Dispatch(&msg);
 506
 507       if (cmsLoop != kForever) {
 508         cmsNext = TimeUntil(msEnd);
 509         if (cmsNext < 0)
 510           return true;
 511       }
 512     }
 513   }
 514 }
 515
 516 bool Thread::WrapCurrent() {
 517   return WrapCurrentWithThreadManager(ThreadManager::Instance());
 518 }
 519
 520 bool Thread::WrapCurrentWithThreadManager(ThreadManager* thread_manager) {
 521   if (started_)
 522     return false;
 523 #if defined(WIN32)
 524   // We explicitly ask for no rights other than synchronization.
 525   // This gives us the best chance of succeeding.
 526   thread_ = OpenThread(SYNCHRONIZE, FALSE, GetCurrentThreadId());
 527   if (!thread_) {
 528     LOG_GLE(LS_ERROR) << "Unable to get handle to thread.";
 529     return false;
 530   }
 531   thread_id_ = GetCurrentThreadId();
 532 #elif defined(POSIX)
 533   thread_ = pthread_self();
 534 #endif
 535   owned_ = false;
 536   started_ = true;
 537   thread_manager->SetCurrentThread(this);
 538   return true;
 539 }
 540
 541 void Thread::UnwrapCurrent() {
 542   // Clears the platform-specific thread-specific storage.
 543   ThreadManager::Instance()->SetCurrentThread(NULL);
 544 #ifdef WIN32
 545   if (!CloseHandle(thread_)) {
 546     LOG_GLE(LS_ERROR) << "When unwrapping thread, failed to close handle.";
 547   }
 548 #endif
 549   started_ = false;
 550 }
 551
 552
 553 AutoThread::AutoThread(SocketServer* ss) : Thread(ss) {
 554   if (!ThreadManager::Instance()->CurrentThread()) {
 555     ThreadManager::Instance()->SetCurrentThread(this);
 556   }
 557 }
 558
 559 AutoThread::~AutoThread() {
 560   Stop();
 561   if (ThreadManager::Instance()->CurrentThread() == this) {
 562     ThreadManager::Instance()->SetCurrentThread(NULL);
 563   }
 564 }
 565
 566 #ifdef WIN32
 567 void ComThread::Run() {
 568   HRESULT hr = CoInitializeEx(NULL, COINIT_MULTITHREADED);
 569   ASSERT(SUCCEEDED(hr));
 570   if (SUCCEEDED(hr)) {
 571     Thread::Run();
 572     CoUninitialize();
 573   } else {
 574     LOG(LS_ERROR) << "CoInitialize failed, hr=" << hr;
 575   }
 576 }
 577 #endif
 578
 579 }  // namespace talk_base