blob: d4cebc4d55ad9a58a8078da8658c17f727176fc0 [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),
henrike@webrtc.org28e20752013-07-10 00:45:36149#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
158Thread::~Thread() {
159 Stop();
160 if (active_)
161 Clear(NULL);
162}
163
164bool 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
183bool 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
194bool 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
221bool 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);
wu@webrtc.orgf6d6ed0c2014-01-03 22:08:47255
256 // Thread priorities are not supported in NaCl.
257#if !defined(__native_client__)
henrike@webrtc.org28e20752013-07-10 00:45:36258 if (priority_ != PRIORITY_NORMAL) {
259 if (priority_ == PRIORITY_IDLE) {
260 // There is no POSIX-standard way to set a below-normal priority for an
261 // individual thread (only whole process), so let's not support it.
262 LOG(LS_WARNING) << "PRIORITY_IDLE not supported";
263 } else {
264 // Set real-time round-robin policy.
265 if (pthread_attr_setschedpolicy(&attr, SCHED_RR) != 0) {
266 LOG(LS_ERROR) << "pthread_attr_setschedpolicy";
267 }
268 struct sched_param param;
269 if (pthread_attr_getschedparam(&attr, &param) != 0) {
270 LOG(LS_ERROR) << "pthread_attr_getschedparam";
271 } else {
272 // The numbers here are arbitrary.
273 if (priority_ == PRIORITY_HIGH) {
274 param.sched_priority = 6; // 6 = HIGH
275 } else {
276 ASSERT(priority_ == PRIORITY_ABOVE_NORMAL);
277 param.sched_priority = 4; // 4 = ABOVE_NORMAL
278 }
279 if (pthread_attr_setschedparam(&attr, &param) != 0) {
280 LOG(LS_ERROR) << "pthread_attr_setschedparam";
281 }
282 }
283 }
284 }
wu@webrtc.orgf6d6ed0c2014-01-03 22:08:47285#endif // !defined(__native_client__)
286
henrike@webrtc.org28e20752013-07-10 00:45:36287 int error_code = pthread_create(&thread_, &attr, PreRun, init);
288 if (0 != error_code) {
289 LOG(LS_ERROR) << "Unable to create pthread, error " << error_code;
290 return false;
291 }
292 started_ = true;
293#endif
294 return true;
295}
296
297void Thread::Join() {
298 if (started_) {
299 ASSERT(!IsCurrent());
300#if defined(WIN32)
301 WaitForSingleObject(thread_, INFINITE);
302 CloseHandle(thread_);
303 thread_ = NULL;
304 thread_id_ = 0;
305#elif defined(POSIX)
306 void *pv;
307 pthread_join(thread_, &pv);
308#endif
309 started_ = false;
310 }
311}
312
313#ifdef WIN32
314// As seen on MSDN.
315// http://msdn.microsoft.com/en-us/library/xcb2z8hs(VS.71).aspx
316#define MSDEV_SET_THREAD_NAME 0x406D1388
317typedef struct tagTHREADNAME_INFO {
318 DWORD dwType;
319 LPCSTR szName;
320 DWORD dwThreadID;
321 DWORD dwFlags;
322} THREADNAME_INFO;
323
324void SetThreadName(DWORD dwThreadID, LPCSTR szThreadName) {
325 THREADNAME_INFO info;
326 info.dwType = 0x1000;
327 info.szName = szThreadName;
328 info.dwThreadID = dwThreadID;
329 info.dwFlags = 0;
330
331 __try {
332 RaiseException(MSDEV_SET_THREAD_NAME, 0, sizeof(info) / sizeof(DWORD),
333 reinterpret_cast<ULONG_PTR*>(&info));
334 }
335 __except(EXCEPTION_CONTINUE_EXECUTION) {
336 }
337}
338#endif // WIN32
339
340void* Thread::PreRun(void* pv) {
341 ThreadInit* init = static_cast<ThreadInit*>(pv);
342 ThreadManager::Instance()->SetCurrentThread(init->thread);
343#if defined(WIN32)
344 SetThreadName(GetCurrentThreadId(), init->thread->name_.c_str());
345#elif defined(POSIX)
346 // TODO: See if naming exists for pthreads.
347#endif
348#if __has_feature(objc_arc)
349 @autoreleasepool
350#elif defined(OSX) || defined(IOS)
351 // Make sure the new thread has an autoreleasepool
352 ScopedAutoreleasePool pool;
353#endif
354 {
355 if (init->runnable) {
356 init->runnable->Run(init->thread);
357 } else {
358 init->thread->Run();
359 }
360 if (init->thread->delete_self_when_complete_) {
361 init->thread->started_ = false;
362 delete init->thread;
363 }
364 delete init;
365 return NULL;
366 }
367}
368
369void Thread::Run() {
370 ProcessMessages(kForever);
371}
372
373bool Thread::IsOwned() {
374 return owned_;
375}
376
377void Thread::Stop() {
378 MessageQueue::Quit();
379 Join();
380}
381
382void Thread::Send(MessageHandler *phandler, uint32 id, MessageData *pdata) {
383 if (fStop_)
384 return;
385
386 // Sent messages are sent to the MessageHandler directly, in the context
387 // of "thread", like Win32 SendMessage. If in the right context,
388 // call the handler directly.
389
390 Message msg;
391 msg.phandler = phandler;
392 msg.message_id = id;
393 msg.pdata = pdata;
394 if (IsCurrent()) {
395 phandler->OnMessage(&msg);
396 return;
397 }
398
399 AutoThread thread;
400 Thread *current_thread = Thread::Current();
401 ASSERT(current_thread != NULL); // AutoThread ensures this
402
403 bool ready = false;
404 {
405 CritScope cs(&crit_);
406 EnsureActive();
407 _SendMessage smsg;
408 smsg.thread = current_thread;
409 smsg.msg = msg;
410 smsg.ready = &ready;
411 sendlist_.push_back(smsg);
henrike@webrtc.org28e20752013-07-10 00:45:36412 }
413
414 // Wait for a reply
415
416 ss_->WakeUp();
417
418 bool waited = false;
mallinath@webrtc.org19f27e62013-10-13 17:18:27419 crit_.Enter();
henrike@webrtc.org28e20752013-07-10 00:45:36420 while (!ready) {
mallinath@webrtc.org19f27e62013-10-13 17:18:27421 crit_.Leave();
henrike@webrtc.org28e20752013-07-10 00:45:36422 current_thread->ReceiveSends();
423 current_thread->socketserver()->Wait(kForever, false);
424 waited = true;
mallinath@webrtc.org19f27e62013-10-13 17:18:27425 crit_.Enter();
henrike@webrtc.org28e20752013-07-10 00:45:36426 }
mallinath@webrtc.org19f27e62013-10-13 17:18:27427 crit_.Leave();
henrike@webrtc.org28e20752013-07-10 00:45:36428
429 // Our Wait loop above may have consumed some WakeUp events for this
430 // MessageQueue, that weren't relevant to this Send. Losing these WakeUps can
431 // cause problems for some SocketServers.
432 //
433 // Concrete example:
434 // Win32SocketServer on thread A calls Send on thread B. While processing the
435 // message, thread B Posts a message to A. We consume the wakeup for that
436 // Post while waiting for the Send to complete, which means that when we exit
437 // this loop, we need to issue another WakeUp, or else the Posted message
438 // won't be processed in a timely manner.
439
440 if (waited) {
441 current_thread->socketserver()->WakeUp();
442 }
443}
444
445void Thread::ReceiveSends() {
henrike@webrtc.org28e20752013-07-10 00:45:36446 // Receive a sent message. Cleanup scenarios:
447 // - thread sending exits: We don't allow this, since thread can exit
448 // only via Join, so Send must complete.
449 // - thread receiving exits: Wakeup/set ready in Thread::Clear()
450 // - object target cleared: Wakeup/set ready in Thread::Clear()
451 crit_.Enter();
452 while (!sendlist_.empty()) {
453 _SendMessage smsg = sendlist_.front();
454 sendlist_.pop_front();
455 crit_.Leave();
456 smsg.msg.phandler->OnMessage(&smsg.msg);
457 crit_.Enter();
458 *smsg.ready = true;
459 smsg.thread->socketserver()->WakeUp();
460 }
henrike@webrtc.org28e20752013-07-10 00:45:36461 crit_.Leave();
462}
463
464void Thread::Clear(MessageHandler *phandler, uint32 id,
465 MessageList* removed) {
466 CritScope cs(&crit_);
467
468 // Remove messages on sendlist_ with phandler
469 // Object target cleared: remove from send list, wakeup/set ready
470 // if sender not NULL.
471
472 std::list<_SendMessage>::iterator iter = sendlist_.begin();
473 while (iter != sendlist_.end()) {
474 _SendMessage smsg = *iter;
475 if (smsg.msg.Match(phandler, id)) {
476 if (removed) {
477 removed->push_back(smsg.msg);
478 } else {
479 delete smsg.msg.pdata;
480 }
481 iter = sendlist_.erase(iter);
482 *smsg.ready = true;
483 smsg.thread->socketserver()->WakeUp();
484 continue;
485 }
486 ++iter;
487 }
488
489 MessageQueue::Clear(phandler, id, removed);
490}
491
492bool Thread::ProcessMessages(int cmsLoop) {
493 uint32 msEnd = (kForever == cmsLoop) ? 0 : TimeAfter(cmsLoop);
494 int cmsNext = cmsLoop;
495
496 while (true) {
497#if __has_feature(objc_arc)
498 @autoreleasepool
499#elif defined(OSX) || defined(IOS)
500 // see: http://developer.apple.com/library/mac/#documentation/Cocoa/Reference/Foundation/Classes/NSAutoreleasePool_Class/Reference/Reference.html
501 // Each thread is supposed to have an autorelease pool. Also for event loops
502 // like this, autorelease pool needs to be created and drained/released
503 // for each cycle.
504 ScopedAutoreleasePool pool;
505#endif
506 {
507 Message msg;
508 if (!Get(&msg, cmsNext))
509 return !IsQuitting();
510 Dispatch(&msg);
511
512 if (cmsLoop != kForever) {
513 cmsNext = TimeUntil(msEnd);
514 if (cmsNext < 0)
515 return true;
516 }
517 }
518 }
519}
520
521bool Thread::WrapCurrent() {
522 return WrapCurrentWithThreadManager(ThreadManager::Instance());
523}
524
525bool Thread::WrapCurrentWithThreadManager(ThreadManager* thread_manager) {
526 if (started_)
527 return false;
528#if defined(WIN32)
529 // We explicitly ask for no rights other than synchronization.
530 // This gives us the best chance of succeeding.
531 thread_ = OpenThread(SYNCHRONIZE, FALSE, GetCurrentThreadId());
532 if (!thread_) {
533 LOG_GLE(LS_ERROR) << "Unable to get handle to thread.";
534 return false;
535 }
536 thread_id_ = GetCurrentThreadId();
537#elif defined(POSIX)
538 thread_ = pthread_self();
539#endif
540 owned_ = false;
541 started_ = true;
542 thread_manager->SetCurrentThread(this);
543 return true;
544}
545
546void Thread::UnwrapCurrent() {
547 // Clears the platform-specific thread-specific storage.
548 ThreadManager::Instance()->SetCurrentThread(NULL);
549#ifdef WIN32
550 if (!CloseHandle(thread_)) {
551 LOG_GLE(LS_ERROR) << "When unwrapping thread, failed to close handle.";
552 }
553#endif
554 started_ = false;
555}
556
557
558AutoThread::AutoThread(SocketServer* ss) : Thread(ss) {
559 if (!ThreadManager::Instance()->CurrentThread()) {
560 ThreadManager::Instance()->SetCurrentThread(this);
561 }
562}
563
564AutoThread::~AutoThread() {
wu@webrtc.org3c5d2b42013-10-18 16:27:26565 Stop();
henrike@webrtc.org28e20752013-07-10 00:45:36566 if (ThreadManager::Instance()->CurrentThread() == this) {
567 ThreadManager::Instance()->SetCurrentThread(NULL);
568 }
569}
570
571#ifdef WIN32
572void ComThread::Run() {
573 HRESULT hr = CoInitializeEx(NULL, COINIT_MULTITHREADED);
574 ASSERT(SUCCEEDED(hr));
575 if (SUCCEEDED(hr)) {
576 Thread::Run();
577 CoUninitialize();
578 } else {
579 LOG(LS_ERROR) << "CoInitialize failed, hr=" << hr;
580 }
581}
582#endif
583
584} // namespace talk_base