tasks.h

Go to the documentation of this file.

 
#ifndef ORIGINAL_TASKS_H
#define ORIGINAL_TASKS_H
 
#include "async.h"
#include "atomic.h"
#include "queue.h"
#include "refCntPtr.h"
#include "array.h"
#include "prique.h"
#include "vector.h"
 
namespace original {
 
    // ==================== Task Delegator (Thread Pool) ====================
 
    class taskDelegator {
        // ==================== Task Base Interface ====================
 
        class taskBase {
        public:
            virtual void run() = 0;
 
            virtual ~taskBase() = default;
        };
 
        // ==================== Concrete Task Class ====================
 
        template<typename TYPE>
        class task final : public taskBase {
            async::promise<TYPE, std::function<TYPE()>> p;  
 
        public:
            task(const task&) = delete;                 
            task& operator=(const task&) = delete;      
            task(task&&) = default;                     
            task& operator=(task&&) = default;          
            task() = default;
 
            template<typename Callback, typename... Args>
            explicit task(Callback&& c, Args&&... args);
 
            void run() override;
 
            async::future<TYPE> getFuture();
        };
 
    public:
        // ==================== Task Priorities ====================
 
        enum class priority : u_integer {
            IMMEDIATE = 0,  
            HIGH = 1,       
            NORMAL = 2,     
            LOW = 3,        
            DEFERRED = 4,   
        };
 
        enum class stopMode {
            DISCARD_DEFERRED,  
            KEEP_DEFERRED,     
            RUN_DEFERRED,      
        };
 
        // Convenience constants
        static constexpr auto IMMEDIATE = priority::IMMEDIATE;
        static constexpr auto HIGH = priority::HIGH;
        static constexpr auto NORMAL = priority::NORMAL;
        static constexpr auto LOW = priority::LOW;
        static constexpr auto DEFERRED = priority::DEFERRED;
 
        static constexpr auto DISCARD_DEFERRED = stopMode::DISCARD_DEFERRED;
        static constexpr auto KEEP_DEFERRED = stopMode::KEEP_DEFERRED;
        static constexpr auto RUN_DEFERRED = stopMode::RUN_DEFERRED;
 
    private:
        // Internal type definitions
        using priorityTask = couple<strongPtr<taskBase>, priority>;  
 
        template<typename COUPLE>
        struct taskComparator {
            bool operator()(const COUPLE& lhs, const COUPLE& rhs) const;
        };
 
        using priorityTaskQueue = prique<priorityTask, taskComparator, vector>;  
 
        array<thread> threads_;              
        priorityTaskQueue tasks_waiting_;    
        queue<strongPtr<taskBase>> task_immediate_;  
        queue<strongPtr<taskBase>> tasks_deferred_;  
        mutable pCondition condition_;       
        mutable pMutex mutex_;               
        bool stopped_;                       
        u_integer active_threads_;           
        u_integer idle_threads_;             
 
        template<typename TYPE>
        async::future<TYPE> submit(priority priority, strongPtr<task<TYPE>>& t);
 
    public:
        taskDelegator(const taskDelegator&) = delete;               
        taskDelegator& operator=(const taskDelegator&) = delete;    
        taskDelegator(taskDelegator&&) = delete;                    
        taskDelegator& operator=(taskDelegator&&) = delete;         
 
        explicit taskDelegator(u_integer thread_cnt = 8);
 
        template<typename Callback, typename... Args>
        auto submit(Callback&& c, Args&&... args);
 
        template<typename Callback, typename... Args>
        auto submit(priority priority, Callback&& c, Args&&... args);
 
        template<typename Callback, typename... Args>
        auto submit(time::duration timeout, Callback&& c, Args&&... args);
 
        u_integer waitingCnt() const noexcept;
 
        u_integer immediateCnt() const noexcept;
 
        void runDeferred();
 
        void runAllDeferred();
 
        u_integer discardDeferred();
 
        void discardAllDeferred();
 
        u_integer deferredCnt() const noexcept;
 
        void stop(stopMode mode = stopMode::KEEP_DEFERRED);
 
        u_integer activeThreads() const noexcept;
 
        u_integer idleThreads() const noexcept;
 
        ~taskDelegator();
    };
} // namespace original
 
// ==================== Task Implementation ====================
 
template <typename TYPE>
template <typename Callback, typename... Args>
original::taskDelegator::task<TYPE>::task(Callback&& c, Args&&... args)
    : p([c = std::forward<Callback>(c), ...args = std::forward<Args>(args)]() mutable {
        return c(args...);
    }) {}
 
template <typename TYPE>
void original::taskDelegator::task<TYPE>::run()
{
    this->p.run();
}
 
template <typename TYPE>
original::async::future<TYPE> original::taskDelegator::task<TYPE>::getFuture()
{
    return this->p.getFuture();
}
 
// ==================== Task Delegator Implementation ====================
 
template <typename COUPLE>
bool original::taskDelegator::taskComparator<COUPLE>::operator()(const COUPLE& lhs, const COUPLE& rhs) const
{
    return static_cast<u_integer>(lhs.second()) < static_cast<u_integer>(rhs.second());
}
 
inline original::taskDelegator::taskDelegator(const u_integer thread_cnt)
    : threads_(thread_cnt),
      stopped_(false),
      active_threads_(0),
      idle_threads_(0) {
    for (auto& thread_ : this->threads_) {
        thread_ = thread {
            [this]{
                while (true) {
                    strongPtr<taskBase> task;
                    {
                        uniqueLock lock(this->mutex_);
                        this->idle_threads_ += 1;
                        this->condition_.wait(this->mutex_, [this] {
                            return this->stopped_ || !this->tasks_waiting_.empty() || !this->task_immediate_.empty();
                        });
 
                        if (this->stopped_ &&
                            this->tasks_waiting_.empty() &&
                            this->task_immediate_.empty()) {
                                this->idle_threads_ -= 1;
                                return;
                        }
 
                        if (!this->task_immediate_.empty()) {
                            task = std::move(this->task_immediate_.pop());
                        } else {
                            task = std::move(this->tasks_waiting_.pop().first());
                        }
                        this->idle_threads_ -= 1;
                    }
 
                    {
                        uniqueLock lock(this->mutex_);
                        this->active_threads_ += 1;
                    }
                    task->run();
                    {
                        uniqueLock lock(this->mutex_);
                        this->active_threads_ -= 1;
                    }
                }
            }
        };
    }
}
 
template <typename Callback, typename ... Args>
auto original::taskDelegator::submit(Callback&& c, Args&&... args)
{
    return this->submit(priority::NORMAL, std::forward<Callback>(c), std::forward<Args>(args)...);
}
 
template <typename Callback, typename ... Args>
auto original::taskDelegator::submit(const priority priority, Callback&& c, Args&&... args)
{
    using ReturnType = decltype(c(args...));
    strongPtr<task<ReturnType>> new_task = makeStrongPtr<task<ReturnType>>(
        std::forward<Callback>(c),
        std::forward<Args>(args)...
    );
    return this->submit<ReturnType>(priority, new_task);
}
 
template <typename Callback, typename ... Args>
auto original::taskDelegator::submit(time::duration timeout, Callback&& c, Args&&... args)
{
    using ReturnType = decltype(c(args...));
    strongPtr<task<ReturnType>> new_task = makeStrongPtr<task<ReturnType>>(
        std::forward<Callback>(c),
        std::forward<Args>(args)...
    );
    auto f = new_task->getFuture();
    {
        uniqueLock lock(this->mutex_);
        if (this->stopped_) {
            throw sysError("taskDelegator already stopped");
        }
        const bool success = this->condition_.waitFor(this->mutex_, timeout, [this]{
            return this->idle_threads_ > 0;
        });
        if (!success) {
            throw sysError("No idle threads available within timeout");
        }
        this->task_immediate_.push(std::move(new_task.template dynamicCastTo<taskBase>()));
    }
    this->condition_.notify();
    return f;
}
 
inline original::u_integer original::taskDelegator::waitingCnt() const noexcept
{
    uniqueLock lock(this->mutex_);
    return this->tasks_waiting_.size();
}
 
inline original::u_integer original::taskDelegator::immediateCnt() const noexcept
{
    uniqueLock lock(this->mutex_);
    return this->task_immediate_.size();
}
 
template <typename TYPE>
original::async::future<TYPE>
original::taskDelegator::submit(const priority priority, strongPtr<task<TYPE>>& t)
{
    auto f = t->getFuture();
    {
        uniqueLock lock(this->mutex_);
        if (this->stopped_) {
            throw sysError("taskDelegator already stopped");
        }
        switch (priority) {
        case priority::IMMEDIATE:
            if (this->idle_threads_ == 0) {
                throw sysError("No idle threads now");
            }
            this->task_immediate_.push(std::move(t.template dynamicCastTo<taskBase>()));
            break;
        case priority::HIGH:
        case priority::NORMAL:
        case priority::LOW:
            this->tasks_waiting_.push(priorityTask{t.template dynamicCastTo<taskBase>(), priority});
            break;
        case priority::DEFERRED:
            this->tasks_deferred_.push(t.template dynamicCastTo<taskBase>());
            return f;
        default:
            throw sysError("Unknown priority");
        }
    }
    this->condition_.notify();
    return f;
}
 
inline void original::taskDelegator::runDeferred()
{
    {
        uniqueLock lock(this->mutex_);
        if (!this->tasks_deferred_.empty()) {
            this->tasks_waiting_.push(priorityTask{this->tasks_deferred_.pop(), priority::DEFERRED});
        } else {
            return;
        }
    }
    this->condition_.notify();
}
 
inline void original::taskDelegator::runAllDeferred()
{
    {
        uniqueLock lock(this->mutex_);
        if (this->tasks_deferred_.empty()) {
            return;
        }
        while (!this->tasks_deferred_.empty()) {
            this->tasks_waiting_.push(priorityTask{this->tasks_deferred_.pop(), priority::DEFERRED});
        }
    }
    this->condition_.notifyAll();
}
 
inline original::u_integer original::taskDelegator::discardDeferred()
{
    uniqueLock lock(this->mutex_);
    if (!this->tasks_deferred_.empty()) {
        this->tasks_deferred_.pop();
    }
    return this->tasks_deferred_.size();
}
 
inline void original::taskDelegator::discardAllDeferred()
{
    uniqueLock lock(this->mutex_);
    if (!this->tasks_deferred_.empty()) {
        this->tasks_deferred_.clear();
    }
}
 
inline original::u_integer original::taskDelegator::deferredCnt() const noexcept
{
    uniqueLock lock(this->mutex_);
    return this->tasks_deferred_.size();
}
 
inline void original::taskDelegator::stop(const stopMode mode)
{
    {
        uniqueLock lock(this->mutex_);
        switch (mode) {
        case RUN_DEFERRED:
            while (!this->tasks_deferred_.empty()) {
                this->tasks_waiting_.push(priorityTask{this->tasks_deferred_.pop(), DEFERRED});
            }
            break;
        case DISCARD_DEFERRED:
            this->tasks_deferred_.clear();
            break;
        case KEEP_DEFERRED:
            break;
        default:
            throw sysError("Unknown stop mode");
        }
        this->stopped_ = true;
    }
    this->condition_.notifyAll();
}
 
inline original::u_integer original::taskDelegator::activeThreads() const noexcept
{
    uniqueLock lock(this->mutex_);
    return this->active_threads_;
}
 
inline original::u_integer original::taskDelegator::idleThreads() const noexcept
{
    uniqueLock lock(this->mutex_);
    return this->idle_threads_;
}
 
inline original::taskDelegator::~taskDelegator()
{
    this->stop(stopMode::RUN_DEFERRED);
    for (auto& thread : threads_) {
        if (thread.joinable())
            thread.join();
    }
}
 
#endif //ORIGINAL_TASKS_H