Pool.h

Go to the documentation of this file.

#pragma once
 
 
#include "objects/containers/Array.h"
#include "objects/wrappers/Function.h"
#include "objects/wrappers/Optional.h"
#include "thread/Scheduler.h"
#include <atomic>
#include <condition_variable>
#include <queue>
#include <thread>
 
NAMESPACE_SPH_BEGIN
 
class Task : public ITask, public ShareFromThis<Task> {
private:
    std::condition_variable waitVar;
    std::mutex waitMutex;
 
    std::atomic<int> tasksLeft{ 1 };
 
    Function<void()> callable = nullptr;
 
    SharedPtr<Task> parent = nullptr;
 
    std::exception_ptr caughtException = nullptr;
 
public:
    explicit Task(const Function<void()>& callable);
 
    ~Task();
 
    virtual void wait() override;
 
    virtual bool completed() const override;
 
    void setParent(SharedPtr<Task> parent);
 
    void setException(std::exception_ptr exception);
 
    bool isRoot() const;
 
    SharedPtr<Task> getParent() const;
 
    static SharedPtr<Task> getCurrent();
 
    void runAndNotify();
 
private:
    void addReference();
 
    void removeReference();
};
 
class ThreadPool : public IScheduler {
    friend class Task;
 
private:
    Array<AutoPtr<std::thread>> threads;
 
    Size granularity;
 
    std::queue<SharedPtr<Task>> tasks;
 
    std::condition_variable taskVar;
    std::mutex taskMutex;
 
    std::condition_variable waitVar;
    std::mutex waitMutex;
 
    std::atomic<bool> stop;
 
    std::atomic<int> tasksLeft;
 
    static SharedPtr<ThreadPool> globalInstance;
 
public:
    ThreadPool(const Size numThreads = 0, const Size granularity = 1000);
 
    ~ThreadPool();
 
    virtual SharedPtr<ITask> submit(const Function<void()>& task) override;
 
    virtual Optional<Size> getThreadIdx() const override;
 
    virtual Size getThreadCnt() const override;
 
    virtual Size getRecommendedGranularity() const override;
 
    void waitForAll();
 
    Size remainingTaskCnt() {
        return tasksLeft;
    }
 
    void setGranularity(const Size newGranularity) {
        granularity = newGranularity;
    }
 
    static SharedPtr<ThreadPool> getGlobalInstance();
 
private:
    SharedPtr<Task> getNextTask(const bool wait);
 
    void processTask(const bool wait);
};
 
NAMESPACE_SPH_END