Reimplement SGThread and friends.

Reimplement the threading stuff using either pthreads
or win32 threads. These simple classes should help us
to stay osg independent for simgears core classes.
This commit is contained in:
Mathias Froehlich
2011-09-07 17:32:43 +02:00
parent 39731debdc
commit 059db643bc
3 changed files with 427 additions and 277 deletions
+8
View File
@@ -1487,6 +1487,14 @@
RelativePath="..\..\simgear\threads\SGQueue.hxx" RelativePath="..\..\simgear\threads\SGQueue.hxx"
> >
</File> </File>
<File
RelativePath="..\..\simgear\threads\SGThread.hxx"
>
</File>
<File
RelativePath="..\..\simgear\threads\SGThread.cxx"
>
</File>
</Filter> </Filter>
<Filter <Filter
Name="Lib_sgstructure" Name="Lib_sgstructure"
+391 -88
View File
@@ -1,107 +1,410 @@
#include <simgear/compiler.h> // SGThread - Simple pthread class wrappers.
//
// Written by Bernie Bright, started April 2001.
//
// Copyright (C) 2001 Bernard Bright - bbright@bigpond.net.au
// Copyright (C) 2011 Mathias Froehlich
//
// This program is free software; you can redistribute it and/or
// modify it under the terms of the GNU General Public License as
// published by the Free Software Foundation; either version 2 of the
// License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful, but
// WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
// General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
//
#if defined(_MSC_VER) || defined(__MINGW32__) #ifdef HAVE_CONFIG_H
# include <time.h> # include <simgear_config.h>
#else
# if defined ( sgi ) && !defined( __GNUC__ )
// This works around a bug triggered when using MipsPro 7.4.1
// and (at least) IRIX 6.5.20
# include <iostream>
# endif
# include <sys/time.h>
#endif
#if _MSC_VER >= 1300
# include <winsock2.h>
#endif #endif
#include <simgear/compiler.h>
#include "SGThread.hxx" #include "SGThread.hxx"
void* #ifdef _WIN32
start_handler( void* arg )
/////////////////////////////////////////////////////////////////////////////
/// win32 threads
/////////////////////////////////////////////////////////////////////////////
#include <list>
#include <windows.h>
struct SGThread::PrivateData {
PrivateData() :
_handle(INVALID_HANDLE_VALUE)
{
}
~PrivateData()
{
if (_handle == INVALID_HANDLE_VALUE)
return;
CloseHandle(_handle);
_handle = INVALID_HANDLE_VALUE;
}
static DWORD WINAPI start_routine(LPVOID data)
{
SGThread* thread = reinterpret_cast<SGThread*>(data);
thread->run();
return 0;
}
bool start(SGThread& thread)
{
if (_handle != INVALID_HANDLE_VALUE)
return false;
_handle = CreateThread(0, 0, start_routine, &thread, 0, 0);
if (_handle == INVALID_HANDLE_VALUE)
return false;
return true;
}
void join()
{
if (_handle == INVALID_HANDLE_VALUE)
return;
DWORD ret = WaitForSingleObject(_handle, 0);
if (ret != WAIT_OBJECT_0)
return;
CloseHandle(_handle);
_handle = INVALID_HANDLE_VALUE;
}
HANDLE _handle;
};
struct SGMutex::PrivateData {
PrivateData()
{
InitializeCriticalSection((LPCRITICAL_SECTION)&_criticalSection);
}
~PrivateData()
{
DeleteCriticalSection((LPCRITICAL_SECTION)&_criticalSection);
}
void lock(void)
{
EnterCriticalSection((LPCRITICAL_SECTION)&_criticalSection);
}
void unlock(void)
{
LeaveCriticalSection((LPCRITICAL_SECTION)&_criticalSection);
}
CRITICAL_SECTION _criticalSection;
};
struct SGWaitCondition::PrivateData {
~PrivateData(void)
{
// The waiters list should be empty anyway
_mutex.lock();
while (!_pool.empty()) {
CloseHandle(_pool.front());
_pool.pop_front();
}
_mutex.unlock();
}
void signal(void)
{
_mutex.lock();
if (!_waiters.empty())
SetEvent(_waiters.back());
_mutex.unlock();
}
void broadcast(void)
{
_mutex.lock();
for (std::list<HANDLE>::iterator i = _waiters.begin(); i != _waiters.end(); ++i)
SetEvent(*i);
_mutex.unlock();
}
bool wait(SGMutex::PrivateData& externalMutex, DWORD msec)
{
_mutex.lock();
if (_pool.empty())
_waiters.push_front(CreateEvent(NULL, FALSE, FALSE, NULL));
else
_waiters.splice(_waiters.begin(), _pool, _pool.begin());
std::list<HANDLE>::iterator i = _waiters.begin();
_mutex.unlock();
externalMutex.unlock();
DWORD result = WaitForSingleObject(*i, msec);
externalMutex.lock();
_mutex.lock();
if (result != WAIT_OBJECT_0)
result = WaitForSingleObject(*i, 0);
_pool.splice(_pool.begin(), _waiters, i);
_mutex.unlock();
return result == WAIT_OBJECT_0;
}
void wait(SGMutex::PrivateData& externalMutex)
{
wait(externalMutex, INFINITE);
}
// Protect the list of waiters
SGMutex::PrivateData _mutex;
std::list<HANDLE> _waiters;
std::list<HANDLE> _pool;
};
#else
/////////////////////////////////////////////////////////////////////////////
/// posix threads
/////////////////////////////////////////////////////////////////////////////
#include <pthread.h>
#include <cassert>
#include <cerrno>
#include <sys/time.h>
struct SGThread::PrivateData {
PrivateData() :
_started(false)
{
}
~PrivateData()
{
// If we are still having a started thread and nobody waited,
// now detach ...
if (!_started)
return;
pthread_detach(_thread);
}
static void *start_routine(void* data)
{
SGThread* thread = reinterpret_cast<SGThread*>(data);
thread->run();
return 0;
}
bool start(SGThread& thread)
{
if (_started)
return false;
int ret = pthread_create(&_thread, 0, start_routine, &thread);
if (0 != ret)
return false;
_started = true;
return true;
}
void join()
{
if (!_started)
return;
pthread_join(_thread, 0);
_started = false;
}
pthread_t _thread;
bool _started;
};
struct SGMutex::PrivateData {
PrivateData()
{
int err = pthread_mutex_init(&_mutex, 0);
assert(err == 0);
(void)err;
}
~PrivateData()
{
int err = pthread_mutex_destroy(&_mutex);
assert(err == 0);
(void)err;
}
void lock(void)
{
int err = pthread_mutex_lock(&_mutex);
assert(err == 0);
(void)err;
}
void unlock(void)
{
int err = pthread_mutex_unlock(&_mutex);
assert(err == 0);
(void)err;
}
pthread_mutex_t _mutex;
};
struct SGWaitCondition::PrivateData {
PrivateData(void)
{
int err = pthread_cond_init(&_condition, NULL);
assert(err == 0);
(void)err;
}
~PrivateData(void)
{
int err = pthread_cond_destroy(&_condition);
assert(err == 0);
(void)err;
}
void signal(void)
{
int err = pthread_cond_signal(&_condition);
assert(err == 0);
(void)err;
}
void broadcast(void)
{
int err = pthread_cond_broadcast(&_condition);
assert(err == 0);
(void)err;
}
void wait(SGMutex::PrivateData& mutex)
{
int err = pthread_cond_wait(&_condition, &mutex._mutex);
assert(err == 0);
(void)err;
}
bool wait(SGMutex::PrivateData& mutex, unsigned msec)
{
struct timespec ts;
#ifdef HAVE_CLOCK_GETTIME
if (0 != clock_gettime(CLOCK_REALTIME, &ts))
return false;
#else
struct timeval tv;
if (0 != gettimeofday(&tv, NULL))
return false;
ts.tv_sec = tv.tv_sec;
ts.tv_nsec = tv.tv_usec * 1000;
#endif
ts.tv_nsec += 1000000*(msec % 1000);
if (1000000000 <= ts.tv_nsec) {
ts.tv_nsec -= 1000000000;
ts.tv_sec += 1;
}
ts.tv_sec += msec / 1000;
int evalue = pthread_cond_timedwait(&_condition, &mutex._mutex, &ts);
if (evalue == 0)
return true;
assert(evalue == ETIMEDOUT);
return false;
}
pthread_cond_t _condition;
};
#endif
SGThread::SGThread() :
_privateData(new PrivateData)
{ {
SGThread* thr = static_cast<SGThread*>(arg); }
thr->run();
return 0; SGThread::~SGThread()
{
delete _privateData;
_privateData = 0;
}
bool
SGThread::start()
{
return _privateData->start(*this);
} }
void void
SGThread::set_cancel( cancel_t mode ) SGThread::join()
{ {
switch (mode) _privateData->join();
{ }
case CANCEL_DISABLE:
pthread_setcancelstate( PTHREAD_CANCEL_DISABLE, 0 ); SGMutex::SGMutex() :
break; _privateData(new PrivateData)
case CANCEL_DEFERRED: {
pthread_setcanceltype( PTHREAD_CANCEL_DEFERRED, 0 ); }
pthread_setcancelstate( PTHREAD_CANCEL_ENABLE, 0 );
break; SGMutex::~SGMutex()
case CANCEL_IMMEDIATE: {
pthread_setcanceltype( PTHREAD_CANCEL_ASYNCHRONOUS, 0 ); delete _privateData;
pthread_setcancelstate( PTHREAD_CANCEL_ENABLE, 0 ); _privateData = 0;
break; }
default:
break; void
} SGMutex::lock()
{
_privateData->lock();
}
void
SGMutex::unlock()
{
_privateData->unlock();
}
SGWaitCondition::SGWaitCondition() :
_privateData(new PrivateData)
{
}
SGWaitCondition::~SGWaitCondition()
{
delete _privateData;
_privateData = 0;
}
void
SGWaitCondition::wait(SGMutex& mutex)
{
_privateData->wait(*mutex._privateData);
} }
bool bool
SGMutex::trylock() SGWaitCondition::wait(SGMutex& mutex, unsigned msec)
{ {
int status = pthread_mutex_lock( &mutex ); return _privateData->wait(*mutex._privateData, msec);
if (status == EBUSY)
{
return false;
}
assert( status == 0 );
return true;
} }
#if defined(_MSC_VER) || defined(__MINGW32__) void
int gettimeofday(struct timeval* tp, void* tzp) { SGWaitCondition::signal()
LARGE_INTEGER t;
if(QueryPerformanceCounter(&t)) {
/* hardware supports a performance counter */
LARGE_INTEGER f;
QueryPerformanceFrequency(&f);
tp->tv_sec = t.QuadPart/f.QuadPart;
tp->tv_usec = ((float)t.QuadPart/f.QuadPart*1000*1000)
- (tp->tv_sec*1000*1000);
} else {
/* hardware doesn't support a performance counter, so get the
time in a more traditional way. */
DWORD t;
t = timeGetTime();
tp->tv_sec = t / 1000;
tp->tv_usec = t % 1000;
}
/* 0 indicates that the call succeeded. */
return 0;
}
#endif
bool
SGPthreadCond::wait( SGMutex& mutex, unsigned long ms )
{ {
struct timeval now; _privateData->signal();
::gettimeofday( &now, 0 );
// Wait time is now + ms milliseconds
unsigned int sec = ms / 1000;
unsigned int nsec = (ms % 1000) * 1000;
struct timespec abstime;
abstime.tv_sec = now.tv_sec + sec;
abstime.tv_nsec = now.tv_usec*1000 + nsec;
int status = pthread_cond_timedwait( &cond, &mutex.mutex, &abstime );
if (status == ETIMEDOUT)
{
return false;
}
assert( status == 0 );
return true;
} }
void
SGWaitCondition::broadcast()
{
_privateData->broadcast();
}
+25 -186
View File
@@ -3,6 +3,7 @@
// Written by Bernie Bright, started April 2001. // Written by Bernie Bright, started April 2001.
// //
// Copyright (C) 2001 Bernard Bright - bbright@bigpond.net.au // Copyright (C) 2001 Bernard Bright - bbright@bigpond.net.au
// Copyright (C) 2011 Mathias Froehlich
// //
// This program is free software; you can redistribute it and/or // This program is free software; you can redistribute it and/or
// modify it under the terms of the GNU General Public License as // modify it under the terms of the GNU General Public License as
@@ -18,41 +19,18 @@
// along with this program; if not, write to the Free Software // along with this program; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. // Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
// //
// $Id$
#ifndef SGTHREAD_HXX_INCLUDED #ifndef SGTHREAD_HXX_INCLUDED
#define SGTHREAD_HXX_INCLUDED 1 #define SGTHREAD_HXX_INCLUDED 1
#include <simgear/compiler.h> #include <simgear/compiler.h>
#include <pthread.h>
#include <cassert>
#include <cerrno>
class SGThread;
extern "C" {
void* start_handler( void* );
};
/** /**
* Encapsulate generic threading methods. * Encapsulate generic threading methods.
* Users derive a class from SGThread and implement the run() member function. * Users derive a class from SGThread and implement the run() member function.
*/ */
class SGThread class SGThread {
{
public: public:
/**
* SGThread cancelation modes.
*/
enum cancel_t
{
CANCEL_DISABLE = 0,
CANCEL_DEFERRED,
CANCEL_IMMEDIATE
};
public:
/** /**
* Create a new thread object. * Create a new thread object.
* When a SGThread object is created it does not begin execution * When a SGThread object is created it does not begin execution
@@ -62,18 +40,9 @@ public:
/** /**
* Start the underlying thread of execution. * Start the underlying thread of execution.
* @param cpu An optional parameter to specify on which CPU to run this
* thread (only supported on IRIX at this time).
* @return Pthread error code if execution fails, otherwise returns 0. * @return Pthread error code if execution fails, otherwise returns 0.
*/ */
int start( unsigned cpu = 0 ); bool start();
/**
* Sends a cancellation request to the underlying thread. The target
* thread will either ignore the request, honor it immediately or defer
* it until it reaches a cancellation point.
*/
void cancel();
/** /**
* Suspends the exection of the calling thread until this thread * Suspends the exection of the calling thread until this thread
@@ -89,82 +58,31 @@ protected:
*/ */
virtual ~SGThread(); virtual ~SGThread();
/**
* Set the threads cancellation mode.
* @param mode The required cancellation mode.
*/
void set_cancel( cancel_t mode );
/** /**
* All threads execute by deriving the run() method of SGThread. * All threads execute by deriving the run() method of SGThread.
* If this function terminates then the thread also terminates. * If this function terminates then the thread also terminates.
*/ */
virtual void run() = 0; virtual void run() = 0;
private:
/**
* Pthread thread identifier.
*/
pthread_t tid;
friend void* start_handler( void* );
private: private:
// Disable copying. // Disable copying.
SGThread( const SGThread& ); SGThread(const SGThread&);
SGThread& operator=( const SGThread& ); SGThread& operator=(const SGThread&);
struct PrivateData;
PrivateData* _privateData;
friend struct PrivateData;
}; };
inline class SGWaitCondition;
SGThread::SGThread()
{
}
inline
SGThread::~SGThread()
{
}
inline int
SGThread::start( unsigned cpu )
{
int status = pthread_create( &tid, 0, start_handler, this );
assert( status == 0 );
(void)status;
#if defined( sgi )
if ( !status && !cpu )
pthread_setrunon_np( cpu );
#endif
return status;
}
inline void
SGThread::join()
{
int status = pthread_join( tid, 0 );
assert( status == 0 );
(void)status;
}
inline void
SGThread::cancel()
{
int status = pthread_cancel( tid );
assert( status == 0 );
(void)status;
}
/** /**
* A mutex is used to protect a section of code such that at any time * A mutex is used to protect a section of code such that at any time
* only a single thread can execute the code. * only a single thread can execute the code.
*/ */
class SGMutex class SGMutex {
{
friend class SGPthreadCond;
public: public:
/** /**
* Create a new mutex. * Create a new mutex.
* Under Linux this is a 'fast' mutex. * Under Linux this is a 'fast' mutex.
@@ -186,86 +104,46 @@ public:
* mutex is already locked and owned by the calling thread, the calling * mutex is already locked and owned by the calling thread, the calling
* thread is suspended until the mutex is unlocked, effectively causing * thread is suspended until the mutex is unlocked, effectively causing
* the calling thread to deadlock. * the calling thread to deadlock.
*
* @see SGMutex::trylock
*/ */
void lock(); void lock();
/**
* Try to lock the mutex for the current thread. Behaves like lock except
* that it doesn't block the calling thread.
* @return true if mutex was successfully locked, otherwise false.
* @see SGMutex::lock
*/
bool trylock();
/** /**
* Unlock this mutex. * Unlock this mutex.
* It is assumed that the mutex is locked and owned by the calling thread. * It is assumed that the mutex is locked and owned by the calling thread.
*/ */
void unlock(); void unlock();
protected: private:
struct PrivateData;
PrivateData* _privateData;
/** friend class SGWaitCondition;
* Pthread mutex.
*/
pthread_mutex_t mutex;
}; };
inline SGMutex::SGMutex()
{
int status = pthread_mutex_init( &mutex, 0 );
assert( status == 0 );
(void)status;
}
inline SGMutex::~SGMutex()
{
int status = pthread_mutex_destroy( &mutex );
assert( status == 0 );
(void)status;
}
inline void SGMutex::lock()
{
int status = pthread_mutex_lock( &mutex );
assert( status == 0 );
(void)status;
}
inline void SGMutex::unlock()
{
int status = pthread_mutex_unlock( &mutex );
assert( status == 0 );
(void)status;
}
/** /**
* A condition variable is a synchronization device that allows threads to * A condition variable is a synchronization device that allows threads to
* suspend execution until some predicate on shared data is satisfied. * suspend execution until some predicate on shared data is satisfied.
* A condition variable is always associated with a mutex to avoid race * A condition variable is always associated with a mutex to avoid race
* conditions. * conditions.
*/ */
class SGPthreadCond class SGWaitCondition {
{
public: public:
/** /**
* Create a new condition variable. * Create a new condition variable.
*/ */
SGPthreadCond(); SGWaitCondition();
/** /**
* Destroy the condition object. * Destroy the condition object.
*/ */
~SGPthreadCond(); ~SGWaitCondition();
/** /**
* Wait for this condition variable to be signaled. * Wait for this condition variable to be signaled.
* *
* @param SGMutex& reference to a locked mutex. * @param SGMutex& reference to a locked mutex.
*/ */
void wait( SGMutex& ); void wait(SGMutex&);
/** /**
* Wait for this condition variable to be signaled for at most * Wait for this condition variable to be signaled for at most
@@ -276,7 +154,7 @@ public:
* *
* @return * @return
*/ */
bool wait( SGMutex& mutex, unsigned long ms ); bool wait(SGMutex& mutex, unsigned msec);
/** /**
* Wake one thread waiting on this condition variable. * Wake one thread waiting on this condition variable.
@@ -294,50 +172,11 @@ public:
private: private:
// Disable copying. // Disable copying.
SGPthreadCond(const SGPthreadCond& ); SGWaitCondition(const SGWaitCondition&);
SGPthreadCond& operator=(const SGPthreadCond& ); SGWaitCondition& operator=(const SGWaitCondition&);
private: struct PrivateData;
PrivateData* _privateData;
/**
* The Pthread conditon variable.
*/
pthread_cond_t cond;
}; };
inline SGPthreadCond::SGPthreadCond()
{
int status = pthread_cond_init( &cond, 0 );
assert( status == 0 );
(void)status;
}
inline SGPthreadCond::~SGPthreadCond()
{
int status = pthread_cond_destroy( &cond );
assert( status == 0 );
(void)status;
}
inline void SGPthreadCond::signal()
{
int status = pthread_cond_signal( &cond );
assert( status == 0 );
(void)status;
}
inline void SGPthreadCond::broadcast()
{
int status = pthread_cond_broadcast( &cond );
assert( status == 0 );
(void)status;
}
inline void SGPthreadCond::wait( SGMutex& mutex )
{
int status = pthread_cond_wait( &cond, &mutex.mutex );
assert( status == 0 );
(void)status;
}
#endif /* SGTHREAD_HXX_INCLUDED */ #endif /* SGTHREAD_HXX_INCLUDED */