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, ¶m) != 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, ¶m) != 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 580
