blob: d21d5f195031ad776cf5731d6a0509ddb49ed477 [file] [log] [blame]
henrike@webrtc.org28e20752013-07-10 00:45:361/*
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
50namespace talk_base {
51
52ThreadManager* ThreadManager::Instance() {
53 LIBJINGLE_DEFINE_STATIC_LOCAL(ThreadManager, thread_manager, ());
54 return &thread_manager;
55}
56
57// static
58Thread* Thread::Current() {
59 return ThreadManager::Instance()->CurrentThread();
60}
61
62#ifdef POSIX
63ThreadManager::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
77ThreadManager::~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
92Thread *ThreadManager::CurrentThread() {
93 return static_cast<Thread *>(pthread_getspecific(key_));
94}
95
96void ThreadManager::SetCurrentThread(Thread *thread) {
97 pthread_setspecific(key_, thread);
98}
99#endif
100
101#ifdef WIN32
102ThreadManager::ThreadManager() {
103 key_ = TlsAlloc();
104#ifndef NO_MAIN_THREAD_WRAPPING
105 WrapCurrentThread();
106#endif
107}
108
109ThreadManager::~ThreadManager() {
110 UnwrapCurrentThread();
111 TlsFree(key_);
112}
113
114Thread *ThreadManager::CurrentThread() {
115 return static_cast<Thread *>(TlsGetValue(key_));
116}
117
118void ThreadManager::SetCurrentThread(Thread *thread) {
119 TlsSetValue(key_, thread);
120}
121#endif
122
123Thread *ThreadManager::WrapCurrentThread() {
124 Thread* result = CurrentThread();
125 if (NULL == result) {
126 result = new Thread();
127 result->WrapCurrentWithThreadManager(this);
128 }
129 return result;
130}
131
132void ThreadManager::UnwrapCurrentThread() {
133 Thread* t = CurrentThread();
134 if (t && !(t->IsOwned())) {
135 t->UnwrapCurrent();
136 delete t;
137 }
138}
139
140struct ThreadInit {
141 Thread* thread;
142 Runnable* runnable;
143};
144
145Thread::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
159Thread::~Thread() {
160 Stop();
161 if (active_)
162 Clear(NULL);
163}
164
165bool 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
184bool 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
195bool 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
222bool 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
293void 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
313typedef struct tagTHREADNAME_INFO {
314 DWORD dwType;
315 LPCSTR szName;
316 DWORD dwThreadID;
317 DWORD dwFlags;
318} THREADNAME_INFO;
319
320void 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
336void* 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
365void Thread::Run() {
366 ProcessMessages(kForever);
367}
368
369bool Thread::IsOwned() {
370 return owned_;
371}
372
373void Thread::Stop() {
374 MessageQueue::Quit();
375 Join();
376}
377
378void 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
438void 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
463void 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
491bool 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
520bool Thread::WrapCurrent() {
521 return WrapCurrentWithThreadManager(ThreadManager::Instance());
522}
523
524bool 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
545void 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
557AutoThread::AutoThread(SocketServer* ss) : Thread(ss) {
558 if (!ThreadManager::Instance()->CurrentThread()) {
559 ThreadManager::Instance()->SetCurrentThread(this);
560 }
561}
562
563AutoThread::~AutoThread() {
564 if (ThreadManager::Instance()->CurrentThread() == this) {
565 ThreadManager::Instance()->SetCurrentThread(NULL);
566 }
567}
568
569#ifdef WIN32
570void 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