AsynchronousTask.h

#pragma once
 
#include <condition_variable>
#include <functional>
#include <mutex>
#include <queue>
#include <thread>
 
#ifdef __linux__
    #include <sched.h>
#endif
namespace CoRiBoDynamics
{
    // Create a pool of threads, each bound to one logical or physical core
    class CoreBoundThreadPool {
    public:
        /*
            To execute a task under the CoreBoundThreadPool, create a subclass of Task that implements the Execute function and add it to the tread pool.
        */
        class Task {
        public:
            virtual ~Task(){};
            virtual void Execute() = 0;
        };
 
        /*
            pool_size: number of threads in the pool
            start_ix: core index of the first core that should bind to the first thread
            hyperthreading_core_merge: if false, each thread binds to consecutive logical cores. if true, it overrides hyperthreading, puts one thread per physical core and allows the thread to bind to either logical core. If CPU is not hyperthreaded, this option should probably not be used.
        */
        CoreBoundThreadPool(int pool_size, int start_ix, bool hyperthreading_core_merge);
        void ExecuteTask(Task* task, int thread_id); // add task to the private queue of thread_id
        void ExecuteTask(Task* task); // add task to the shared queue
        void ExecuteTaskSet(std::function<Task*(int)> task_set, int num);
        void ExecuteTaskSet(std::function<Task*(int)> task_set, std::function<int(int)> thread_id, int num);
        void BlockUntilCompletion(int thread_id); // wait until thread_id is idle.
        void BlockUntilCompletion(); // wait until all currently added tasks have completed. may wait for tasks added later, but no guaranties for this
        virtual ~CoreBoundThreadPool();
        int PoolSize();
 
        int smart_index(int ix); 
 
    protected:
        class mutex_queue {
        public:
            mutex_queue(){
                m_sub_queue = nullptr;
            }
 
            explicit mutex_queue(mutex_queue* sub_queue){
                m_sub_queue = sub_queue;
            }
 
            void push(Task* t);
 
            Task* pop();
 
            void KillAllTasks();
 
            bool isEmpty();
 
        private:
            std::mutex  m_mutex;
            //std::queue<Task*> m_queue;
            std::vector<Task*> m_queue;
            mutex_queue* m_sub_queue;
        };
 
 
        class WorkUnit {
        public:
 
            WorkUnit(size_t cpu_mask, mutex_queue* sub_queue);
 
            void WaitUntilIdle();
 
            void TerminateUnit();
 
            void AddToQueue(Task* t);
 
            void NudgeThread();
 
        protected:
            bool                        m_stay_in_loop;
            std::thread             m_thread;
            std::condition_variable m_internal_barrier;
            std::condition_variable m_external_barrier;
            std::mutex              m_mutex;
            mutex_queue                 m_queue;
            enum thread_state {IDLE, EXECUTING} m_state;
 
            void ThreadLoop();
 
            void setCpuMask(size_t cpu_mask);
        };
        std::vector<WorkUnit*> m_WorkUnit;
 
        bool m_DestructorCalled;
        std::mutex  m_mutex;
 
        mutex_queue     m_shared_queue;
 
        bool m_hyperthreading_core_merge;
    };
 
}