system_wrappers/source/thread_posix.cc - external/webrtc/stable/src - Git at Google

 /*
  *  Copyright (c) 2012 The WebRTC project authors. All Rights Reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */

 // The state of a thread is controlled by the two member variables
 // _alive and _dead.
 // _alive represents the state the thread has been ordered to achieve.
 // It is set to true by the thread at startup, and is set to false by
 // other threads, using SetNotAlive() and Stop().
 // _dead represents the state the thread has achieved.
 // It is written by the thread encapsulated by this class only
 // (except at init). It is read only by the Stop() method.
 // The Run() method fires _event when it's started; this ensures that the
 // Start() method does not continue until after _dead is false.
 // This protects against premature Stop() calls from the creator thread, but
 // not from other threads.

 // Their transitions and states:
 // _alive    _dead  Set by
 // false     true   Constructor
 // true      false  Run() method entry
 // false     any    Run() method runFunction failure
 // any       false  Run() method exit (happens only with _alive false)
 // false     any    SetNotAlive
 // false     any    Stop         Stop waits for _dead to become true.
 //
 // Summarized a different way:
 // Variable   Writer               Reader
 // _alive     Constructor(false)   Run.loop
 //            Run.start(true)
 //            Run.fail(false)
 //            SetNotAlive(false)
 //            Stop(false)
 //
 // _dead      Constructor(true)    Stop.loop
 //            Run.start(false)
 //            Run.exit(true)

 #include "thread_posix.h"

 #include <algorithm>

 #include <assert.h>
 #include <errno.h>
 #include <string.h> // strncpy
 #include <time.h>   // nanosleep
 #include <unistd.h>
 #ifdef WEBRTC_LINUX
 #include <sys/types.h>
 #include <sched.h>
 #include <sys/syscall.h>
 #include <linux/unistd.h>
 #include <sys/prctl.h>
 #endif

 #if defined(WEBRTC_MAC)
 #include <mach/mach.h>
 #endif

 #include "system_wrappers/interface/critical_section_wrapper.h"
 #include "system_wrappers/interface/event_wrapper.h"
 #include "system_wrappers/interface/trace.h"

 namespace webrtc {

 int ConvertToSystemPriority(ThreadPriority priority, int minPrio, int maxPrio)
 {
     assert(maxPrio - minPrio > 2);
     const int topPrio = maxPrio - 1;
     const int lowPrio = minPrio + 1;

     switch (priority)
     {
     case kLowPriority:
         return lowPrio;
     case kNormalPriority:
         // The -1 ensures that the kHighPriority is always greater or equal to
         // kNormalPriority.
         return (lowPrio + topPrio - 1) / 2;
     case kHighPriority:
         return std::max(topPrio - 2, lowPrio);
     case kHighestPriority:
         return std::max(topPrio - 1, lowPrio);
     case kRealtimePriority:
         return topPrio;
     }
     assert(false);
     return lowPrio;
 }

 extern "C"
 {
     static void* StartThread(void* lpParameter)
     {
         static_cast<ThreadPosix*>(lpParameter)->Run();
         return 0;
     }
 }

 ThreadWrapper* ThreadPosix::Create(ThreadRunFunction func, ThreadObj obj,
                                    ThreadPriority prio, const char* threadName)
 {
     ThreadPosix* ptr = new ThreadPosix(func, obj, prio, threadName);
     if (!ptr)
     {
         return NULL;
     }
     const int error = ptr->Construct();
     if (error)
     {
         delete ptr;
         return NULL;
     }
     return ptr;
 }

 ThreadPosix::ThreadPosix(ThreadRunFunction func, ThreadObj obj,
                          ThreadPriority prio, const char* threadName)
     : _runFunction(func),
       _obj(obj),
       _crit_state(CriticalSectionWrapper::CreateCriticalSection()),
       _alive(false),
       _dead(true),
       _prio(prio),
       _event(EventWrapper::Create()),
       _name(),
       _setThreadName(false),
 #if (defined(WEBRTC_LINUX) || defined(WEBRTC_ANDROID))
       _pid(-1),
 #endif
       _attr(),
       _thread(0)
 {
     if (threadName != NULL)
     {
         _setThreadName = true;
         strncpy(_name, threadName, kThreadMaxNameLength);
         _name[kThreadMaxNameLength - 1] = '\0';
     }
 }

 uint32_t ThreadWrapper::GetThreadId() {
 #if defined(WEBRTC_ANDROID) || defined(WEBRTC_LINUX)
   return static_cast<uint32_t>(syscall(__NR_gettid));
 #elif defined(WEBRTC_MAC)
   return static_cast<uint32_t>(mach_thread_self());
 #else
   return reinterpret_cast<uint32_t>(pthread_self());
 #endif
 }

 int ThreadPosix::Construct()
 {
     int result = 0;
 #if !defined(WEBRTC_ANDROID)
     // Enable immediate cancellation if requested, see Shutdown()
     result = pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
     if (result != 0)
     {
         return -1;
     }
     result = pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
     if (result != 0)
     {
         return -1;
     }
 #endif
     result = pthread_attr_init(&_attr);
     if (result != 0)
     {
         return -1;
     }
     return 0;
 }

 ThreadPosix::~ThreadPosix()
 {
     pthread_attr_destroy(&_attr);
     delete _event;
     delete _crit_state;
 }

 #define HAS_THREAD_ID !defined(WEBRTC_IOS) && !defined(WEBRTC_MAC)
 #if HAS_THREAD_ID
 bool ThreadPosix::Start(unsigned int& threadID)
 #else
 bool ThreadPosix::Start(unsigned int& /*threadID*/)
 #endif
 {
     if (!_runFunction)
     {
         return false;
     }
     int result = pthread_attr_setdetachstate(&_attr, PTHREAD_CREATE_DETACHED);
     // Set the stack stack size to 1M.
     result |= pthread_attr_setstacksize(&_attr, 1024*1024);
 #ifdef WEBRTC_THREAD_RR
     const int policy = SCHED_RR;
 #else
     const int policy = SCHED_FIFO;
 #endif
     _event->Reset();
     // If pthread_create was successful, a thread was created and is running.
     // Don't return false if it was successful since if there are any other
     // failures the state will be: thread was started but not configured as
     // asked for. However, the caller of this API will assume that a false
     // return value means that the thread never started.
     result |= pthread_create(&_thread, &_attr, &StartThread, this);
     if (result != 0)
     {
         return false;
     }

     // Wait up to 10 seconds for the OS to call the callback function. Prevents
     // race condition if Stop() is called too quickly after start.
     if (kEventSignaled != _event->Wait(WEBRTC_EVENT_10_SEC))
     {
         WEBRTC_TRACE(kTraceError, kTraceUtility, -1,
                      "posix thread event never triggered");
         // Timed out. Something went wrong.
         _runFunction = NULL;
         return true;
     }

 #if HAS_THREAD_ID
     threadID = static_cast<unsigned int>(_thread);
 #endif
     sched_param param;

     const int minPrio = sched_get_priority_min(policy);
     const int maxPrio = sched_get_priority_max(policy);
     if ((minPrio == EINVAL) || (maxPrio == EINVAL))
     {
         WEBRTC_TRACE(kTraceError, kTraceUtility, -1,
                      "unable to retreive min or max priority for threads");
         return true;
     }
     if (maxPrio - minPrio <= 2)
     {
       // There is no room for setting priorities with any granularity.
       return true;
     }
     param.sched_priority = ConvertToSystemPriority(_prio, minPrio, maxPrio);
     result = pthread_setschedparam(_thread, policy, &param);
     if (result == EINVAL)
     {
         WEBRTC_TRACE(kTraceError, kTraceUtility, -1,
                      "unable to set thread priority");
     }
     return true;
 }

 // CPU_ZERO and CPU_SET are not available in NDK r7, so disable
 // SetAffinity on Android for now.
 #if (defined(WEBRTC_LINUX) && (!defined(WEBRTC_ANDROID)))
 bool ThreadPosix::SetAffinity(const int* processorNumbers,
                               const unsigned int amountOfProcessors) {
   if (!processorNumbers || (amountOfProcessors == 0)) {
     return false;
   }
   cpu_set_t mask;
   CPU_ZERO(&mask);

   for (unsigned int processor = 0;
       processor < amountOfProcessors;
       processor++) {
     CPU_SET(processorNumbers[processor], &mask);
   }
 #if defined(WEBRTC_ANDROID)
   // Android.
   const int result = syscall(__NR_sched_setaffinity,
                              _pid,
                              sizeof(mask),
                              &mask);
 #else
   // "Normal" Linux.
   const int result = sched_setaffinity(_pid,
                                        sizeof(mask),
                                        &mask);
 #endif
   if (result != 0) {
     return false;
   }
   return true;
 }

 #else
 // NOTE: On Mac OS X, use the Thread affinity API in
 // /usr/include/mach/thread_policy.h: thread_policy_set and mach_thread_self()
 // instead of Linux gettid() syscall.
 bool ThreadPosix::SetAffinity(const int* , const unsigned int)
 {
     return false;
 }
 #endif

 void ThreadPosix::SetNotAlive()
 {
     CriticalSectionScoped cs(_crit_state);
     _alive = false;
 }

 bool ThreadPosix::Shutdown()
 {
 #if !defined(WEBRTC_ANDROID)
     if (_thread && (0 != pthread_cancel(_thread)))
     {
         return false;
     }

     return true;
 #else
     return false;
 #endif
 }

 bool ThreadPosix::Stop()
 {
     bool dead = false;
     {
         CriticalSectionScoped cs(_crit_state);
         _alive = false;
         dead = _dead;
     }

     // TODO (hellner) why not use an event here?
     // Wait up to 10 seconds for the thread to terminate
     for (int i = 0; i < 1000 && !dead; i++)
     {
         timespec t;
         t.tv_sec = 0;
         t.tv_nsec = 10*1000*1000;
         nanosleep(&t, NULL);
         {
             CriticalSectionScoped cs(_crit_state);
             dead = _dead;
         }
     }
     if (dead)
     {
         return true;
     }
     else
     {
         return false;
     }
 }

 void ThreadPosix::Run()
 {
     {
         CriticalSectionScoped cs(_crit_state);
         _alive = true;
         _dead  = false;
     }
 #if (defined(WEBRTC_LINUX) || defined(WEBRTC_ANDROID))
     _pid = GetThreadId();
 #endif
     // The event the Start() is waiting for.
     _event->Set();

     if (_setThreadName)
     {
 #ifdef WEBRTC_LINUX
         prctl(PR_SET_NAME, (unsigned long)_name, 0, 0, 0);
 #endif
         WEBRTC_TRACE(kTraceStateInfo, kTraceUtility, -1,
                      "Thread with name:%s started ", _name);
     } else
     {
         WEBRTC_TRACE(kTraceStateInfo, kTraceUtility, -1,
                      "Thread without name started");
     }
     bool alive = true;
     do
     {
         if (_runFunction)
         {
             if (!_runFunction(_obj))
             {
                 alive = false;
             }
         }
         else
         {
             alive = false;
         }
         {
             CriticalSectionScoped cs(_crit_state);
             if (!alive) {
               _alive = false;
             }
             alive = _alive;
         }
     }
     while (alive);

     if (_setThreadName)
     {
         // Don't set the name for the trace thread because it may cause a
         // deadlock. TODO (hellner) there should be a better solution than
         // coupling the thread and the trace class like this.
         if (strcmp(_name, "Trace"))
         {
             WEBRTC_TRACE(kTraceStateInfo, kTraceUtility, -1,
                          "Thread with name:%s stopped", _name);
         }
     }
     else
     {
         WEBRTC_TRACE(kTraceStateInfo, kTraceUtility, -1,
                      "Thread without name stopped");
     }
     {
         CriticalSectionScoped cs(_crit_state);
         _dead = true;
     }
 }
 } // namespace webrtc
	/*
	* Copyright (c) 2012 The WebRTC project authors. All Rights Reserved.
	*
	* Use of this source code is governed by a BSD-style license
	* that can be found in the LICENSE file in the root of the source
	* tree. An additional intellectual property rights grant can be found
	* in the file PATENTS. All contributing project authors may
	* be found in the AUTHORS file in the root of the source tree.
	*/

	// The state of a thread is controlled by the two member variables
	// _alive and _dead.
	// _alive represents the state the thread has been ordered to achieve.
	// It is set to true by the thread at startup, and is set to false by
	// other threads, using SetNotAlive() and Stop().
	// _dead represents the state the thread has achieved.
	// It is written by the thread encapsulated by this class only
	// (except at init). It is read only by the Stop() method.
	// The Run() method fires _event when it's started; this ensures that the
	// Start() method does not continue until after _dead is false.
	// This protects against premature Stop() calls from the creator thread, but
	// not from other threads.

	// Their transitions and states:
	// _alive _dead Set by
	// false true Constructor
	// true false Run() method entry
	// false any Run() method runFunction failure
	// any false Run() method exit (happens only with _alive false)
	// false any SetNotAlive
	// false any Stop Stop waits for _dead to become true.
	//
	// Summarized a different way:
	// Variable Writer Reader
	// _alive Constructor(false) Run.loop
	// Run.start(true)
	// Run.fail(false)
	// SetNotAlive(false)
	// Stop(false)
	//
	// _dead Constructor(true) Stop.loop
	// Run.start(false)
	// Run.exit(true)

	#include "thread_posix.h"

	#include <algorithm>

	#include <assert.h>
	#include <errno.h>
	#include <string.h> // strncpy
	#include <time.h> // nanosleep
	#include <unistd.h>
	#ifdef WEBRTC_LINUX
	#include <sys/types.h>
	#include <sched.h>
	#include <sys/syscall.h>
	#include <linux/unistd.h>
	#include <sys/prctl.h>
	#endif

	#if defined(WEBRTC_MAC)
	#include <mach/mach.h>
	#endif

	#include "system_wrappers/interface/critical_section_wrapper.h"
	#include "system_wrappers/interface/event_wrapper.h"
	#include "system_wrappers/interface/trace.h"

	namespace webrtc {

	int ConvertToSystemPriority(ThreadPriority priority, int minPrio, int maxPrio)
	{
	assert(maxPrio - minPrio > 2);
	const int topPrio = maxPrio - 1;
	const int lowPrio = minPrio + 1;

	switch (priority)
	{
	case kLowPriority:
	return lowPrio;
	case kNormalPriority:
	// The -1 ensures that the kHighPriority is always greater or equal to
	// kNormalPriority.
	return (lowPrio + topPrio - 1) / 2;
	case kHighPriority:
	return std::max(topPrio - 2, lowPrio);
	case kHighestPriority:
	return std::max(topPrio - 1, lowPrio);
	case kRealtimePriority:
	return topPrio;
	}
	assert(false);
	return lowPrio;
	}

	extern "C"
	{
	static void* StartThread(void* lpParameter)
	{
	static_cast<ThreadPosix*>(lpParameter)->Run();
	return 0;
	}
	}

	ThreadWrapper* ThreadPosix::Create(ThreadRunFunction func, ThreadObj obj,
	ThreadPriority prio, const char* threadName)
	{
	ThreadPosix* ptr = new ThreadPosix(func, obj, prio, threadName);
	if (!ptr)
	{
	return NULL;
	}
	const int error = ptr->Construct();
	if (error)
	{
	delete ptr;
	return NULL;
	}
	return ptr;
	}

	ThreadPosix::ThreadPosix(ThreadRunFunction func, ThreadObj obj,
	ThreadPriority prio, const char* threadName)
	: _runFunction(func),
	_obj(obj),
	_crit_state(CriticalSectionWrapper::CreateCriticalSection()),
	_alive(false),
	_dead(true),
	_prio(prio),
	_event(EventWrapper::Create()),
	_name(),
	_setThreadName(false),
	#if (defined(WEBRTC_LINUX) \|\| defined(WEBRTC_ANDROID))
	_pid(-1),
	#endif
	_attr(),
	_thread(0)
	{
	if (threadName != NULL)
	{
	_setThreadName = true;
	strncpy(_name, threadName, kThreadMaxNameLength);
	_name[kThreadMaxNameLength - 1] = '\0';
	}
	}

	uint32_t ThreadWrapper::GetThreadId() {
	#if defined(WEBRTC_ANDROID) \|\| defined(WEBRTC_LINUX)
	return static_cast<uint32_t>(syscall(__NR_gettid));
	#elif defined(WEBRTC_MAC)
	return static_cast<uint32_t>(mach_thread_self());
	#else
	return reinterpret_cast<uint32_t>(pthread_self());
	#endif
	}

	int ThreadPosix::Construct()
	{
	int result = 0;
	#if !defined(WEBRTC_ANDROID)
	// Enable immediate cancellation if requested, see Shutdown()
	result = pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
	if (result != 0)
	{
	return -1;
	}
	result = pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
	if (result != 0)
	{
	return -1;
	}
	#endif
	result = pthread_attr_init(&_attr);
	if (result != 0)
	{
	return -1;
	}
	return 0;
	}

	ThreadPosix::~ThreadPosix()
	{
	pthread_attr_destroy(&_attr);
	delete _event;
	delete _crit_state;
	}

	#define HAS_THREAD_ID !defined(WEBRTC_IOS) && !defined(WEBRTC_MAC)
	#if HAS_THREAD_ID
	bool ThreadPosix::Start(unsigned int& threadID)
	#else
	bool ThreadPosix::Start(unsigned int& /threadID/)
	#endif
	{
	if (!_runFunction)
	{
	return false;
	}
	int result = pthread_attr_setdetachstate(&_attr, PTHREAD_CREATE_DETACHED);
	// Set the stack stack size to 1M.
	result \|= pthread_attr_setstacksize(&_attr, 1024*1024);
	#ifdef WEBRTC_THREAD_RR
	const int policy = SCHED_RR;
	#else
	const int policy = SCHED_FIFO;
	#endif
	_event->Reset();
	// If pthread_create was successful, a thread was created and is running.
	// Don't return false if it was successful since if there are any other
	// failures the state will be: thread was started but not configured as
	// asked for. However, the caller of this API will assume that a false
	// return value means that the thread never started.
	result \|= pthread_create(&_thread, &_attr, &StartThread, this);
	if (result != 0)
	{
	return false;
	}

	// Wait up to 10 seconds for the OS to call the callback function. Prevents
	// race condition if Stop() is called too quickly after start.
	if (kEventSignaled != _event->Wait(WEBRTC_EVENT_10_SEC))
	{
	WEBRTC_TRACE(kTraceError, kTraceUtility, -1,
	"posix thread event never triggered");
	// Timed out. Something went wrong.
	_runFunction = NULL;
	return true;
	}

	#if HAS_THREAD_ID
	threadID = static_cast<unsigned int>(_thread);
	#endif
	sched_param param;

	const int minPrio = sched_get_priority_min(policy);
	const int maxPrio = sched_get_priority_max(policy);
	if ((minPrio == EINVAL) \|\| (maxPrio == EINVAL))
	{
	WEBRTC_TRACE(kTraceError, kTraceUtility, -1,
	"unable to retreive min or max priority for threads");
	return true;
	}
	if (maxPrio - minPrio <= 2)
	{
	// There is no room for setting priorities with any granularity.
	return true;
	}
	param.sched_priority = ConvertToSystemPriority(_prio, minPrio, maxPrio);
	result = pthread_setschedparam(_thread, policy, &param);
	if (result == EINVAL)
	{
	WEBRTC_TRACE(kTraceError, kTraceUtility, -1,
	"unable to set thread priority");
	}
	return true;
	}

	// CPU_ZERO and CPU_SET are not available in NDK r7, so disable
	// SetAffinity on Android for now.
	#if (defined(WEBRTC_LINUX) && (!defined(WEBRTC_ANDROID)))
	bool ThreadPosix::SetAffinity(const int* processorNumbers,
	const unsigned int amountOfProcessors) {
	if (!processorNumbers \|\| (amountOfProcessors == 0)) {
	return false;
	}
	cpu_set_t mask;
	CPU_ZERO(&mask);

	for (unsigned int processor = 0;
	processor < amountOfProcessors;
	processor++) {
	CPU_SET(processorNumbers[processor], &mask);
	}
	#if defined(WEBRTC_ANDROID)
	// Android.
	const int result = syscall(__NR_sched_setaffinity,
	_pid,
	sizeof(mask),
	&mask);
	#else
	// "Normal" Linux.
	const int result = sched_setaffinity(_pid,
	sizeof(mask),
	&mask);
	#endif
	if (result != 0) {
	return false;
	}
	return true;
	}

	#else
	// NOTE: On Mac OS X, use the Thread affinity API in
	// /usr/include/mach/thread_policy.h: thread_policy_set and mach_thread_self()
	// instead of Linux gettid() syscall.
	bool ThreadPosix::SetAffinity(const int* , const unsigned int)
	{
	return false;
	}
	#endif

	void ThreadPosix::SetNotAlive()
	{
	CriticalSectionScoped cs(_crit_state);
	_alive = false;
	}

	bool ThreadPosix::Shutdown()
	{
	#if !defined(WEBRTC_ANDROID)
	if (_thread && (0 != pthread_cancel(_thread)))
	{
	return false;
	}

	return true;
	#else
	return false;
	#endif
	}

	bool ThreadPosix::Stop()
	{
	bool dead = false;
	{
	CriticalSectionScoped cs(_crit_state);
	_alive = false;
	dead = _dead;
	}

	// TODO (hellner) why not use an event here?
	// Wait up to 10 seconds for the thread to terminate
	for (int i = 0; i < 1000 && !dead; i++)
	{
	timespec t;
	t.tv_sec = 0;
	t.tv_nsec = 1010001000;
	nanosleep(&t, NULL);
	{
	CriticalSectionScoped cs(_crit_state);
	dead = _dead;
	}
	}
	if (dead)
	{
	return true;
	}
	else
	{
	return false;
	}
	}

	void ThreadPosix::Run()
	{
	{
	CriticalSectionScoped cs(_crit_state);
	_alive = true;
	_dead = false;
	}
	#if (defined(WEBRTC_LINUX) \|\| defined(WEBRTC_ANDROID))
	_pid = GetThreadId();
	#endif
	// The event the Start() is waiting for.
	_event->Set();

	if (_setThreadName)
	{
	#ifdef WEBRTC_LINUX
	prctl(PR_SET_NAME, (unsigned long)_name, 0, 0, 0);
	#endif
	WEBRTC_TRACE(kTraceStateInfo, kTraceUtility, -1,
	"Thread with name:%s started ", _name);
	} else
	{
	WEBRTC_TRACE(kTraceStateInfo, kTraceUtility, -1,
	"Thread without name started");
	}
	bool alive = true;
	do
	{
	if (_runFunction)
	{
	if (!_runFunction(_obj))
	{
	alive = false;
	}
	}
	else
	{
	alive = false;
	}
	{
	CriticalSectionScoped cs(_crit_state);
	if (!alive) {
	_alive = false;
	}
	alive = _alive;
	}
	}
	while (alive);

	if (_setThreadName)
	{
	// Don't set the name for the trace thread because it may cause a
	// deadlock. TODO (hellner) there should be a better solution than
	// coupling the thread and the trace class like this.
	if (strcmp(_name, "Trace"))
	{
	WEBRTC_TRACE(kTraceStateInfo, kTraceUtility, -1,
	"Thread with name:%s stopped", _name);
	}
	}
	else
	{
	WEBRTC_TRACE(kTraceStateInfo, kTraceUtility, -1,
	"Thread without name stopped");
	}
	{
	CriticalSectionScoped cs(_crit_state);
	_dead = true;
	}
	}
	} // namespace webrtc