Not so Simple Thread Pool (part 2)

Question

Previous version of my ThreadPool is here

This one is improved and updated using advices:

• added variables list (vector) to not use std::any. it's very basic and simple.
• added thread-safe queue
• !main.cpp is for testing classes only
• some other small fixes

1. don't understand what is incorrect in Process_() method
2. std::unique_ptr in tasks queue is used to send Task object in their lambda (examples in main.cpp)
3. std::unique_ptrstd::thread[] threads_; is just for safe memory free on destruction

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

threadpool.hpp

#ifndef INCLUDE_GUARD_THREADPOOL_HPP
#define INCLUDE_GUARD_THREADPOOL_HPP

#include <thread>
#include <mutex>
#include <deque>
#include <condition_variable>

#include <task.hpp>
#include <queue.hpp>

namespace vsock {

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

    class ThreadPool {
    public:

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

    public:

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

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

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

        void Reset();
        void Reset(const DestroyType destroy_type);
        void Reset(const std::size_t concurency);
        void Reset(const std::size_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_;
        TaskQueue tasks_;

        std::size_t threads_count_;
        std::size_t tasks_running_;

        bool working_;
        bool paused_;
        bool waiting_;

        std::mutex tasks_mutex_;

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

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

    };

    //////////////////////////////////////////////////////////////////////////////////
    // 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::forward<F>(job), std::forward<Args>(args)...);
        tasks_.PushBack(std::move(task_ptr));
        tasks_available_cv_.notify_one();
        return 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::forward<F>(job), std::forward<Args>(args)...);
        tasks_.PushBack(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::size_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::size_t concurency) :
        ThreadPool(concurency, DestroyType::SMOOTH)
    {}

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

    void ThreadPool::ClearTasks() noexcept {
        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::size_t concurency) {
        Reset(concurency, DestroyType::SMOOTH);
    }

    void ThreadPool::Reset(const std::size_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) {
        tasks_.PushBack(std::move(task));
        tasks_available_cv_.notify_one();
    }

    void ThreadPool::AddAsyncTask(std::unique_ptr<Task> task) {
        tasks_.PushBack(std::move(task));
        tasks_available_cv_.notify_one();
    }

    void ThreadPool::Wait() noexcept {
        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() noexcept {
        const std::scoped_lock tasks_lock(tasks_mutex_);
        paused_ = true;
    }

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

    std::size_t ThreadPool::ChooseThreadsCount_(const std::size_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::size_t index = 0; index < threads_count_; ++index) {
            threads_[index] = std::thread(&ThreadPool::Process_, this);
        }

    }

    void ThreadPool::StopThreads_() {
        {
            const std::scoped_lock tasks_lock(tasks_mutex_);
            working_ = false;
        }
        tasks_available_cv_.notify_all();
        for (std::size_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_() {
        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;
            }

            ++tasks_running_;
            

            std::unique_ptr<Task> task;
            tasks_.PopFront(task);
            tasks_lock.unlock();
            bool not_finished = (*task)();
            tasks_lock.lock();
            if (not_finished) {
                tasks_.PushBack(std::move(task));
            }

        }
    }

}

task.hpp

#ifndef INCLUDE_GUARD_TASK_HPP
#define INCLUDE_GUARD_TASK_HPP

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

#include <varlist.hpp>

namespace vsock {

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

    class Task {
    private:
        enum class TaskType : std::uint8_t { ASYNC, SYNC, LOOP };
    public:

        Task() = default;

        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);

        bool IsVoidResult();

        bool operator()();

    public:

        VarList vars;

    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 };

    };

    //////////////////////////////////////////////////////////////////////////////////
    // 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::bind(std::forward<F>(job), std::forward<Args>(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::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::bind(std::forward<F>(job), std::forward<Args>(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::bind(std::forward<F>(condition), std::forward<Args>(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::bind(std::forward<F>(loop), std::forward<Args>(args)...)
        );
    }

}

#endif // INCLUDE_GUARD_TASK_HPP

task.cpp

#include <task.hpp>

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

namespace vsock {

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

    Task::Task(Task&& other) :
        vars(std::move(other.vars)),
        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_,{}) }
    {}

    Task& Task::operator=(Task&& other) {
        if (this != &other) {
            vars = std::move(other.vars);
            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_, {});
        }
        return *this;
    }

    bool Task::IsVoidResult() {
        return is_void_;
    }

    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;
            }
        }
    }

}

queue.hpp

#ifndef INCLUDE_GUARD_QUEUE_HPP
#define INCLUDE_GUARD_QUEUE_HPP

#include <deque>
#include <mutex>
#include <memory>
#include <utility>

#include <task.hpp>

namespace vsock {

    //////////////////////////////////////////////////////////////////////////////////
    // TaskQueue class declaration
    ////////////////////////////////////////////////////////////////////////////////


    class TaskQueue : private std::deque<std::unique_ptr<Task>> {
    private:

        friend class ThreadPool;

        using value_t = std::unique_ptr<Task>;
        using deque_t = std::deque<value_t>;

    private:

        void PushBack(value_t&& task);
        void Clear() noexcept;
        bool Empty() const noexcept;
        void PopFront(value_t& task) noexcept;

    private:

        mutable std::mutex mtx_;

    };

}

#endif

queue.cpp

#include <queue.hpp>

namespace vsock {

    //////////////////////////////////////////////////////////////////////////////////
    // TaskQueue class defenition
    ////////////////////////////////////////////////////////////////////////////////    

    void TaskQueue::PushBack(value_t&& task) {
        const std::scoped_lock rw_lock(mtx_);
        std::deque<std::unique_ptr<Task>>::push_back(std::move(task));
    }

    void TaskQueue::Clear() noexcept {
        const std::scoped_lock rw_lock(mtx_);
        std::deque<std::unique_ptr<Task>>::clear();
    }

    bool TaskQueue::Empty() const noexcept {
        const std::scoped_lock rw_lock(mtx_);
        return std::deque<std::unique_ptr<Task>>::empty();
    }

    void TaskQueue::PopFront(value_t& task) noexcept {
        const std::scoped_lock rw_lock(mtx_);
        task = std::move(std::deque<std::unique_ptr<Task>>::front());
        std::deque<std::unique_ptr<Task>>::pop_front();
    }

}

varlist.hpp

#ifndef INCLUDE_GUARD_VARLIST_HPP
#define INCLUDE_GUARD_VARLIST_HPP

#include <varnode.hpp>

namespace vsock {

    //////////////////////////////////////////////////////////////////////////////////
    // VarList class declaration
    ////////////////////////////////////////////////////////////////////////////////

    class VarList {
    public:

        VarList() = default;
        VarList(VarList&&) = default;
        VarList& operator=(VarList&&) = default;

        template<typename T>
        void Add(T&& var);

        template<typename T>
        T& Emplace(T&& var);

        template<typename T>
        T& Get(std::size_t index);

        template<typename T>
        const T& Get(std::size_t index) const;

        void Remove(std::size_t index);

        void Clear() noexcept;
        bool Empty() const noexcept;
        std::size_t Size() const noexcept;

    private:

        std::unique_ptr<VarNode[]> nodes_{ nullptr };
        std::size_t size_{ 0 };
        std::size_t capacity_{ 0 };

        void Resize_(std::size_t new_capacity);

    };

    //////////////////////////////////////////////////////////////////////////////////
    // VarList class defenition (template methods)
    ////////////////////////////////////////////////////////////////////////////////    

    template<typename T>
    inline void VarList::Add(T&& var) {
        Resize_(size_ + 1);
        (nodes_.get() + (size_++))->Put(std::forward<T>(var));
    }

    template<typename T>
    inline T& VarList::Emplace(T&& var) {
        Resize_(size_ + 1);
        return (nodes_.get() + (size_++))->Emplace(std::forward<T>(var));
    }

    template<typename T>
    inline T& VarList::Get(std::size_t index) {
        if (index >= size_) {
            throw std::out_of_range("Out of range");
        }
        return ((nodes_).get() + index)->Get<T>();
    }

    template<typename T>
    inline  const T& VarList::Get(std::size_t index) const {
        if (index >= size_) {
            throw std::out_of_range("Out of range");
        }
        return ((nodes_).get() + index)->Get<T>();
    }

}

#endif // INCLUDE_GUARD_VARLIST_HPP

varlist.cpp

#include <varlist.hpp>

namespace vsock {

    //////////////////////////////////////////////////////////////////////////////////
    // VarList class defenition
    ////////////////////////////////////////////////////////////////////////////////

    void VarList::Remove(std::size_t index) {
        if (index >= size_) {
            throw std::out_of_range("Out of range");
        }
        bool removed{ false };
        for (auto it = ((nodes_).get() + index); it != ((nodes_).get() + size_); ++it) {
            *it = std::move(*(it + 1));
            removed = true;
        }
        if (removed) {
            --size_;
        }
    }

    void VarList::Clear() noexcept {
        if (Empty()) {
            return;
        }
        nodes_.reset();
        capacity_ = 0;
        size_ = 0;
    }

    bool VarList::Empty() const noexcept {
        return size_ == 0;
    }

    std::size_t VarList::Size() const noexcept {
        return size_;
    }

    void VarList::Resize_(std::size_t new_capacity) {
        if (new_capacity <= capacity_) {
            return;
        }

        std::size_t future_capacity = std::max(new_capacity, capacity_ * 2);
        std::unique_ptr<VarNode[]> new_nodes = std::make_unique<VarNode[]>(future_capacity);

        auto from = new_nodes.get();
        for (auto it = nodes_.get(); it != (nodes_.get() + size_); ++it, ++from) {
            *from = std::move(*it);
        }

        capacity_ = future_capacity;
        nodes_.reset();
        nodes_ = std::move(new_nodes);
    }

}

varnode.hpp

#ifndef INCLUDE_GUARD_VARNODE_HPP
#define INCLUDE_GUARD_VARNODE_HPP

#include <utility>
#include <memory>
#include <stdexcept>

namespace vsock {

    //////////////////////////////////////////////////////////////////////////////////
    // VarNode class declaration
    ////////////////////////////////////////////////////////////////////////////////

    class VarNode {
    public:

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

    public:

        VarNode();
        VarNode(VarNode&& other);
        VarNode& operator=(VarNode&& rhs);
        ~VarNode();

        template<typename T>
        VarNode(T&& data);

        template<typename T>
        void Put(T&& data);

        template<typename T>
        T& Emplace(T&& data);

        template<typename T>
        const T& Get() const;

        template<typename T>
        T& Get();

        void Drop();
        bool Empty() const noexcept;

    private:

        template<typename T>
        void* Put_(T&& data);

        template<typename T>
        static void Delete_(void* ptr);

        void Swap_(VarNode& other) noexcept;

    private:

        void* data_;
        void (*delete_fnc_)(void*);
        std::size_t hash_code_;

    };

    //////////////////////////////////////////////////////////////////////////////////
    // VarNode class defenition (template methods)
    ////////////////////////////////////////////////////////////////////////////////

    template<typename T>
    VarNode::VarNode(T&& data) :
        data_{ Put_(std::forward<T>(data)) }
    {}

    template<typename T>
    inline void VarNode::Put(T&& data) {
        data_ = Put_(std::forward<T>(data));
    }

    template<typename T>
    inline T& VarNode::Emplace(T&& data) {
        Put(std::forward<T>(data));
        return *(reinterpret_cast<T*>(data_));
    }

    template<typename T>
    inline const T& VarNode::Get() const {
        if (typeid(T).hash_code() != hash_code_) {
            throw std::runtime_error("bad type");
        }
        return *(reinterpret_cast<T*>(data_));
    }

    template<typename T>
    inline T& VarNode::Get() {
        if (typeid(T).hash_code() != hash_code_) {
            throw std::runtime_error("bad type");
        }      
        return *(reinterpret_cast<T*>(data_));
    }

    template<typename T>
    inline void* VarNode::Put_(T&& data) {
        void* result = reinterpret_cast<void*>(new T(std::forward<T>(data)));
        Drop();
        hash_code_ = typeid(T).hash_code();
        delete_fnc_ = VarNode::Delete_<T>;
        return result;
    }

    template<typename T>
    inline void VarNode::Delete_(void* ptr) {
        delete reinterpret_cast<T*>(ptr);
        ptr = nullptr;
    }

}

#endif // INCLUDE_GUARD_VARNODE_HPP

varnode.cpp

#include <varnode.hpp>

namespace vsock {

    //////////////////////////////////////////////////////////////////////////////////
    // VarNode class defenition
    ////////////////////////////////////////////////////////////////////////////////
 
    VarNode::VarNode() :
        data_{ nullptr },
        delete_fnc_{ nullptr },
        hash_code_{ 0 }
    {}

    VarNode::VarNode(VarNode&& other) :
        data_{ std::exchange(other.data_,nullptr) },
        delete_fnc_{ std::exchange(other.delete_fnc_,nullptr) },
        hash_code_{ std::exchange(other.hash_code_,0) }
    {}

    VarNode& VarNode::operator=(VarNode&& rhs) {
        if (&rhs != this) {
            auto old = VarNode(std::move(rhs));
            Swap_(old);
        }
        return *this;
    }

    VarNode::~VarNode() {
        Drop();
    }

    void VarNode::Drop() {
        if (Empty()) {
            return;
        }
        delete_fnc_(data_);
        data_ = nullptr;
        delete_fnc_ = nullptr;
        hash_code_ = 0;
    }

    bool VarNode::Empty() const noexcept {
        return !data_;
    }

    void VarNode::Swap_(VarNode& other) noexcept {
        std::swap(other.data_, data_);
        std::swap(other.delete_fnc_, delete_fnc_);
        std::swap(other.hash_code_, hash_code_);
    }

}

main.cpp

#include <atomic>
#include <numeric>
#include <string>
#include <thread>
#include <cstddef>
#include <chrono>
#include <iostream>
#include <random>
#include <list>
#include <mutex>
#include <threadpool.hpp>
#include <queue.hpp>
#include <varlist.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();
}

void RunTests() {
    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->vars.Add(0);
        task->vars.Add(10);
        task->vars.Add("hello"s);
        task->SetCondition([](Task& task) -> bool {
            const int it = task.vars.Get<int>(0);
            const int to = task.vars.Get<int>(1);
            return it < to;
        }, std::ref(*task));
        task->SetLoopJob([](Task& task) -> void {
            int& it = task.vars.Get<int>(0);
            const std::string str = task.vars.Get<std::string>(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->vars.Add(std::ref(c1));
        task->SetCondition([](Task& task) -> bool {
            std::atomic_bool& c = task.vars.Get<std::reference_wrapper<std::atomic_bool>>(0);
            return c;
        }, std::ref(*task));
        task->SetLoopJob([](Task& task, ThreadPool* pool) -> void {
            std::atomic_bool& c = task.vars.Get<std::reference_wrapper<std::atomic_bool>>(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 #L8: -------------------\n";
        auto work = [](Task& task, const std::string& str, int& r, ThreadPool* p) {
            cout << "Hello, " << str << "!\n";
            task.vars.Add(str + "123"s);
            std::unique_ptr<Task> t(std::make_unique<Task>());
            r = 10;
            cout << "r = " << r << '\n';
            t->vars.Add(std::ref(r));
            auto subjob = [](Task& ts) {
                int& v = ts.vars.Get<std::reference_wrapper<int>>(0);
                v *= 2;
                cout << "v = " << v << '\n';
            };
            t->SetAsyncJob(subjob, std::ref(*t));
            p->AddAsyncTask(std::move(t));
        };
        int result{ 0 };
        std::unique_ptr<Task> task(std::make_unique<Task>());
        task->SetAsyncJob(work, std::ref(*task), "Bro"s, std::ref(result), &pool);

        pool.AddAsyncTask(std::move(task));
        pool.Wait();
        cout << "result = " << result << '\n';
    }

    {
        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 < 20; ++z) {
            results.emplace_back(pool.AddSyncTask(HardTest1, 10000));
        }
        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 < 50; ++z) {
            std::uniform_int_distribution<std::mt19937::result_type> size(10, 30000);
            mtx_.lock();
            int value = size(rng);
            cout << "task #" << z << " posted with value " << value << "\n";
            mtx_.unlock();
            results.emplace_back(std::move(make_pair(z, pool.AddSyncTask(HardTest2, value))));
        }
        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();
    }

    {
        cout << "Test #G5: -------------------\n";
        std::unique_ptr<Task> main_task = std::make_unique<Task>();
        std::vector<std::string> strs;
        std::condition_variable done;
        bool processed{ false };
        strs.push_back("initial str");
        strs.push_back("another initial str");
        cout << "initial content:\n";
        for (const auto& s : strs) {
            cout << s << '\n';
        }
        std::function<int(int, int)> calc = [](const int a, const int b) {
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
            return a * b;
        };
        main_task->vars.Add(std::ref(strs));
        main_task->vars.Add(std::move(calc));
        main_task->vars.Add(std::ref(done));
        main_task->SetAsyncJob([](ThreadPool& pool, Task& task, const int count, bool& processed) {
            std::vector<std::string>& strs = task.vars.Get<std::reference_wrapper<std::vector<std::string>>>(0);
            auto fnc = task.vars.Get<std::function<int(int, int)>>(1);
            std::condition_variable& done = task.vars.Get<std::reference_wrapper<std::condition_variable>>(2);
            for (int i = 0; i < count; ++i) {
                int a = RandomN(1, 1000);
                int b = RandomN(1, 1000);
                std::future<int> res = pool.AddSyncTask(fnc, a, b);
                strs.push_back("calculating "s + std::to_string(a) + " * " + std::to_string(b) + " = " + std::to_string(res.get()));
            }
            processed = true;
            done.notify_one();
        }, std::ref(pool), std::ref(*main_task), 20, std::ref(processed));
        pool.AddAsyncTask(std::move(main_task));

        std::unique_ptr<Task> wait_task = std::make_unique<Task>();
        wait_task->vars.Add(std::ref(done));
        wait_task->SetAsyncJob([](Task& task, const std::vector<std::string>& strs, bool& processed) {
            std::condition_variable& done = task.vars.Get<std::reference_wrapper<std::condition_variable>>(0);
            std::mutex mtx;
            std::unique_lock lock(mtx);
            while (!processed) {
                done.wait(lock);
            }
            cout << "processed content:\n";
            for (const auto& s : strs) {
                cout << s << '\n';
            }

        }, std::ref(*wait_task), std::ref(strs), std::ref(processed));

        pool.AddAsyncTask(std::move(wait_task));

        pool.Wait();
    }

    {
        cout << "Test #G5: -------------------\n";
        std::unique_ptr<Task> main_task = std::make_unique<Task>();
        int val{ 10 };
        main_task->vars.Add(std::ref(val));
        main_task->SetAsyncJob([](Task& task, ThreadPool& pool) {
            std::unique_ptr<Task> second_task = std::make_unique<Task>();
            int& val = second_task->vars.Emplace(std::ref(task.vars.Get<std::reference_wrapper<int>>(0)));
            ++val;
            cout << "inside 1 val: " << val << '\n';
            second_task->SetAsyncJob([](Task& task) {
                int& val = task.vars.Get<std::reference_wrapper<int>>(0);
                ++val;
                cout << "inside 2 val: " << val << '\n';
            }, std::ref(*second_task));
            pool.AddAsyncTask(std::move(second_task));
        }, std::ref(*main_task), std::ref(pool));
        pool.AddAsyncTask(std::move(main_task));
        pool.Wait();
        cout << "outside val: " << val << '\n';
    }

}

class Test {
public:
    int val{ 10 };
};

int main() {
    RunTests();

    cout << "Done!" << endl;
}

Answer 1

It's always great to see improvements being made based on reviews! There are still quite a few issues left though. I'll start with addressing the points you make in your question:

Either always or never guard something with a mutex

1. don't understand what is incorrect in Process_() method

Whenever you have a variable that you guard at some point with a mutex, if you access that variable somewhere else without holding the mutex, it has a 99.9% chance of being a bug, and if it's not, it's most likely a premature optimization. Consider these lines in Process_():

tasks_lock.unlock();
if (waiting_ && tasks_running_ == 0 && (paused_ || tasks_.Empty())) {
    tasks_done_cv_.notify_all();
}
tasks_lock.lock();

All the variables read in the if-condition are accessed elsewhere while holding tasks_lock, but here you have it unlocked. Consider another thread calling ThreadPool::Wait() at the same time the above lines are executed. Then it's possible that in Process_(), waiting_ is read before Wait() sets it to true. This means you will not send a notification. Wait() will then proceed to call wait(), but this will then potentially block indefinitely. Similar issues probably exist with the other variables.

There is also no performance benefit to unlocking tasks_lock here. While you'll often find a recommendation to call notify_one()/notify_all() without holding the mutex associated with a condition variable, that assumes that you will not lock that mutex for a while. This avoids the case where the waiting thread wakes up, finds the mutex still locked, and has to go back to sleep immediately. Since you immediately relock the mutex after calling notify_all(), you will virtually ensure that the waiting threads wake up finding the mutex locked again. So just don't unlock it.

TaskQueue::Empty() is unsafe

While you lock the mutex so calling empty() on the actual queue is safe, the problem is that when this function returns, the mutex is released. Between the function returning and the caller looking at the result, another thread can come and either add something to the queue or pop it. Thus, the return value of TaskQueue::Empty() cannot be trusted. Since this makes the function useless, remove it.

What you want instead is to add a condition variable to TaskQueue so it can wait itself for tasks to be added to the queue. Of course, that brings the issue of how to handle destroying the queue when there are still waiters. However, that's the same issue as when destroying a ThreadPool, so a similar solution can be applied to TaskQueue.

Unnecessary use of std::unique_ptr

2. std::unique_ptr in TaskQueue is used to send Task object in their lambda (examples in main.cpp)

You can already move lambdas and Tasks around without std::unique_ptr. You should be able to make it so you can write code like this:

ThreadPool pool;
Task task;
task.SetStuff(…);
pool.AddASyncTask(std::move(task));

The ThreadPool will in turn std::move() the task into its own storage space (tasks_).

3. std::unique_ptr<std::thread[]> threads_ is just for safe memory free on destruction

But a std::vector or std::deque will also safely free memory on destruction. There is no reason to go so low level.

You can in fact remove all occurences of std::unique_ptr and std::make_unique() from your code. This will simplify your code a lot, as well as making it more efficient.

You are reinventing the wheel in several places

• added variables list (vector) to not use std::any.

But now you have VarNode, which is a reimplementation of std::any, and VarList, which is a reimplementation of std::vector<VarNode>. This is not an improvement, this is worse than what you had in the first version.

If you really need the features of std::any, then by all means keep using std::any. The point I was trying to make was that std::any has a lot of overhead (which you can see in your own implementation of VarNode), so I would try to avoid having to use it:

Do you really need all this functionality?

Again: your thread pool has a lot of bells and whistles that make it a lot more complicated than most other C++ thread pool implementations I've seen on Code Review. And I don't think you gain anything with this. I'll go over a few examples from your main(), and show how it could be rewritten to work with a threadpool that only had an void AddTask(std::function<void()> job).

In test #L1, the values passed as parameters are already known when adding the task. Instead of parameters, we can just capture those values:

pool.AddTask([a = 10, b = 20] {
    std::cout << "a + b = " << (a + b) << "\n";
});

In test #L3, we can also let the lambda capture the reference:

int val = 10;

pool.AddTask([&a = val] {
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    a = a * 10;
});

In test #L4, two things happen: a task can add new tasks to the pool itself, and it can return a future. Let's split these two things into two examples, first adding new tasks from within a task itself:

pool.AddTask([&pool] {
    pool.AddTask([] {
        std::cout << "Hello, ";
    });
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    pool.AddTask([] {
        std::cout << "World!\n";
    });
});

For returning a value via a future, that could be done by capturing a std::promise or std::packaged_task by value in the function passed to AddTask(). The problem is that std::promise and std::packaged_task are move-only, and std::function requires the task to be copyable. In C++23 this is solved with std::move_only_function:

ThreadPool::AddTask(std::move_only_function<void()> task);
…
std::promise<int> promise;
auto future = promise.get_future();
pool.AddTask([promise = std::move(promise)] mutable {
    promise.set_value(42);
});
std::cout << "result = " << future.get() << '\n';

This is the only point where I would say that adding an overload of AddTask to ThreadPool to support returning futures would be useful. It would basically take care of the above, so that you could write this instead:

auto future = pool.AddTask<int>([] {
    return 42;
});
std::cout << "result = " << future.get() << '\n';

In test #L6 you are using the SetCondition() and SetLoopJob() features of Task, with variables being passed between them. It's a lot of code, but if we just manually re-add tasks to the loop, it can be rewritten like so:

struct LoopJob {
    ThreadPool& pool;
    int it, to;

    void operator()() {
        std::cout << "loop #" << it << '\n';
        std::this_thread::sleep_for(std::chrono::milliseconds(30));
        if (it++ < to) {
            pool.AddTask(*this);
        }
    }
};
pool.AddTask(LoopJob{pool, 0, 10});

It also doesn't win any beauty awards, but it is significantly shorter than your test case, and doesn't rely on Tasks, VarNodes and all the other things needed to support those. You might be able to rewrite the functor as a recursive lambda instead in C++23 to save a few more lines.