thread.h

Go to the documentation of this file.

// Evo C++ Library
/* Copyright 2019 Justin Crowell
Distributed under the BSD 2-Clause License -- see included file LICENSE.txt for details.
*/

#pragma once
#ifndef INCL_evo_thread_h
#define INCL_evo_thread_h

#include "impl/systhread.h"
#include "thread_inert.h"
#include "atomic.h"
#include "lock.h"
#include "type.h"
#if defined(EVO_CPP11)
    #include <functional>
#endif


// Disable certain MSVC warnings for this file, if !EVO_CPP11
#if defined(_MSC_VER) && !defined(EVO_CPP11)
    #pragma warning(push)
    #pragma warning(disable:4355)
#endif

namespace evo {


inline ulong get_tid()
    { return SysThread::id(); }


struct SpinLock : public AtomicFlag {
    typedef SmartLock<AtomicFlag>      Lock;        
    typedef SmartSleepLock<AtomicFlag> SleepLock;   

    SpinLock()
        { }

private:
    // Disable copying
    SpinLock(SpinLock&);
    SpinLock& operator=(SpinLock&);
};


struct SleepLock : public SpinLock {
    typedef SmartSleepLock<AtomicFlag> Lock;    

    SleepLock()
        { }

private:
    // Disable copying
    SleepLock(SleepLock&);
    SleepLock& operator=(SleepLock&);

    // Disable sleepless spin-lock
    using SpinLock::lock;
};


struct Mutex : public SysMutex {
    typedef SmartLock<Mutex> Lock;      

    Mutex()
        { }

private:
    // Disable copying
    Mutex(const Mutex&);
    Mutex& operator=(const Mutex&);
};


struct MutexRW {
    typedef SmartLock<MutexRW>     Lock;          
    typedef SmartLock<MutexRW>     LockWrite;     
    typedef SmartLockRead<MutexRW> LockRead;      

    MutexRW()
        { counter_ = 0; }

    ~MutexRW() {
        assert( counter_ == 0 ); // shouldn't be any read locks
    }

    bool trylock()
        { return write_mutex_.trylock(); }

    bool trylock(ulong timeout_ms)
        { return write_mutex_.trylock(timeout_ms); }

    void lock()
        { write_mutex_.lock(); }

    void unlock()
        { write_mutex_.unlock(); }

    bool trylock_read() {
        bool result = false;
        if (read_mutex_.trylock()) {
            if (counter_ > 0 || write_mutex_.trylock()) {
                result = true;
                ++counter_;
                assert(counter_ > 0);
            }
            read_mutex_.unlock();
        }
        return result;
    }

    void lock_read() {
        read_mutex_.lock();
        if (++counter_ == 1)
            write_mutex_.lock();
        assert(counter_ > 0);
        read_mutex_.unlock();
    }

    void unlock_read() {
        read_mutex_.lock();
        assert(counter_ > 0);
        if (--counter_ == 0)
            write_mutex_.unlock();
        read_mutex_.unlock();
    }

private:
    ulong counter_;
    Mutex read_mutex_;
    Mutex write_mutex_;
};


struct Condition {
    typedef SmartLock<Condition> Lock;      

#if defined(_WIN32)
    // Windows
    typedef CONDITION_VARIABLE Handle;

    static const ulong INF = INFINITE;

    Condition() {
        InitializeConditionVariable(&handle);
        owned = false;
        mutex = new Mutex;
        owned = true;
    }

    Condition(Mutex& mutex) : mutex(&mutex), owned(false)
        { InitializeConditionVariable(&handle); }

    bool wait(ulong timeout_ms=Condition::INF, bool locked=true) {
        if (!locked) {
            if (timeout_ms >= Condition::INF)
                mutex->lock();
            else
                mutex->trylock(timeout_ms);
        }
        return (SleepConditionVariableCS(&handle, &mutex->handle, timeout_ms) != 0);
    }

    void notify()
        { WakeConditionVariable(&handle); }

    void notify_all()
        { WakeAllConditionVariable(&handle); }

#else
    // Linux/Unix
    typedef pthread_cond_t Handle;

    static const ulong INF = ULONG_MAX;     

    Condition() {
        init();
        owned = false;
        mutex = new Mutex;
        owned = true;
    }

    Condition(Mutex& mutex) : mutex(&mutex), owned(false) {
        init();
    }

    bool wait(ulong timeout_ms=Condition::INF, bool locked=true) {
        if (timeout_ms >= Condition::INF) {
            if (!locked)
                mutex->lock();
            return (pthread_cond_wait(&handle, &mutex->handle) == 0);
        } else {
            if (!locked)
                mutex->trylock(timeout_ms);
            // Clock choice should be consistent with init() below
            struct timespec ts;
            #if defined(_POSIX_TIMERS) && defined(CLOCK_REALTIME) && !defined(EVO_USE_GETTIMEOFDAY)
                #if defined(CLOCK_MONOTONIC) && !defined(__APPLE__)
                    ::clock_gettime(CLOCK_MONOTONIC, &ts);
                #else
                    ::clock_gettime(CLOCK_REALTIME, &ts);
                #endif
            #else
            {
                struct timeval tv;
                ::gettimeofday(&tv, NULL);
                SysLinux::set_timespec_tv(ts, tv);
            }
            #endif
            SysLinux::add_timespec_ms(ts, timeout_ms);
            return (pthread_cond_timedwait(&handle, &mutex->handle, &ts) == 0);
        }
    }

    void notify()
        { pthread_cond_signal(&handle); }

    void notify_all()
        { pthread_cond_broadcast(&handle); }

private:
    void init() {
        // Clock choice should be consistent with wait() above
    #if defined(_POSIX_TIMERS) && defined(CLOCK_REALTIME) && !defined(EVO_USE_GETTIMEOFDAY) && !defined(__APPLE__)
        clockid_t id;
        #if defined(CLOCK_MONOTONIC)
            id = CLOCK_MONOTONIC;
        #else
            id = CLOCK_REALTIME;
        #endif
        pthread_condattr_t attr;
        pthread_condattr_init(&attr);
        pthread_condattr_setclock(&attr, id);
        pthread_cond_init(&handle, &attr);
    #else
        pthread_cond_init(&handle, NULL);
    #endif
    }

public:
#endif

    ~Condition() {
        if (owned)
            delete mutex;
    }

    bool wait_inf(bool locked=true) {
        bool result = wait(Condition::INF, locked);
        assert( result ); // should never timeout
        return result;
    }

    bool trylock()
        { return mutex->trylock(); }

    bool trylock(ulong timeout_ms)
        { return mutex->trylock(timeout_ms); }

    void lock()
        { mutex->lock(); }

    void unlock()
        { mutex->unlock(); }

    bool lock_wait(ulong timeout_ms=Condition::INF) {
        const bool result = wait(timeout_ms, false);
        mutex->unlock();
        return result;
    }

    void lock_notify() {
        mutex->lock();
        notify();
        mutex->unlock();
    }

    void lock_notify_all() {
        mutex->lock();
        notify_all();
        mutex->unlock();
    }

    Handle handle;      
    Mutex* mutex;       
    bool   owned;       

private:
    // Disable copying
    Condition(const Condition&);
    Condition& operator=(const Condition&);
};


// Implementation helpers for Thread
namespace impl {
    // Basic smart pointer used internally without depending on ptr.h
    template<class T> struct BasicSmartPtr {
        T* ptr;

        BasicSmartPtr(T* ptr=NULL) : ptr(ptr)
            { }

        ~BasicSmartPtr() {
            if (ptr != NULL)
                delete ptr;
        }

    private:
        // Disable copying
        BasicSmartPtr(const BasicSmartPtr&);
        BasicSmartPtr& operator=(const BasicSmartPtr&);
    };
}
struct Thread {
#if defined(EVO_CPP11)
    typedef std::function<void (void*)> Func;   
#else
    typedef void (*Func)(void*);                
#endif

    struct Init {
        Func  func;     
        void* arg;      

        Init() : func(NULL), arg(NULL)
            { }

        Init(Func func, void* arg=NULL) : func(func), arg(arg)
            { }

        Init(const Init& src) : func(src.func), arg(src.arg)
            { }
    };

    Init thread_init;   

    Thread() : thread_active_(false)
        { }

    Thread(Func func, void* arg=NULL) : thread_init(func, arg), thread_active_(false)
        { }

    Thread(const Init& init) : thread_init(init), thread_active_(false)
        { }

    virtual ~Thread()
        { thread_join(); }

    SysThread::Handle thread_handle() const
        { return thread_impl_.handle; }

    bool thread_active() const
        { return thread_active_; }

    bool thread_start() {
        if (!thread_active_ && thread_init.func != NULL) {
            impl::BasicSmartPtr<Init> init(new Init(thread_init));
            if (thread_impl_.start(Thread::thread_handler, init.ptr) == ENone) {
                // init.ptr freed by thread
                init.ptr       = NULL;
                thread_active_ = true;
                return true;
            }
        }
        return false;
    }

    virtual bool thread_detach() {
        if (thread_active_) {
            thread_impl_.detach();
            thread_active_ = false;
        }
        return true;
    }

    bool thread_join() {
        if (thread_active_ && thread_impl_.join()) {
            thread_active_ = false;
            return true;
        }
        return false;
    }

    static void yield()
        { SysThread::yield(); }

    static ulong id()
        { return SysThread::id(); }

protected:
    SysThread thread_impl_;
    bool      thread_active_;

private:
    // Platform-specific handler
    static EVO_THREAD_RUN_DEFINE(thread_handler, ptr) {
        assert(ptr != NULL);
        Thread::Init* init_ptr = (Thread::Init*)ptr;
        if (init_ptr != NULL) {
            Thread::Init init(*init_ptr);
            delete init_ptr;
            if (init.func != NULL)
                init.func(init.arg);
        }
        EVO_THREAD_RUN_RETURN;
    }
};


struct ThreadClass : public Thread {
    Condition condmutex;        

    ThreadClass() : Thread(ThreadClass::thread_base_handler, this)
        { }

    virtual ~ThreadClass() {
        if (thread_active_)
            abort();    // Thread must be stopped/joined first!
    }

    ThreadClass& thread_cancel(bool locked=false) {
        if (thread_active_)
            thread_cancel_flag_.store(1);
        if (!locked)
            condmutex.lock();
        condmutex.notify_all();
        condmutex.unlock();
        return *this;
    }

    bool thread_cancel_join(bool locked=false)
        { return thread_cancel(locked).thread_join(); }

    virtual void thread_run() = 0;

    bool cancel_check() {
        SysThread::yield();
        return (thread_cancel_flag_.load() != 0);
    }

private:
    AtomicInt thread_cancel_flag_;

    // Disable copying
    ThreadClass(const ThreadClass&);
    ThreadClass& operator=(const ThreadClass&);

    // Disable access to: thread_init, thread_detach
    using Thread::thread_init;
    bool thread_detach()
        { return false; }

    // Thread handler
    static void thread_base_handler(void* ptr)
        { ((ThreadClass*)ptr)->thread_run(); }
};


template<class T=Thread>
struct ThreadScope : public T {
    typedef ThreadScope<T> This;    

    ThreadScope()
        { this->thread_start(); }

    ~ThreadScope()
        { cancel_join(); }

    SysThread::Handle handle() const
        { return this->thread_handle(); }

    bool active() const
        { return this->thread_active(); }

    bool start()
        { return this->thread_start(); }

    This& cancel(bool locked=false)
        { this->thread_cancel(locked); return *this; }

    bool cancel_join(bool locked=false)
        { return this->thread_cancel(locked).thread_join(); }

    bool join()
        { return this->thread_join(); }

private:
    ThreadScope(const ThreadScope&);
    ThreadScope& operator=(const ThreadScope&);
};

template<>
struct ThreadScope<Thread> : public Thread {
    typedef ThreadScope<Thread> This;

    ThreadScope(Thread::Func func, void* arg=NULL) : Thread(func, arg)
        { this->thread_start(); }

    ~ThreadScope()
        { this->thread_join(); }

    SysThread::Handle handle() const
        { return this->thread_handle(); }

    bool active() const
        { return this->thread_active(); }

    bool start()
        { return this->thread_start(); }

    bool join()
        { return this->thread_join(); }

    bool detach()
        { return this->thread_detach(); }

private:
    ThreadScope();
    ThreadScope(const ThreadScope&);
    ThreadScope& operator=(const ThreadScope&);
};


// Implementation helpers for ThreadGroup
namespace impl {
    // Class-based thread node (derives from ThreadClass)
    template<class S, class T>
    struct ThreadGroupNode {
        typedef ThreadGroupNode<S,T> Node;
        static const bool THREAD_CLASS = true;

        T thread;
        Node* next;
        Node* prev;
        bool  used;

        ThreadGroupNode(Thread::Init&, S& state, Node* prev_node=NULL) : thread(state) {
            next = NULL;
            prev = prev_node;
            used = false;
        }

        void cancel()
            { thread.thread_cancel(); }
    };

    // Function-based thread node (specialize on Thread)
    template<class S>
    struct ThreadGroupNode<S,Thread> {
        typedef ThreadGroupNode<S,Thread> Node;
        static const bool THREAD_CLASS = false;

        Thread thread;
        Node* next;
        Node* prev;
        bool  used;

        ThreadGroupNode(Thread::Init& init, S& state, Node* prev_node=NULL) : thread(init) {
            next = NULL;
            prev = prev_node;
            used = false;
            thread.thread_init.arg = &state;
        }

        void cancel()
            { }
    };
}
template<class T, class S=typename T::SharedState, class M=MutexInert>
struct ThreadGroup {
    typedef ThreadGroup<T,S,M>    This;         
    typedef typename Thread::Init ThreadInit;   
    typedef S                     SharedState;  
    typedef M                     MutexT;       

    ThreadGroup() {
        size_  = 0;
        first_ = last_ = NULL;
        active_      = false;
        cancel_flag_ = false;
    }

    ThreadGroup(Thread::Func func) : thread_init(func) {
        size_  = 0;
        first_ = last_ = NULL;
        active_      = false;
        cancel_flag_ = false;
    }

    virtual ~ThreadGroup() {
        cancel();
        join();
    }

    bool active() const
        { typename MutexT::Lock lock(mutex_); return active_; }

    bool cancelled() const
        { typename MutexT::Lock lock(mutex_); return cancel_flag_; }

    ulong size() const
        { typename MutexT::Lock lock(mutex_); return size_; }

    bool start(uint count=1) {
        typename MutexT::Lock lock(mutex_);
        if (count > 0 && !cancel_flag_ && (Node::THREAD_CLASS || thread_init.func != NULL)) {
            active_ = true;
            if (first_ == NULL) {
                first_ = last_ = new Node(thread_init, shared_state);
                last_->thread.thread_start();
                ++size_;
                --count;
            }

            while (count > 0) {
                last_->next = new Node(thread_init, shared_state, last_);
                last_ = last_->next;
                last_->thread.thread_start();
                ++size_;
                --count;
            }
            return true;
        }
        return false;
    }

    This& cancel() {
        typename MutexT::Lock lock(mutex_);
        if (Node::THREAD_CLASS && active_ && !cancel_flag_) {
            cancel_flag_ = true;
            Node* node = first_;
            while (node != NULL) {
                node->cancel();
                node = node->next;
            }
        }
        return *this;
    }

    bool join() {
        typename MutexT::Lock lock(mutex_);
        if (active_) {
            Node* node = first_;
            while (node != NULL) {
                Node* next = node->next;
                node->cancel();
                node->thread.thread_join();
                delete node;
                node = next;
            }
            size_ = 0;
            first_ = last_ = NULL;
            cancel_flag_ = false;
            active_      = false;
        }
        return true;
    }

    SharedState shared_state;       
    ThreadInit  thread_init;        

protected:
    typedef impl::ThreadGroupNode<S,T> Node;

    ulong size_;
    Node* first_;
    Node* last_;
    bool  active_;
    bool  cancel_flag_;
    mutable MutexT mutex_;

private:
    // Disable copying
    ThreadGroup(const This&);
    This& operator=(const This&);
};


#if EVO_ATOMIC_SAFE_STATICS == 1 || defined(DOXYGEN)

template<class T>
struct Singleton {
    static T* instance() {
        static Mutex mutex;
        static AtomicSmartPtr static_ptr;
        T* p = static_ptr.load(EVO_ATOMIC_CONSUME);
        if (p == NULL) {
            Mutex::Lock lock(mutex);
            p = static_ptr.load(EVO_ATOMIC_RELAXED);
            if (p == NULL) {
                p = new T;
                static_ptr.store(p, EVO_ATOMIC_RELEASE);
            }
        }
        return p;
    }

private:
    struct AtomicSmartPtr : public Atomic<T*> {
        AtomicSmartPtr()
            { }

        ~AtomicSmartPtr() {
            T* ptr = this->load();
            if (ptr != NULL)
                delete ptr;
        }
    };
};
#endif


}
#if defined(_MSC_VER) && !defined(EVO_CPP11)
    #pragma warning(pop)
#endif
#endif