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