win32/Thread.hh

Go to the documentation of this file.

 
#ifndef CRL_MULTISENSE_THREAD_HH
#define CRL_MULTISENSE_THREAD_HH
 
#ifndef WIN32_LEAN_AND_MEAN
#define WIN32_LEAN_AND_MEAN 1
#endif
#include <windows.h>
 
#include <stdint.h>
#include <errno.h>
#include <string.h>
 
#include <vector>
#include <deque>
 
#include "MultiSense/details/utility/Portability.hh"
 
#include "utility/Exception.hh"
 
namespace crl {
namespace multisense {
namespace details {
namespace utility {
 
//
// Forward declarations.
 
class ScopedLock;
 
//
// A simple class to wrap pthread creation and joining
 
class Thread {
public:
 
    static CRL_CONSTEXPR uint32_t FLAGS_DETACH = (1 << 0);
 
    Thread(LPTHREAD_START_ROUTINE functionP,
           void    *contextP=NULL,
           uint32_t flags=0,
           int32_t  scheduler=-1,
           int32_t  priority=0) : m_flags(flags) {
 
        //
        // -1 means the user wants default scheduling behavior
 
        if (-1 != scheduler) {
 
            //
            // Set our scheduling policy
 
            if (scheduler != 0 /* SCHED_OTHER */)
                CRL_EXCEPTION("This platform only supports SCHED_OTHER");
 
            if (priority != 0)
                CRL_EXCEPTION("Priority can not be set at this time.");
        }
 
        //
        // Finally, create the thread
 
        m_threadHandle = CreateThread (NULL, 0, functionP, contextP, 0, &m_threadId);
        if (m_threadHandle == NULL)
            CRL_EXCEPTION("CreateThread() failed: %d", GetLastError());
 
        //
        // Automatically detach, if asked to do so
 
        if (FLAGS_DETACH & m_flags)
        {
            CloseHandle (m_threadHandle);
        }
    };
 
    ~Thread() {
        if (!(m_flags & FLAGS_DETACH) &&
            0 != WaitForSingleObject(m_threadHandle, INFINITE))
            CRL_DEBUG("WaitForSingleObject() failed: %d\n", GetLastError());
    };
 
private:
 
    uint32_t  m_flags;
    DWORD m_threadId;
    HANDLE m_threadHandle;
};
 
//
// A simple mutex class
 
class Mutex {
public:
    friend class ScopedLock;
 
    Mutex() {
        InitializeCriticalSection(&m_mutex);
    }
 
    ~Mutex() {
        DeleteCriticalSection(&m_mutex);
    };
 
private:
    CRITICAL_SECTION m_mutex;
};
 
//
// A simple scoped lock class
 
class ScopedLock
{
public:
 
    ScopedLock(Mutex& mutex) {
        this->lock(&mutex.m_mutex);
    };
 
    ScopedLock(CRITICAL_SECTION *lockP) {
        this->lock(lockP);
    };
 
    ScopedLock(CRITICAL_SECTION& lock) {
        this->lock(&lock);
    };
 
    ~ScopedLock() {
        LeaveCriticalSection(m_lockP);
    };
 
private:
 
    void lock(CRITICAL_SECTION *lockP) {
        m_lockP = lockP;
        EnterCriticalSection(m_lockP);
    };
 
    CRITICAL_SECTION *m_lockP;
};
 
// A futex-based semaphore.
//
// This implementation does not work across processes.
 
class Semaphore {
public:
 
    //
    // Wait for a post (decrement). If thread contention,
    // we may wake up, but be unable to snatch
    // the bait.. hence the while loop.
 
    bool wait() {
        do {
            if (0 == wait_(INFINITE))
                return true;
        } while (1);
    };
 
    //
    // Wait for a post, retrying until timeout
 
    bool timedWait(const double& timeout) {
 
        if (timeout < 0.0)
            CRL_EXCEPTION("invalid timeout: %f", timeout);
 
        do {
            int32_t ret = wait_((DWORD)(timeout * 1000));
 
            if (0 == ret)
                return true;
            else if (ETIMEDOUT == ret)
                return false;
 
        } while (1);
    };
 
    //
    // Post to the semaphore (increment.) Here we
    // signal the futex to wake up any waiters.
 
    bool post() {
 
        return ReleaseSemaphore(m_handle, 1, NULL) != FALSE;
 
    };
 
    //
    // Decrement the semaphore to zero in one-shot.. may
    // fail with thread contention, returns true if
    // successful
 
    bool clear() {
        while(WaitForSingleObject (m_handle, 0) == WAIT_OBJECT_0)
        {
        }
        return true;
    };
 
    int32_t waiters  () { return m_waiters; };
    bool    decrement() { return wait();    };
    bool    increment() { return post();    };
 
    Semaphore(LONG max=0) :
        m_waiters(0)
    {
        m_handle = CreateSemaphore (NULL, 0, (max == 0 || max > LONG_MAX) ? LONG_MAX : max, NULL);
        if (m_handle == NULL)
            CRL_EXCEPTION ("CreateSemaphore() failed: %d\n", GetLastError());
    }
 
    ~Semaphore()
    {
        if (m_handle != NULL)
            CloseHandle (m_handle);
    }
 
private:
 
    //
    // This actually does the synchronized decrement if possible, and goes
    // to sleep on the futex if not.
 
    inline int32_t wait_(DWORD ts=INFINITE) {
        InterlockedIncrement (&m_waiters);
        const int32_t ret = WaitForSingleObject (m_handle, ts);
        InterlockedDecrement (&m_waiters);
 
        if (ret == WAIT_OBJECT_0)
            return 0;
        else if (ret == WAIT_TIMEOUT)
            return ETIMEDOUT;
        else
            return EAGAIN;
    };
 
    HANDLE m_handle;
    LONG   m_waiters;
};
 
//
// A templatized variable signaler
 
template<class T> class WaitVar {
public:
 
    void post(const T& data) {
        {
            ScopedLock lock(m_lock);
            m_val = data;
        }
        m_sem.post();
    };
 
    bool wait(T& data) {
        m_sem.wait();
        {
            ScopedLock lock(m_lock);
            data = m_val;
        }
        return true;
    };
 
    bool timedWait(T& data,
                   const double& timeout) {
 
        if (false == m_sem.timedWait(timeout))
            return false;
        {
            ScopedLock lock(m_lock);
            data = m_val;
        }
        return true;
    }
 
    //
    // Use a semaphore with max value of 1. The WaitVar will
    // either be in a signaled state, or not.
 
    WaitVar() : m_val(),
                m_lock(),
                m_sem(1) {};
 
private:
 
    T                 m_val;
    Mutex     m_lock;
    Semaphore m_sem;
};
 
//
// A templatized wait queue
 
template <class T> class WaitQueue {
public:
 
    void post(const T& data) {
        bool postSem=true;
        {
            ScopedLock lock(m_lock);
 
            //
            // Limit deque size, if requested
 
            if (m_maximum > 0 &&
                m_maximum == m_queue.size()) {
 
                //
                // If at max entries, we will pop_front the oldest,
                // push_back the newest, and leave the semaphore alone
 
                m_queue.pop_front();
                postSem = false;
            }
 
            m_queue.push_back(data);
        }
        if (postSem)
            m_sem.post();
    };
 
    void kick() {
        m_sem.post();
    };
 
    bool wait(T& data) {
        m_sem.wait();
        {
            ScopedLock lock(m_lock);
 
            if (0 == m_queue.size())
                return false;
            else {
                data = m_queue.front();
                m_queue.pop_front();
                return true;
            }
        }
    }
 
    uint32_t waiters() {
        return m_sem.waiters();
    };
 
    uint32_t size() {
        ScopedLock lock(m_lock);
        return m_queue.size();
    }
 
    void clear() {
        ScopedLock lock(m_lock);
        m_queue.clear();
        while(false == m_sem.clear());
    }
 
    WaitQueue(std::size_t max=0) :
        m_maximum(max) {};
 
private:
 
    const std::size_t m_maximum;
    std::deque<T>     m_queue;
    Mutex             m_lock;
    Semaphore         m_sem;
};
 
}}}} // namespaces
 
#endif /* #ifndef CRL_MULTISENSE_THREAD_HH */