Simple ThreadPool

Question

ThreadPool, Task

I continue to study C++ for the fourth month, every day from morning to night. Today I want to submit my Thread Pool for review. I'm not really sure if I have the right code style, and if there are any errors with move-semantic. In the file main.cpp I left the tests to check that the class is working properly. I wrote the tests in a hurry (you don't need to check tests correctness). I would appreciate any comments!

GitHub: https://github.com/vansergh/lib-threadpool

task.hpp

#ifndef INCLUDE_GUARD_TASK_HPP
#define INCLUDE_GUARD_TASK_HPP

#include <any>
#include <cstdint>
#include <cstddef>
#include <functional>
#include <future>
#include <memory>

namespace vsock {

    //////////////////////////////////////////////////////////////////////////////////
    // Task class declaration
    ////////////////////////////////////////////////////////////////////////////////

    class Task {
    private:
        enum class TaskType : std::uint8_t { ASYNC, SYNC, LOOP };
    public:
        Task(const Task&) = delete;
        Task& operator=(const Task&) = delete;
    public:

        Task() = default;

        ~Task();

        Task(Task&& other);
        Task& operator=(Task&& other);

        template<typename F, typename... Args>
        auto SetSyncJob(F&& job, Args&&... args);

        template<typename F, typename... Args>
        void SetAsyncJob(F&& loop, Args&&... args);

        template<typename F, typename... Args>
        void SetLoopJob(F&& loop, Args&&... args);

        template<typename F, typename... Args>
        void SetCondition(F&& condition, Args&&... args);

        template<typename... Args>
        void AddVariables(Args&&... vars);

        std::any& GetVariable(std::size_t index);

        bool IsVoidResult() {
            return is_void_;
        }

        bool operator()();

    private:

        TaskType type_{ TaskType::ASYNC };
        bool is_void_{ true };
        std::unique_ptr<std::packaged_task<void(void)>> sync_task_;
        std::unique_ptr<std::function<void(Task&)>> async_task_{ nullptr };
        std::unique_ptr<std::function<bool(Task&)>> condition_{ nullptr };
        std::vector<std::any> vars_;

    };

    //////////////////////////////////////////////////////////////////////////////////
    // ThreadPool class defenition (template methods)
    ////////////////////////////////////////////////////////////////////////////////

    template<typename F, typename ...Args>
    inline auto Task::SetSyncJob(F&& job, Args && ...args) {

        type_ = TaskType::SYNC;
        condition_.reset();
        async_task_.reset();

        using return_type = std::invoke_result_t<F, Args...>;
        using promise_type = std::promise<return_type>;
        using bind_type = std::function<return_type(void)>;

        is_void_ = std::is_void_v<return_type>;

        const std::shared_ptr<bind_type> bind_fnc_ptr = std::make_shared<bind_type>(std::move(std::bind(std::move(job), std::move(args)...)));
        const std::shared_ptr<promise_type> task_promise_ptr = std::make_shared<promise_type>();
        sync_task_.reset();
        sync_task_ = std::make_unique<std::packaged_task<void(void)>>(
            std::move(std::packaged_task([bind_fnc_ptr, task_promise_ptr]() {
            try {
                if constexpr (std::is_void_v<return_type>) {
                    (*bind_fnc_ptr)();
                    task_promise_ptr->set_value();
                }
                else {
                    task_promise_ptr->set_value((*bind_fnc_ptr)());
                }
            }
            catch (...) {
                try {
                    task_promise_ptr->set_exception(std::current_exception());
                }
                catch (...) {
                    throw std::runtime_error("set_exception() failed");
                }
            }
        })));


        return task_promise_ptr->get_future();
    }

    template<typename F, typename ...Args>
    inline void Task::SetAsyncJob(F&& job, Args && ...args) {
        type_ = TaskType::ASYNC;
        sync_task_.reset();
        condition_.reset();
        is_void_ = true;
        async_task_.reset();
        async_task_ = std::make_unique<std::function<void(Task&)>>(std::move(
            std::bind(std::move(job), std::move(args)...)
        ));
    }

    template<typename F, typename ...Args>
    inline void Task::SetCondition(F&& condition, Args && ...args) {
        type_ = TaskType::LOOP;
        is_void_ = true;
        sync_task_.reset();
        condition_.reset();
        condition_ = std::make_unique<std::function<bool(Task&)>>(std::move(
            std::bind(std::move(condition), std::move(args)...)
        ));
    }

    template<typename F, typename ...Args>
    inline void Task::SetLoopJob(F&& loop, Args && ...args) {
        type_ = TaskType::LOOP;
        is_void_ = true;
        sync_task_.reset();
        async_task_.reset();
        async_task_ = std::make_unique<std::function<void(Task&)>>(std::move(
            std::bind(std::move(loop), std::move(args)...)
        ));
    }

    template<typename... Args>
    inline void Task::AddVariables(Args&&... vars) {
        (vars_.emplace_back(std::move(vars)), ...);
    }

}

#endif

task.cpp

#include <task.hpp>

#include <utility>
#include <stdexcept>
#include <type_traits>

namespace vsock {

    //////////////////////////////////////////////////////////////////////////////////
    // Task class defenition
    ////////////////////////////////////////////////////////////////////////////////

    Task::~Task() {
        /*
        task_.reset();
        condition_.reset();
        vars_.clear();
        loop_.reset();
        */
    }

    Task::Task(Task&& other) :
        type_{ std::exchange(other.type_,TaskType::ASYNC) },
        is_void_{ std::exchange(other.is_void_,true) },
        sync_task_{ std::exchange(other.sync_task_,{}) },
        async_task_{ std::exchange(other.async_task_,{}) },
        condition_{ std::exchange(other.condition_,{}) },
        vars_(std::move(other.vars_))
    {}

    Task& Task::operator=(Task&& other) {
        if (this != &other) {
            type_ = std::exchange(other.type_, TaskType::ASYNC);
            is_void_ = std::exchange(other.is_void_, true);
            sync_task_ = std::exchange(other.sync_task_, {});
            async_task_ = std::exchange(other.async_task_, {});
            condition_ = std::exchange(other.condition_, {});
            vars_ = std::move(other.vars_);
        }
        return *this;
    }

    std::any& Task::GetVariable(std::size_t index) {
        return vars_[index];
    }

    bool Task::operator()() {
        switch (type_) {
            case TaskType::SYNC: {
                (*sync_task_)();
                return false;
            } break;
            case TaskType::LOOP: {
                if (!condition_.get() || !async_task_.get()) {
                    throw std::runtime_error("condition or loop is not set");
                }
                if ((*condition_)(*this)) {
                    (*async_task_)(*this);
                    return true;
                }
                return false;
            } break;
            default: { // TaskType::ASYNC
                (*async_task_)(*this);
                return false;
            }
        }
    }

}

threadpool.hpp

#ifndef INCLUDE_GUARD_THREADPOOL_HPP
#define INCLUDE_GUARD_THREADPOOL_HPP

#include <cstdint>
#include <cstddef> 
#include <thread>
#include <mutex>
#include <memory>
#include <deque>
#include <condition_variable>

#include <task.hpp>

namespace vsock {

    //////////////////////////////////////////////////////////////////////////////////
    // ThreadPool class declaration
    ////////////////////////////////////////////////////////////////////////////////

    class ThreadPool {
    public:

        ThreadPool(const ThreadPool&) = delete;
        ThreadPool(ThreadPool&&) = delete;
        ThreadPool& operator=(const ThreadPool&) = delete;
        ThreadPool& operator=(ThreadPool&&) = delete;

    public:

        enum class DestroyType : std::uint8_t {
            SMOOTH,
            SHARP
        };

        ThreadPool();
        ThreadPool(const DestroyType destroy_type);
        ThreadPool(const std::uint32_t concurency);
        ThreadPool(const std::uint32_t concurency, const DestroyType destroy_type);
        ~ThreadPool();

        void Wait();
        void Pause();
        void Continue();
        void ClearTasks();

        void Reset();
        void Reset(const DestroyType destroy_type);
        void Reset(const std::uint32_t concurency);
        void Reset(const std::uint32_t concurency, const DestroyType destroy_type);

        void AddSyncTask(std::unique_ptr<Task> task);
        void AddAsyncTask(std::unique_ptr<Task> task);

        template<typename F, typename...Args>
        auto AddSyncTask(F&& job, Args&&... args) -> std::future<std::invoke_result_t<F, Args...>>;

        template<typename F, typename...Args>
        void AddAsyncTask(F&& job, Args&&... args);

    private:

        DestroyType destroy_type_;

        std::unique_ptr<std::thread[]> threads_;
        std::deque<std::unique_ptr<Task>> tasks_;

        std::uint32_t threads_count_;
        std::uint64_t tasks_running_;

        bool working_;
        bool paused_;
        bool waiting_;

        mutable std::mutex tasks_mutex_;

        std::condition_variable tasks_available_cv_;
        std::condition_variable tasks_done_cv_;

        [[nodiscard]] std::uint32_t ChooseThreadsCount_(const std::uint32_t threads_count) const noexcept;
        void CreateThreads_();
        void StopThreads_();
        void DestroyThreads_();
        void Finish_();
        void Process_(const std::uint32_t thread_index);

    };

    //////////////////////////////////////////////////////////////////////////////////
    // ThreadPool class defenition (template methods)
    ////////////////////////////////////////////////////////////////////////////////

    template<typename F, typename...Args>
    auto ThreadPool::AddSyncTask(F&& job, Args&&... args) -> std::future<std::invoke_result_t<F, Args...>> {
        std::unique_ptr<Task> task_ptr(std::make_unique<Task>());
        auto result = task_ptr->SetSyncJob(std::move(job), std::move(args)...);
        {
            const std::scoped_lock tasks_lock(tasks_mutex_);
            tasks_.push_back(std::move(task_ptr));
        }
        tasks_available_cv_.notify_one();
        return std::move(result);
    }

    template<typename F, typename...Args>
    void ThreadPool::AddAsyncTask(F&& job, Args&&... args) {
        std::unique_ptr<Task> task_ptr(std::make_unique<Task>());
        task_ptr->SetAsyncJob(std::move(job), std::move(args)...);
        {
            const std::scoped_lock tasks_lock(tasks_mutex_);
            tasks_.push_back(std::move(task_ptr));
        }
        tasks_available_cv_.notify_one();
    }

}

#endif // INCLUDE_GUARD_THREADPOOL_HPP

threadpool.cpp

#include <utility>
#include <threadpool.hpp>

namespace vsock {

    ThreadPool::ThreadPool(const std::uint32_t concurency, const DestroyType destroy_type) :
        destroy_type_{ destroy_type },
        threads_{ std::make_unique<std::thread[]>(ChooseThreadsCount_(concurency)) },
        tasks_{ },
        threads_count_{ ChooseThreadsCount_(concurency) },
        tasks_running_{ 0 },
        working_{ false },
        paused_{ false },
        waiting_{ false }
    {
        CreateThreads_();
    }

    ThreadPool::ThreadPool() :
        ThreadPool(std::thread::hardware_concurrency(), DestroyType::SMOOTH)
    {}

    ThreadPool::ThreadPool(const DestroyType destroy_type) :
        ThreadPool(std::thread::hardware_concurrency(), destroy_type)
    {}

    ThreadPool::ThreadPool(const std::uint32_t concurency) :
        ThreadPool(concurency, DestroyType::SMOOTH)
    {}

    ThreadPool::~ThreadPool() {
        Finish_();
    }

    void ThreadPool::ClearTasks() {
        const std::scoped_lock tasks_lock(tasks_mutex_);
        tasks_.clear();
    }

    void ThreadPool::Reset() {
        Reset(std::thread::hardware_concurrency(), DestroyType::SMOOTH);
    }

    void ThreadPool::Reset(const DestroyType destroy_type) {
        Reset(std::thread::hardware_concurrency(), destroy_type);
    }

    void ThreadPool::Reset(const std::uint32_t concurency) {
        Reset(concurency, DestroyType::SMOOTH);
    }

    void ThreadPool::Reset(const std::uint32_t concurency, const DestroyType destroy_type) {
        std::unique_lock tasks_lock(tasks_mutex_);
        destroy_type_ = destroy_type;
        const bool was_paused = paused_;
        paused_ = true;
        tasks_lock.unlock();
        Finish_();
        threads_count_ = ChooseThreadsCount_(concurency);
        threads_ = std::make_unique<std::thread[]>(threads_count_);
        CreateThreads_();
        tasks_lock.lock();
        paused_ = was_paused;
    }

    void ThreadPool::AddSyncTask(std::unique_ptr<Task> task) {
        {
            const std::scoped_lock tasks_lock(tasks_mutex_);
            tasks_.push_back(std::move(task));
        }
        tasks_available_cv_.notify_one();
    }

    void ThreadPool::AddAsyncTask(std::unique_ptr<Task> task) {
        {
            const std::scoped_lock tasks_lock(tasks_mutex_);
            tasks_.push_back(std::move(task));
        }
        tasks_available_cv_.notify_one();
    }

    void ThreadPool::Wait() {
        std::unique_lock tasks_lock(tasks_mutex_);
        waiting_ = true;
        tasks_done_cv_.wait(
            tasks_lock,
            [this] {return (tasks_running_ == 0) && (paused_ || tasks_.empty());}
        );
        waiting_ = false;
    }

    void ThreadPool::Pause() {
        const std::scoped_lock tasks_lock(tasks_mutex_);
        paused_ = true;
    }

    void ThreadPool::Continue() {
        {
            const std::scoped_lock tasks_lock(tasks_mutex_);
            paused_ = false;
        }
        tasks_available_cv_.notify_all();
    }

    std::uint32_t ThreadPool::ChooseThreadsCount_(const std::uint32_t threads_count) const noexcept {
        if (threads_count > 0) {
            return threads_count;
        }
        if (std::thread::hardware_concurrency() > 0) {
            return std::thread::hardware_concurrency();
        }
        return 1;
    }

    void ThreadPool::CreateThreads_() {

        {
            const std::scoped_lock tasks_lock(tasks_mutex_);
            tasks_running_ = threads_count_;
            working_ = true;
        }

        for (std::uint32_t index = 0; index < threads_count_; ++index) {
            threads_[index] = std::thread(&ThreadPool::Process_, this, index);
        }

    }

    void ThreadPool::StopThreads_() {
        {
            const std::scoped_lock tasks_lock(tasks_mutex_);
            working_ = false;
        }
        tasks_available_cv_.notify_all();
        for (std::uint32_t i = 0; i < threads_count_; ++i) {
            threads_[i].join();
        }
    }

    void ThreadPool::DestroyThreads_() {
        ClearTasks();
        StopThreads_();
    }

    void ThreadPool::Finish_() {
        if (destroy_type_ == DestroyType::SHARP) {
            DestroyThreads_();
        }
        else {
            Wait();
            StopThreads_();
        }
    }

    void ThreadPool::Process_([[maybe_unused]] const std::uint32_t thread_index) {
        std::unique_lock tasks_lock(tasks_mutex_);
        while (true) {
            --tasks_running_;
            tasks_lock.unlock();
            if (waiting_ && tasks_running_ == 0 && (paused_ || tasks_.empty())) {
                tasks_done_cv_.notify_all();
            }
            tasks_lock.lock();
            tasks_available_cv_.wait(tasks_lock,
                [this] {
                return !(paused_ || tasks_.empty()) || !working_;
            });
            if (!working_) {
                break;
            }
            std::unique_ptr<Task> task = std::exchange(tasks_.front(), {});
            tasks_.pop_front();
            ++tasks_running_;
            tasks_lock.unlock();
            bool not_finished = (*task)();
            if (not_finished) {
                tasks_lock.lock();
                tasks_.push_back(std::move(task));
                tasks_lock.unlock();
            }
            tasks_lock.lock();
        }
    }

}

main.cpp (tests)

#include <atomic>
#include <numeric>
#include <string>
#include <thread>
#include <cstddef>
#include <chrono>
#include <iostream>
#include <random>
#include <list>
#include <mutex>
#include <threadpool.hpp>

using namespace std;
using namespace vsock;
using namespace chrono;

static std::random_device dev;
static std::mt19937 rng(dev());

static std::mutex mtx_;

int RandomN(const int from, const int to) {
    std::uniform_int_distribution<std::mt19937::result_type> dist(from, to);
    return dist(rng);
}

std::size_t HardTest2(std::size_t size) {
    std::size_t i, num = 1, primes = 0;
    while (num <= size) {
        i = 2;
        while (i <= num) {
            if (num % i == 0)
                break;
            i++;
        }
        if (i == num)
            primes++;
        num++;
    }
    return primes;
}

bool HardTest1(std::size_t size) {
    std::vector<int> arr(size);
    std::iota(arr.begin(), arr.end(), 1);
    vector<int> res;
    for (std::size_t i : arr) {
        res.insert(res.begin(), i);
    }
    arr.clear();
    for (std::size_t i : res) {
        arr.insert(arr.begin(), i);
    }
    return true;
}

void PrintTaskWithID(int id, const int sleep_from, const int sleep_to) {
    int time = RandomN(sleep_from, sleep_to);

    mtx_.lock();
    cout << "PrintTaskWithID: task# " << id << " will sleep " << time << endl;
    mtx_.unlock();

    std::this_thread::sleep_for(std::chrono::milliseconds(time));

    mtx_.lock();
    cout << "PrintTaskWithID: task# " << id << " waked up after " << time << endl;
    mtx_.unlock();
}


void PrintTask(const int sleep_from, const int sleep_to) {
    int time = RandomN(sleep_from, sleep_to);

    mtx_.lock();
    cout << "PrintTask: " << " will sleep " << time << endl;
    mtx_.unlock();

    std::this_thread::sleep_for(std::chrono::milliseconds(time));

    mtx_.lock();
    cout << "PrintTask: " << " waked up after " << time << endl;
    mtx_.unlock();
}

int main() {


    cout << "\n=========================================================================\n"s;
    ThreadPool pool;
    {
        cout << "Test #L1: -------------------\n";
        pool.AddAsyncTask([](int a, int b) {
            cout << "a + b = " << (a + b) << "\n";
        }, 10, 20);
        pool.Wait();
    }

    {
        cout << "Test #L2: -------------------\n";

        cout << "Result will be in 10 msec\n";

        auto result = pool.AddSyncTask([](int a, int b) {
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
            return a * b;
        }, 6, 10);

        cout << "a * b = " << result.get() << "\n";
        pool.Wait();
    }

    {
        cout << "Test #L3: -------------------\n";

        int val = 10;

        auto result = pool.AddSyncTask([](int& a) {
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
            a = a * 10;
        }, std::ref(val));

        cout << "val = " << val << '\n';
        cout << "result.wait();\n";
        result.wait();

        cout << "val = " << val << '\n';
        pool.Wait();
    }

    {
        cout << "Test #L4: -------------------\n";

        int val = 10;

        auto result = pool.AddSyncTask([](int& a, ThreadPool* pool) {
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
            a = a * 10;
            auto res = pool->AddSyncTask([](int& b) {
                b = b * 5;
            }, std::ref(a));
            res.wait();
            auto res2 = pool->AddSyncTask([](int& b) -> int {
                return b * 10;
            }, std::ref(a));
            cout << "thread> a = " << a << '\n';
            cout << "thread> res2 = " << res2.get() << '\n';
        }, std::ref(val), &pool);

        cout << "val = " << val << '\n';
        cout << "result.wait();\n";
        result.wait();

        cout << "val = " << val << '\n';
        pool.Wait();
    }

    {
        cout << "Test #L5: -------------------\n";
        std::vector<int> vec;
        std::atomic_bool inited = false;
        std::atomic_bool processed = false;

        pool.AddAsyncTask([](std::atomic_bool& p) {
            while (!p);
            cout << "processed!\n";
        }, std::ref(processed));

        auto fut1 = pool.AddSyncTask([](std::atomic_bool& i) -> int {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            i = true;
            return 100;
        }, std::ref(inited));

        auto fut2 = pool.AddSyncTask([](std::vector<int>& v, std::future<int>& r, std::atomic_bool& i) {
            int count = r.get();
            cout << "thread1> inited = " << i << '\n';
            cout << "thread1> count = " << count << '\n';
            v.resize(count);
            for (int z = 0; z < count; ++z) {
                v[z] = z;
            }
            cout << "thread1> v.size = " << v.size() << '\n';

        }, std::ref(vec), std::ref(fut1), std::ref(inited));

        pool.AddAsyncTask([](std::atomic_bool& i, std::future<void>& f, std::vector<int>& v, std::atomic_bool& p) {
            while (!i);
            cout << "thread2> inited!\n";
            f.wait();
            cout << "thread2> fut received!\n";
            int c = v.size() * 2;
            v.resize(c);
            for (int z = 0; z < c; ++z) {
                v[z] = 999;
            }
            cout << "thread2> v.size = " << v.size() << '\n';
            cout << "thread2> v[10] = " << v[10] << '\n';
            p = true;
        }, std::ref(inited), std::ref(fut2), std::ref(vec), std::ref(processed));

        pool.Wait();
    }

    {
        cout << "Test #L6: -------------------\n";
        std::unique_ptr<Task> task = std::make_unique<Task>();
        task->AddVariables(0, 10);
        task->AddVariables("hello"s);
        task->SetCondition([](Task& task) -> bool {
            const int it = std::any_cast<int>(task.GetVariable(0));
            const int to = std::any_cast<int>(task.GetVariable(1));
            return it < to;
        }, std::ref(*task));
        task->SetLoopJob([](Task& task) -> void {
            int& it = std::any_cast<int&>(task.GetVariable(0));
            const std::string str = std::any_cast<std::string>(task.GetVariable(2));
            cout << "loop #" << it << ": " << str << '\n';
            ++it;
            std::this_thread::sleep_for(std::chrono::milliseconds(30));
        }, std::ref(*task));
        pool.AddAsyncTask(std::move(task));
        pool.Wait();
    }

    {
        cout << "Test #L7: -------------------\n";
        std::unique_ptr<Task> task = std::make_unique<Task>();
        std::atomic_bool c1{ true };
        task->AddVariables(std::ref(c1));
        task->SetCondition([](Task& task) -> bool {
            std::atomic_bool& c = std::any_cast<std::reference_wrapper<std::atomic_bool>>(task.GetVariable(0));
            return c;
        }, std::ref(*task));
        task->SetLoopJob([](Task& task, ThreadPool* pool) -> void {
            std::atomic_bool& c = std::any_cast<std::reference_wrapper<std::atomic_bool>>(task.GetVariable(0));
            mtx_.lock();
            cout << "loop start. c = " << c << "\n";
            mtx_.unlock();
            auto ret = pool->AddSyncTask([]() -> pair<int, int> {
                return std::make_pair(::RandomN(1, 100), ::RandomN(1, 100));
            });
            auto res = ret.get();
            cout << "first = " << res.first << ", second = " << res.second << "\n";
            if (res.first > res.second) {
                c = false;
            }
            cout << "loop end. c = " << c << "\n";
        }, std::ref(*task), & pool);
        pool.AddAsyncTask(std::move(task));
        pool.Wait();
    }

    {
        cout << "Test #G1: -------------------\n";
        for (int z = 0; z < 10; ++z) {
            pool.AddAsyncTask(PrintTask, 1, 1000);
        }
        pool.Wait();
    }

    {
        cout << "Test #G2: -------------------\n";
        for (int z = 0; z < 10; ++z) {
            pool.AddAsyncTask(PrintTaskWithID, z, 1, 1000);
        }
        pool.Wait();
    }

    {
        cout << "Test #G3: -------------------\n";
        std::vector<std::future<bool>> results;
        for (int z = 0; z < 10; ++z) {
            results.emplace_back(pool.AddSyncTask(HardTest1, 100000));
        }
        pool.AddAsyncTask([&](std::vector<std::future<bool>>& r) {
            auto zero = std::chrono::seconds(0);
            while (!r.empty()) {
                for (auto it = r.begin(); it != r.end();) {
                    if (it->wait_for(zero) == std::future_status::ready) {
                        it = r.erase(it);
                        mtx_.lock();
                        cout << "result is ready\n";
                        mtx_.unlock();
                    }
                    else {
                        ++it;
                    }
                }
            }
            cout << "all done!\n";
        }, std::ref(results));
        cout << "waiting results...\n";
        pool.Wait();
    }


    {
        cout << "Test #G4: -------------------\n";
        using results_t = std::vector<pair<std::size_t, std::future<std::size_t>>>;
        results_t results;
        for (int z = 0; z < 40; ++z) {
            std::uniform_int_distribution<std::mt19937::result_type> size(10, 100000);
            results.emplace_back(std::move(make_pair(z, pool.AddSyncTask(HardTest2, size(rng)))));
        }
        pool.AddAsyncTask([&](results_t& r) {
            auto zero = std::chrono::seconds(0);
            std::vector<std::size_t> res_nums(r.size());
            while (!r.empty()) {
                for (auto it = r.begin(); it != r.end();) {
                    if (it->second.wait_for(zero) == std::future_status::ready) {
                        res_nums[it->first] = it->second.get();
                        mtx_.lock();
                        cout << "result #" << it->first << " is ready with value " << res_nums[it->first] << "\n";
                        mtx_.unlock();
                        it = r.erase(it);
                    }
                    else {
                        ++it;
                    }
                }
            }
        }, std::ref(results));
        cout << "waiting results...\n";
        pool.Wait();
    }


}

Answer 1

It looks overcomplicated

Your threadpool is far from simple. Do you need all this functionality? If not, I would apply the YAGNI principle and remove anything you don't need. If you do need all this functionality, try to avoid puttig too much functionality in a few classes, and instead prefer to have more simple classes that compose together. For example:

Create a thread-safe queue class

Your ThreadPool not only manages the threads, it also manages the tasks that the threads have to complete. You can separate these two things. In particular, create a thread-safe queue that can be used to store the list of tasks. This should greatly simplify the ThreadPool itself.

Consider not using Task

I see you spent a lot of effort on class Task. However, the threads processing tasks don't care about any of that functionality, they just want to invoke the task. If you didn't care about unfinished tasks, you could just store tasks as std::function<void()>. The caller can then add whatever function object they want to the queue. If they need to add variables and/or get a future, they can do that themselves. You could still make helper classes/functions to make that easier, but by removing the dependency of ThreadPool on Task, you simplify that code and you will actually make it more flexible.

Inefficient and unnecessary things

There are several things that you do in the code that, while looking reasonable at first glance, are actually unnecessary and/or inefficient.

First of all, if a class has non-copyable member variables, the whole class will be non-copyable by default as well. So you don't need to explicitly delete the copy constructor and copy assignment operator.

Related to this, the compiler will automatically generate a move constructor and move assignment operator that will move each member individually. This means you don't have to write those yourself if that's the only thing you are doing anyway. Not defining those in a source file means the compiler can inline those operations in other source files.

std::any is very flexible, but it comes with both CPU and memory overhead. Don't use it if it's not really necessary. Also note that std::any_casts might be "safe", but only in the sense that they will throw an exception if the data doesn't have the expected type.

There are unnecessary std::move()s, and they can actually make the compiler output worse code than when not using std::move(). Only use them on named variables, not on temporaries. For example, in this line:

const std::shared_ptr<bind_type> bind_fnc_ptr =
    std::make_shared<bind_type>(
        std::move(std::bind(std::move(job), std::move(args)...))
    );

The outer std::move() is redundant. Also, don't std::move() when returning variables declared inside a function, this will prevent return-value optimization (RVO). You do this for example in AddSyncTask().

Containers already store values, there is often no need to store std::unique_ptrs in a container. So instead of:

std::deque<std::unique_ptr<Task>> tasks_;

Just use:

std::deque<Task> tasks_;

And instead of a std::unique_ptr storing an array, also just use a container:

std::deque<std::thread> threads_;

You should almost never have to make member variables mutable. Since none of the member functions of ThreadPool are const-qualified, you can't even use a const ThreadPool, so mutable doesn't make any sense.

Why are sync_task_, async_task_ and condition_ stored in a std::unqiue_ptr? That shouldn't be necessary, as std::packaged_tasks and std::functions can be moved to anyway.

Why does Process()_ take a member variable thread_index? It doesn't seem to be used at all.

Use std::size_t for sizes, counts and indices

Instead of using std::uint*_ts, prefer to use std::size_t for anything that holds a size, count or index. This type is guaranteed to be large enough to be able to address anything that fits in memory.

I also don't see why you made threads_count_ a std::uint32_t but tasks_running_ as std::uint64_t.

Incorrect locking in Process_()

Whenever you have a variable that is shared between threads, you must always use a lock to access them. Reading or writing them without holding a lock will virtually guarantee you will have a race condition somewhere. Consider Process_(), where you unlock the mutex, then check some shared variables, and based on that you call notify_all(). Since these variables can change between the check and the subsequent code, this might cause notify_all() to not be called when other threads might depend on it.

Note that it's perfectly fine to call notify_all() even with a lock held, and especially if you are going to relock the mutex anyway, there is no performance benefit from not keeping the mutex locked.

Use std::forward() instead of std::move() where appropriate

When you have forwarding references like Args&&... args, use std::forward<Args>(args)... instead of std::move(args)... when passing them. This also goes for single arguments like F&&. See this StackOverflow question for more information.

Avoid manual for-loops

I see a lot of for-loops with counters that could be replaced with range for-loops. There are also some loops where you work with iterators in order to be able to remove items from the container you are looping over. Consider using std::erase_if() instead. There is less chances of making mistakes this way.