Thread.cpp

Go to the documentation of this file.

#include <dcl/Config.h>
 
#ifdef __WINNT__
    #include <windows.h>
    #include <VersionHelpers.h> // IsWindowsVistaOrGreater
#else
    #include <sys/ioctl.h>
    #include <sys/poll.h>
    #include <sys/time.h>
    #include <unistd.h>
#endif
#ifdef __sun__
    #include <sys/filio.h>      // FIONREAD
#endif
 
#if __DCL_PTHREAD
    #include <pthread.h>
    #include <sched.h>      // sched_yield
#endif
 
#include <dcl/Exception.h>
#include <dcl/Thread.h>
 
#if __DCL_HAVE_THIS_FILE__
#undef __THIS_FILE__
static const char_t __THIS_FILE__[] = __T("dcl/Thread.cpp");
#endif
 
__DCL_BEGIN_NAMESPACE
 
IMPLEMENT_CLASSINFO(Thread, Object)
 
String Thread::toString() const
{
    StringBuilder r = className() + __T(":");
    if (__name.isEmpty())
        r += __T("(noname)");
    return r;
}
 
Thread::Thread(const char_t* _name /*= NULL*/)
{
#ifdef __WINNT__
     __hThread = NULL;
#endif
 
     __threadId = 0;
    if (_name)
        __name = _name;
}
 
#if __DCL_PTHREAD
    #define __key_t                             pthread_key_t
    #define __key_create(key)                   pthread_key_create(&key, NULL)
    #define __key_delete(key)                   pthread_key_delete(key)
    #define __key_get(key)                      pthread_getspecific(key)
    #define __key_set(key, value)               pthread_setspecific(key, value)
    #define __key_set_success(key, value)   pthread_setspecific(key, value) == 0
#elif defined(__WINNT__)
    #define __key_t                             DWORD
    #define __key_create(key)                   key = TlsAlloc()
    #define __key_delete(key)                   TlsFree(key)
    #define __key_get(key)                      TlsGetValue(key)
    #define __key_set(key, value)               TlsSetValue(key, value)
    #define __key_set_success(key, value)       TlsSetValue(key, value) == TRUE
#endif
 
static __key_t __keySelfThread = (__key_t) -1;
 
void Thread::start()
    __DCL_THROWS1(SysError*)
{
    __DCL_ASSERT(__keySelfThread != (__key_t)-1);
 
#if __DCL_PTHREAD
    int n = pthread_create(
            &__threadId,
            (pthread_attr_t*)NULL,
            Thread::startRoutine, (void*)this
        );
    if (n != 0)
            throw new SysError(n);
#elif defined(__WINNT__)
    DWORD id = 0;
    __hThread = CreateThread(
        NULL,           //          __in    LPSECURITY_ATTRIBUTES lpThreadAttributes,
        0,              // default  __in    SIZE_T dwStackSize,
        startRoutine,   //          __in    LPTHREAD_START_ROUTINE lpStartAddress,
        (LPVOID) this,  //          __in    LPVOID lpParameter,
        0,              // run immediate    __in    DWORD dwCreationFlags,
        &id             //          __out   LPDWORD lpThreadId
    );
    __threadId = id;
    if (__hThread == NULL)
        throw new SysError(GetLastError());
#endif
}
 
bool Thread::init()
{
    return __key_set_success(__keySelfThread, this);
}
 
Thread* Thread::getSelfThread()
{
    __DCL_ASSERT(__keySelfThread != (__key_t)-1);
    return (Thread*) __key_get(__keySelfThread);
}
 
#if __DCL_PTHREAD
void*
#elif defined(__WINNT__)
DWORD
#endif
Thread::startRoutine(void* _pThread)
{
    int r = -1;
    Thread* pThread = (Thread*) _pThread;
    try {
        if (pThread->init())
            r = pThread->run();
    }
    catch (Exception* cause) {
        __DCL_TRACE2(__T("Thread execution fail! [%ls] %ls\n"),
            pThread->toString().data(), cause->toStringAll().data());
        cause->destroy();
    }
#if __DCL_PTHREAD
    return (void*)(size_t)r;
#elif defined(__WINNT__)
    return (DWORD)r;
#endif
}
 
int Thread::join()
{
#if __DCL_PTHREAD
    void* r;
    pthread_join(__threadId, &r);
#elif defined(__WINNT__)
    DWORD r = 0;
    WaitForSingleObject(__hThread, INFINITE);
    GetExitCodeThread(__hThread, &r);
    CloseHandle(__hThread);
    __hThread = NULL;
#endif
 
    __threadId = 0;
    return (int)(size_t)r;
}
 
void Thread::sleep(unsigned int _mills)
{
#if __DCL_PTHREAD
    /*
    ::sleep(_milliseconds / 1000);
    ::usleep((_milliseconds % 1000) * 1000);
    */
    poll(NULL, 0, _mills);
#elif defined(__WINNT__)
    Sleep(_mills);
#endif
}
 
void Thread::yield()
{
#if __DCL_PTHREAD
    // pthread_yield(); deprecated
    sched_yield();
#elif defined(__WINNT__)
    SwitchToThread();
#endif
}
 
thread_t Thread::self()
{
#if __DCL_PTHREAD
    return pthread_self();
#elif defined(__WINNT__)
    return GetCurrentThreadId();
#endif
}
 
void Thread::once(thread_once_t& _onceControl, initRoutine _initRoutine)
{
    __DCL_ASSERT_PARAM(_initRoutine != NULL);
#if __DCL_PTHREAD
    pthread_once(&_onceControl, _initRoutine);
#elif defined(__WINNT__)
    InitOnceExecuteOnce(&_onceControl, (PINIT_ONCE_FN)_initRoutine, NULL, NULL);
#endif
}
 
thread_key_t Thread::keyCreate() __DCL_THROWS1(SysError*)
{
#if __DCL_PTHREAD
    thread_key_t rKey;
    int failed = pthread_key_create(&rKey, NULL);
    if (failed)
        throw new SysError(failed);
    return rKey;
#elif defined(__WINNT__)
    thread_key_t rKey = TlsAlloc();
    if (rKey == TLS_OUT_OF_INDEXES)
        throw new SysError(GetLastError());
    return rKey;
#endif
}
 
void Thread::keyDelete(thread_key_t _key) __DCL_THROWS1(SysError*)
{
#if __DCL_PTHREAD
    int failed = pthread_key_delete(_key);
    if (failed)
        throw new SysError(failed);
#elif defined(__WINNT__)
    if (!TlsFree(_key))
        throw new SysError(GetLastError());
#endif
}
void Thread::keySetValue(thread_key_t _key, void* _value) __DCL_THROWS1(SysError*)
{
#if __DCL_PTHREAD
    int failed = pthread_setspecific(_key, _value);
    if (failed)
        throw new SysError(failed);
#elif defined(__WINNT__)
    if (!TlsSetValue(_key, _value))
        throw new SysError(GetLastError());
#endif
}
void* Thread::keyGetValue(thread_key_t _key) __DCL_THROWS1(SysError*)
{
#if __DCL_PTHREAD
    return pthread_getspecific(_key);
#elif defined(__WINNT__)
    void* p = TlsGetValue(_key);
    if (p == NULL && GetLastError() != ERROR_SUCCESS)
        throw new SysError(GetLastError());
    return p;
#endif
}
 
#if __DCL_PTHREAD
#if defined(__APPLE__)
    #define __lock_t            pthread_mutex_t
    #define __init(lock)        pthread_mutex_init(&lock, NULL)
    #define __destroy(lock)     pthread_mutex_destroy(&lock)
    #define __lock(lock)        pthread_mutex_lock(&lock)
    #define __unlock(lock)      pthread_mutex_unlock(&lock)
#else
    #define __lock_t            pthread_spinlock_t
    #define __init(lock)        pthread_spin_init(&lock, PTHREAD_PROCESS_PRIVATE)
    #define __destroy(lock)     pthread_spin_destroy(&lock)
    #define __lock(lock)        pthread_spin_lock(&lock)
    #define __unlock(lock)      pthread_spin_unlock(&lock)
#endif
#elif defined(__WINNT__)
    #define __lock_t            CRITICAL_SECTION
    #define __init(lock)        InitializeCriticalSectionAndSpinCount(&lock, INFINITE)
    #define __destroy(lock)     DeleteCriticalSection(&lock)
    #define __lock(lock)        EnterCriticalSection(&lock)
    #define __unlock(lock)      LeaveCriticalSection(&lock)
#endif
 
#define __LOCK_COUNT    7
static __lock_t __lock[__LOCK_COUNT];
 
void Thread::crtLock(const void* _p)
{
    __lock_t& lock = __lock[(size_t) _p % __LOCK_COUNT];
    __lock(lock);
}
 
void Thread::crtUnlock(const void* _p)
{
    __lock_t& lock = __lock[(size_t) _p % __LOCK_COUNT];
    __unlock(lock);
}
 
#if __DCL_PTHREAD
long Thread::incrementAndGet(volatile long& _n)
{
    __lock_t& lock = __lock[((size_t)&_n) % __LOCK_COUNT];
    __lock(lock);
    long n = ++(_n);
    __unlock(lock);
    return n;
}
 
long Thread::decrementAndGet(volatile long& _n)
{
    __lock_t& lock = __lock[((size_t)&_n) % __LOCK_COUNT];
    __lock(lock);
    long n = --(_n);
    __unlock(lock);
    return n;
}
 
long long Thread::incrementAndGet(volatile long long& _n)
{
    __lock_t& lock = __lock[((size_t)&_n) % __LOCK_COUNT];
    __lock(lock);
    long long n = ++(_n);
    __unlock(lock);
    return n;
}
 
long long Thread::decrementAndGet(volatile long long& _n)
{
    __lock_t& lock = __lock[((size_t)&_n) % __LOCK_COUNT];
    __lock(lock);
    long long n = --(_n);
    __unlock(lock);
    return n;
}
#elif defined(__WINNT__)
long Thread::incrementAndGet(volatile long& _n)
{
    return InterlockedIncrement(&_n);
}
 
long Thread::decrementAndGet(volatile long& _n)
{
    return InterlockedDecrement(&_n);
}
 
long long Thread::decrementAndGet(volatile long long& _n)
{
    return InterlockedDecrement64(&_n);
}
 
long long Thread::incrementAndGet(volatile long long& _n)
{
    return InterlockedIncrement64(&_n);
}
#endif
 
void __initializeThreadEnvironment()
{
    __key_create(__keySelfThread);
 
    for (int i = 0; i < __LOCK_COUNT; i++)
        __init(__lock[i]);
}
 
void __cleanupThreadEnvironment()
{
    __key_delete(__keySelfThread);
 
    for (int i = 0; i < __LOCK_COUNT; i++)
        __destroy(__lock[i]);
}
 
Thread::Event::Event()
{
    __bWaiting = false;
 
#if __DCL_PTHREAD
    __DCL_VERIFY(pipe(__fds) == 0);
#elif defined(__WINNT__)
    __DCL_VERIFY(__hEvent = CreateEvent(
            NULL,   // lpEventAttributes
            FALSE, // bManualReset
            FALSE, // bInitialState
            NULL    // lpName
        )
    );
#endif
}
 
Thread::Event::~Event()
{
#if __DCL_PTHREAD
    __DCL_VERIFY(close(__fds[0]) == 0);
    __DCL_VERIFY(close(__fds[1]) == 0);
#elif defined(__WINNT__)
    __DCL_VERIFY(CloseHandle(__hEvent) != FALSE);
#endif
}
 
void Thread::Event::set()
{
#if __DCL_PTHREAD
    char c = '1';
    __DCL_VERIFY(write(__fds[1], &c, sizeof(c)) == sizeof(c));
#elif defined(__WINNT__)
    __DCL_VERIFY(SetEvent(__hEvent) != FALSE);
#endif
}
 
void Thread::Event::reset()
{
#if __DCL_PTHREAD
    int nbytes = 0;
    ioctl(__fds[READ_FD_INDEX], FIONREAD, &nbytes);
    while (nbytes > 0) {
        char buf[100];
        int n = read(__fds[0], buf, nbytes < 100 ? nbytes : 100);
        if (n > 0)
            nbytes -= n;
        else {
            // 0: read all, -1:error
            break;
        }
    }
#elif defined(__WINNT__)
    __DCL_VERIFY(ResetEvent(__hEvent) != FALSE);
#endif
}
 
bool Thread::Event::wait(unsigned int _milliseconds)
{
    bool r = true;
    __bWaiting = true;
#if __DCL_PTHREAD
    struct pollfd pfd;
    pfd.fd = __fds[READ_FD_INDEX];
    pfd.events = POLLIN;
    pfd.revents = 0;
 
    int n = poll(&pfd, 1, _milliseconds);
    if (n <= 0) {
        r = false;
        __DCL_ASSERT(n == 0);
        errno = ETIMEDOUT;
    }
#elif defined(__WINNT__)
    switch(WaitForSingleObject(__hEvent, _milliseconds)) {
        case WAIT_OBJECT_0:
            break;
        case WAIT_TIMEOUT:
            r = false;
            SetLastError(ERROR_TIMEOUT);
            break;
        default :
            r = false;
            __DCL_ASSERT(false);
    }
 
#endif
    __bWaiting = false;
    return r;
}
 
#ifdef __WINNT__
Thread::Mutex::Mutex(DWORD _dwSpinCount)
{
    InitializeCriticalSectionAndSpinCount(&__cs, _dwSpinCount);
}
#endif
 
Thread::Mutex::Mutex()
{
    __locker = 0;
#if __DCL_PTHREAD
    __DCL_VERIFY(pthread_mutex_init(&__mutex, NULL) == 0);
#elif defined(__WINNT__)
    InitializeCriticalSection(&__cs);
#endif
}
 
Thread::Mutex::~Mutex()
{
#if __DCL_PTHREAD
    __DCL_VERIFY(pthread_mutex_destroy(&__mutex) == 0);
#elif defined(__WINNT__)
    DeleteCriticalSection(&__cs);
#endif
}
 
void Thread::Mutex::lock()
{
#if __DCL_PTHREAD
    __DCL_VERIFY(pthread_mutex_lock(&__mutex) == 0);
#elif defined(__WINNT__)
    EnterCriticalSection(&__cs);
#endif
    __locker = Thread::self();
}
 
bool Thread::Mutex::tryLock()
{
    bool r = false;
#if __DCL_PTHREAD
    int r_ = pthread_mutex_trylock(&__mutex);
    __DCL_ASSERT(r_ != EINVAL);
    __DCL_ASSERT(r_ == 0 || r_ == EBUSY);
    r = r_ == 0;
#elif defined(__WINNT__)
    r = TryEnterCriticalSection(&__cs) == TRUE;
#endif
    if (r) {
        __locker = Thread::self();
    }
    return r;
}
 
void Thread::Mutex::unlock()
{
    __DCL_ASSERT(__locker == Thread::self());
    __locker = 0;
#if __DCL_PTHREAD
    __DCL_VERIFY(pthread_mutex_unlock(&__mutex) == 0);
#elif defined(__WINNT__)
    LeaveCriticalSection(&__cs);
#endif
}
 
#if __DCL_PTHREAD && !defined(__APPLE__)
Thread::SpinLock::SpinLock(bool _pshared /* = false */)
{
    __locker = 0;
    __DCL_VERIFY(pthread_spin_init(&__spinLock,
        _pshared ? PTHREAD_PROCESS_SHARED : PTHREAD_PROCESS_PRIVATE
    ) == 0);
}
 
Thread::SpinLock::~SpinLock()
{
    __DCL_VERIFY(pthread_spin_destroy(&__spinLock) == 0);
}
 
void Thread::SpinLock::lock()
{
    __DCL_VERIFY(pthread_spin_lock(&__spinLock) == 0);
    __locker = Thread::self();
}
 
bool Thread::SpinLock::tryLock()
{
    int r_ = pthread_spin_trylock(&__spinLock);
    __DCL_ASSERT(r_ != EINVAL);
    __DCL_ASSERT(r_ == 0 || r_ == EBUSY);
    bool r = r_ == 0;
    if (r) {
        __locker = Thread::self();
    }
    return r;
}
 
void Thread::SpinLock::unlock()
{
    __DCL_ASSERT(__locker == Thread::self());
    __locker = 0;
    __DCL_VERIFY(pthread_spin_unlock(&__spinLock) == 0);
}
#endif      // __DCL_PTHREAD
 
#if __DCL_PTHREAD || (defined(__WINNT__) && _WIN32_WINNT >= 0x0600)
Thread::ReadWriteLock::ReadWriteLock()
{
#if __DCL_PTHREAD
    __DCL_VERIFY(pthread_rwlock_init(&__rwlock, NULL) == 0);
#else
    InitializeSRWLock(&__srwlock);
#endif
}
 
Thread::ReadWriteLock::~ReadWriteLock()
{
#if __DCL_PTHREAD
    __DCL_VERIFY(pthread_rwlock_destroy(&__rwlock) == 0);
#endif
}
 
void Thread::ReadWriteLock::readLock()
{
#if __DCL_PTHREAD
    __DCL_VERIFY(pthread_rwlock_rdlock(&__rwlock) == 0);
#else
    AcquireSRWLockShared(&__srwlock);
#endif
}
 
void Thread::ReadWriteLock::writeLock()
{
#if __DCL_PTHREAD
    __DCL_VERIFY(pthread_rwlock_wrlock(&__rwlock) == 0);
#else
    AcquireSRWLockExclusive(&__srwlock);
#endif
}
 
#if __DCL_PTHREAD
bool Thread::ReadWriteLock::tryReadLock()
{
    int r = pthread_rwlock_tryrdlock(&__rwlock);
    __DCL_ASSERT(r != EINVAL);
    __DCL_ASSERT(r == 0 || r == EBUSY);
    return r == 0;
}
 
bool Thread::ReadWriteLock::tryWriteLock()
{
    int r = pthread_rwlock_trywrlock(&__rwlock);
    __DCL_ASSERT(r != EINVAL);
    __DCL_ASSERT(r == 0 || r == EBUSY);
    return r == 0;
}
 
void Thread::ReadWriteLock::unlock()
{
    __DCL_VERIFY(::pthread_rwlock_unlock(&__rwlock) == 0);
}
#else
void Thread::ReadWriteLock::readUnlock()
{
    ReleaseSRWLockShared(&__srwlock);
}
void Thread::ReadWriteLock::writeUnlock()
{
    ReleaseSRWLockExclusive(&__srwlock);
}
#endif  // __DCL_PTHREAD        
#endif  // __DCL_PTHREAD || (defined(__WINNT__) && _WIN32_WINNT >= 0x0600)
 
Thread::CondVar::CondVar()
{
    __bWaiting = false;
#if __DCL_PTHREAD
    __DCL_VERIFY(pthread_cond_init(&__cond, NULL) == 0);
#elif defined(__WINNT__) && _WIN32_WINNT >= 0x0600
    // Windows Vista, Windows Server 2008
    __DCL_ASSERT(IsWindowsVistaOrGreater());
    InitializeConditionVariable(&__cond);
#endif
 
}
 
Thread::CondVar::~CondVar()
{
#if __DCL_PTHREAD
    __DCL_VERIFY(pthread_cond_destroy(&__cond) == 0);
#endif
}
 
void Thread::CondVar::signal()
{
#if __DCL_PTHREAD
    __DCL_VERIFY(pthread_cond_signal(&__cond) == 0);
#elif defined(__WINNT__)
    #if _WIN32_WINNT >= 0x0600
    WakeConditionVariable(&__cond);
    #else
    __cond.set();
    #endif
#endif
}
 
#if __DCL_PTHREAD || (defined(__WINNT__) && _WIN32_WINNT >= 0x0600)
void Thread::CondVar::broadcast()
{
#if __DCL_PTHREAD
    __DCL_VERIFY(pthread_cond_broadcast(&__cond) == 0);
#elif defined(__WINNT__)
    WakeAllConditionVariable(&__cond);
#endif
}
#endif
 
bool Thread::CondVar::wait(Mutex& _mutex, unsigned int _milliseconds)
{
    bool r = true;
    __bWaiting = true;
#if __DCL_PTHREAD
    // INFINITE == -1
    if (_milliseconds == (unsigned int) INFINITE) {
        __DCL_VERIFY(pthread_cond_wait(&__cond, (pthread_mutex_t*)&_mutex) == 0);
        r = true;
    }
    else {
        // millisoconds 1/1000
        // microseconds 1/1000000
        // nanoseconds  1/1000000000
        struct timeval tv;
        struct timezone tz;
        struct timespec ts;
        __DCL_VERIFY(gettimeofday(&tv, &tz) == 0);
        ts.tv_sec = tv.tv_sec + (_milliseconds / 1000);
        ts.tv_nsec = (tv.tv_usec * 1000) + ((_milliseconds % 1000) * 1000000);
        int n = pthread_cond_timedwait(&__cond, (pthread_mutex_t*)&_mutex, &ts);
        if (n != 0) {
            r = false;
            __DCL_ASSERT(n == ETIMEDOUT);
            errno = n;
        }
    }
#elif defined(__WINNT__)
    #if _WIN32_WINNT >= 0x0600
    if (!SleepConditionVariableCS(&__cond,
            (CRITICAL_SECTION*)&_mutex, _milliseconds)) {
        r = false;
        __DCL_ASSERT(GetLastError() == ERROR_TIMEOUT);
    }
    #else
    _mutex.unlock();
    r = __cond.wait(INFINITE);
    _mutex.lock();
    if (r)
        __cond.reset();
    #endif
#endif
 
    __bWaiting = false;
    return r;
}
 
#if defined(__WINNT__) && _WIN32_WINNT >= 0x0600
bool Thread::CondVar::wait(ReadWriteLock& _lock, bool _shared, unsigned int _milliseconds)
{
    bool r = true;
    __bWaiting = true;
    if (!SleepConditionVariableSRW(&__cond, (SRWLOCK *)&_lock, 
            _shared ? CONDITION_VARIABLE_LOCKMODE_SHARED : 0, _milliseconds)) {
        r = false;
        __DCL_ASSERT(GetLastError() == ERROR_TIMEOUT);
    }
    __bWaiting = false;
    return r;
}
#endif
 
#if __DCL_PTHREAD && !defined(__APPLE__)
Thread::Barrier::Barrier(unsigned int _count)
{
    __DCL_VERIFY(pthread_barrier_init(&__barrier, NULL, _count) == 0);
}
 
Thread::Barrier::~Barrier()
{
    __DCL_VERIFY(pthread_barrier_destroy(&__barrier) == 0);
}
 
void Thread::Barrier::wait()
{
    __DCL_VERIFY(pthread_barrier_wait(&__barrier) == 0);
}
 
#endif
 
__DCL_END_NAMESPACE