ThreadPool.h

Go to the documentation of this file.

// Copyright 2004 Mark Berrill. All Rights Reserved. This work is distributed with permission,
// but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
// PARTICULAR PURPOSE.
#ifndef included_AMP_ThreadPool
#define included_AMP_ThreadPool
 
#include <atomic>
#include <condition_variable>
#include <cstring>
#include <functional>
#include <mutex>
#include <stdexcept>
#include <string>
#include <thread>
#include <typeinfo>
#include <vector>
 
#include "AMP/utils/threadpool/ThreadPoolId.h"
#include "AMP/utils/threadpool/ThreadPoolQueue.h"
#include "AMP/utils/threadpool/ThreadPoolWorkItem.h"
 
 
namespace AMP {
// clang-format off
 
 
class alignas(16) ThreadPool final
{
public:
    constexpr static uint16_t MAX_THREADS = 128; // The maximum number of threads (must be a multiple of 64)
 
public:
 
 
    typedef ThreadPoolWorkItem WorkItem;
 
 
    template <typename return_type>
    class WorkItemRet : public ThreadPool::WorkItem
    {
    public:
        virtual void run() override = 0;
        virtual bool has_result() const override final { return !std::is_same<return_type,void>::value; }
        return_type get_results() const { return d_result; }
        virtual ~WorkItemRet() {}
    protected:
        return_type d_result;
    protected:
        inline WorkItemRet() : d_result( return_type() ) { }
    private:
        WorkItemRet( const WorkItemRet & );            // Private copy constructor
        WorkItemRet &operator=( const WorkItemRet & ); // Private assignment operator
    };
 
 
public:
 
    ThreadPool( const int N = 0, const std::string &affinity = "none",
        const std::vector<int> &procs = std::vector<int>(), int queueSize = 4096 );
 
 
    ThreadPool( const ThreadPool & ) = delete;
 
 
    ThreadPool &operator=( const ThreadPool & ) = delete;
 
 
    ~ThreadPool();
 
 
    static int getNumberOfProcessors();
 
 
    static int getCurrentProcessor();
 
 
    static std::vector<int> getProcessAffinity();
 
 
    static void setProcessAffinity( const std::vector<int>& procs );
 
 
    static std::vector<int> getThreadAffinity();
 
 
    std::vector<int> getThreadAffinity( int thread ) const;
 
 
    static void setThreadAffinity( const std::vector<int>& procs );
 
 
    void setThreadAffinity( int thread, const std::vector<int>& procs ) const;
 
 
    inline int getNumThreads() const { return d_N_threads; }
 
 
    void setNumThreads( const int N, const std::string &affinity = "none",
        const std::vector<int> &procs = std::vector<int>() );
 
 
    inline void setMaxWaitTimeDebug( const int time ) { d_max_wait_time = time; }
 
 
    int getThreadNumber() const;
 
 
 
    inline bool isValid( const ThreadPoolID &id ) const;
 
 
    inline bool isFinished( const ThreadPoolID& id ) const { return id.finished(); }
 
 
    template <class return_type>
    static inline return_type getFunctionRet( const ThreadPoolID &id );
 
 
    template<class Ret, class... Args1, class... Args2>
    static inline WorkItem* createWork( std::function<Ret(Args1...)> routine, std::tuple<Args2...> &&args );
 
 
    template<class Ret, class... Args1, class... Args2>
    static inline WorkItem* createWork( Ret( *routine )( Args1... ), std::tuple<Args2...> &&args );
 
 
    template<class Ret, class... Args1, class... Args2>
    static inline WorkItem* createWork( std::function<Ret(Args1...)> routine, Args2... args );
 
 
    template <class Ret, class... Args1, class... Args2>
    static inline WorkItem* createWork( Ret( *routine )( Args1... ), Args2... args );
 
 
    inline ThreadPoolID add_work( ThreadPool::WorkItem *work, int priority = 0 );
 
 
    inline std::vector<ThreadPoolID> add_work( const std::vector<ThreadPool::WorkItem *> &work,
        const std::vector<int> &priority = std::vector<int>() );
 
 
    inline void wait( const ThreadPoolID &id ) const;
 
 
    inline size_t wait_any( const std::vector<ThreadPoolID> &ids ) const;
 
 
    inline void wait_all( const std::vector<ThreadPoolID> &ids ) const;
 
 
    inline std::vector<int> wait_some( int N_wait, const std::vector<ThreadPoolID> &ids, int max_wait = 10000000 ) const;
 
 
    void wait_pool_finished() const;
 
 
    static bool is_valid( const ThreadPool *tpool );
 
 
    static void set_OS_warnings( int behavior = 0 );
 
 
    int N_queued( ) const { return d_queue.size(); }
 
 
    void setErrorHandler( std::function<void(const std::string&)> fun );
 
 
public: // Static interface
 
    static inline int numThreads( const ThreadPool* tpool ) { return tpool ? tpool->getNumThreads() : 0; }
 
    static inline ThreadPoolID add_work( ThreadPool* tpool, ThreadPool::WorkItem *work, int priority = 0 );
 
 
    static inline std::vector<ThreadPoolID> add_work( ThreadPool* tpool, const std::vector<ThreadPool::WorkItem *> &work,
        const std::vector<int> &priority = std::vector<int>() );
 
 
    static inline void wait_all( const ThreadPool* tpool, const std::vector<ThreadPoolID> &ids );
 
 
    static inline void wait_pool_finished( const ThreadPool* tpool ) { if ( tpool ) { tpool->wait_pool_finished(); } }
 
 
private:
 
    // Implimentation of condition_variable which does not require a lock
    class condition_variable final
    {
      public:
        condition_variable() { }
        ~condition_variable() { }
        inline void wait() const { std::unique_lock<std::mutex> lock(d_mutex); d_cv.wait(lock); }
        inline void wait_for( double seconds ) const
        {
            std::unique_lock<std::mutex> lock(d_mutex);
            if ( seconds < 4e-6 )
                d_cv.wait_for(lock,std::chrono::nanoseconds(static_cast<int>(1e9*seconds)));
            else if ( seconds < 4e-3 )
                d_cv.wait_for(lock,std::chrono::microseconds(static_cast<int>(1e6*seconds)));
            else if ( seconds < 4 )
                d_cv.wait_for(lock,std::chrono::milliseconds(static_cast<int>(1e3*seconds)));
            else
                d_cv.wait_for(lock,std::chrono::seconds(static_cast<int>(seconds)));
        }
        inline void notify_one() const { d_cv.notify_one(); }
        inline void notify_all() const { d_cv.notify_all(); }
      private:
        mutable std::condition_variable d_cv;
        mutable std::mutex d_mutex;
    };
 
 
private:
 
    // Function to check the startup
    void check_startup( );
 
    // Function to add an array of work items
    void add_work(
        size_t N, ThreadPool::WorkItem *work[], const int *priority, ThreadPoolID *id );
 
    // Function to get a work item that has finished
    static inline WorkItem *getFinishedWorkItem( const ThreadPoolID& id )
    {
        return id.finished() ? id.getWork():nullptr;
    }
 
    // This function provides a wrapper (needed for the threads)
    static inline void create_new_thread( ThreadPool *tpool, int id )
    {
        tpool->tpool_thread( id );
    }
 
    /* This is the function that controls the individual thread and allows it to do work.
     * Note: this version uses a last in - first out work scheduling.
     * param thread_init - Structure address containing the startup information for the thread */
    void tpool_thread( int id );
 
    // Function to check if the current thread is a member of the thread pool
    inline bool isMemberThread() const { return getThreadNumber()>=0; }
 
    // Function to wait for some work items to finish
    std::vector<bool> wait_some( size_t N_work, const ThreadPoolID *ids, size_t N_wait, int max_wait ) const;
    
    // Function to wait for N work items to finish (may return early)
    void wait_N( int N, double time ) const;
    
    // Check if we are waiting too long and pring debug info
    void print_wait_warning( ) const;
 
 
private:
 
    // Typedefs
    typedef volatile std::atomic_uint32_t vint32_t; // volatile atomic int
    typedef volatile std::atomic_uint64_t vint64_t; // volatile atomic int64
    typedef condition_variable cond_t;              // condition variable
 
    // Internal data
    uint32_t d_NULL_HEAD;                 // Null data buffer to check memory bounds
    volatile mutable bool d_signal_empty; // Do we want to send a signal when the queue is empty
    uint16_t d_N_threads;                 // Number of threads
    int d_max_wait_time;                  // The maximum time waiting before printing a warning message
    mutable vint32_t d_signal_count;      // Signal count
    vint32_t d_num_active;                // Number of threads that are currently active
    vint64_t d_id_assign;                 // An internal variable used to store the current id to assign
    vint64_t d_active[MAX_THREADS/64];    // Which threads are currently active
    vint64_t d_cancel[MAX_THREADS/64];    // Which threads should be deleted
    vint64_t d_N_added;                   // Number of items added to the work queue
    vint64_t d_N_started;                 // Number of items started
    vint64_t d_N_finished;                // Number of items finished
    mutable cond_t d_wait_finished;       // Condition variable to signal when work is finished
    mutable cond_t d_wait_work;           // Condition variable to signal when there is new work
    ThreadPoolListQueue d_queue;          // The work queue
    std::function<void(const std::string&)> d_errorHandler; // Error handler
    std::thread *d_thread;                // Handles to the threads
    uint32_t d_NULL_TAIL;                 // Null data buffer to check memory bounds
};
 
 
} // AMP namespace
 
 
#include "AMP/utils/threadpool/ThreadPool.hpp"
 
 
// clang-format on
#endif