Note: We no longer publish the latest version of our code here. We primarily use a kumc-bmi github organization. The heron ETL repository, in particular, is not public. Peers in the informatics community should see MultiSiteDev for details on requesting access.

source: webrtc/talk/base/thread.cc @ 0:4bda6873e34c

pub_scrub_3792 tip
Last change on this file since 0:4bda6873e34c was 0:4bda6873e34c, checked in by Michael Prittie <mprittie@…>, 6 years ago

Scrubbed password for publication.

File size: 15.4 KB
Line 
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
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#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);
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, &param) != 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, &param) != 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
292void 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
312typedef struct tagTHREADNAME_INFO {
313  DWORD dwType;
314  LPCSTR szName;
315  DWORD dwThreadID;
316  DWORD dwFlags;
317} THREADNAME_INFO;
318
319void 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
335void* 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
364void Thread::Run() {
365  ProcessMessages(kForever);
366}
367
368bool Thread::IsOwned() {
369  return owned_;
370}
371
372void Thread::Stop() {
373  MessageQueue::Quit();
374  Join();
375}
376
377void 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
440void 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
459void 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
487bool 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
516bool Thread::WrapCurrent() {
517  return WrapCurrentWithThreadManager(ThreadManager::Instance());
518}
519
520bool 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
541void 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
553AutoThread::AutoThread(SocketServer* ss) : Thread(ss) {
554  if (!ThreadManager::Instance()->CurrentThread()) {
555    ThreadManager::Instance()->SetCurrentThread(this);
556  }
557}
558
559AutoThread::~AutoThread() {
560  Stop();
561  if (ThreadManager::Instance()->CurrentThread() == this) {
562    ThreadManager::Instance()->SetCurrentThread(NULL);
563  }
564}
565
566#ifdef WIN32
567void 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
Note: See TracBrowser for help on using the repository browser.