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