2
\$\begingroup\$

There are things that I am not 100% sure. Is it possible that dangling references can be create in add_task method? Is it possible that this implementation will deadlock? I had to use shared_ptr in order to store packaged_task in the queue. Is there a better way to do it?

class ThreadPool {
public:
    ThreadPool(int number_of_workers) {
        for (int i=0; i<number_of_workers; ++i) {
            worker_threads.push_back(std::thread{std::bind(&ThreadPool::thread_func, this)});
        }
    }
    
    ~ThreadPool() = default;
    
    ThreadPool(const ThreadPool&) = delete;
    ThreadPool(ThreadPool&&) = delete;
    ThreadPool& operator=(const ThreadPool&) = delete;
    ThreadPool& operator=(ThreadPool&&) = delete;
    
    template<typename Func, typename... Args>
    decltype(auto) add_task(Func&& func, Args... args) {
        using ReturnType = decltype(func(args...));
        auto ptask = std::make_shared<std::packaged_task<ReturnType()>>(
            std::bind(std::forward<Func>(func), std::forward<Args>(args)...));
        auto fut = ptask->get_future();
        {
            std::unique_lock lk{pool_mutex};
            task_queue.push([pt = std::move(ptask)]() { (*pt)(); });
        }
        cond.notify_one();
        return fut;
    }

    void stop() {
        {
            std::unique_lock lk{pool_mutex};
            stopped = true;
        }
        cond.notify_all();
        for (auto& th : worker_threads) {
            if (th.joinable()) {
                th.join();
            }
        }
    }

private:
    void thread_func() {
        while(true) {
            std::unique_lock lk{pool_mutex};
            if (task_queue.empty() && stopped) {
                break;
            }

            if (task_queue.empty()) {
                cond.wait(lk, [this]() { return !task_queue.empty() || stopped; });
            }

            if (!task_queue.empty()) {
                auto callable = std::move(task_queue.front());
                task_queue.pop();
                lk.unlock();
                callable();
            }
            else if (stopped) {
                break;
            }
        }
    }

private:
    std::mutex pool_mutex;
    std::vector<std::thread> worker_threads;
    bool stopped = false;
    std::condition_variable cond;
    std::queue<std::function<void()>> task_queue;
};

Note, that add_task returns a future that can be used to wait for a specific task to complete.

\$\endgroup\$
1
\$\begingroup\$
~ThreadPool() = default;

We need to call stop() in the destructor!


template<typename Func, typename... Args>
decltype(auto) add_task(Func&& func, Args... args) {
    using ReturnType = decltype(func(args...));

For perfect forwarding we need to use && for args:

template<typename Func, typename... Args>
decltype(auto) add_task(Func&& func, Args&&... args) {
    using ReturnType = decltype(func(std::forward<Args>(args)...));

ThreadPool(int number_of_workers) {

We need to check that this number is greater than zero. Also it might be better to use an unsigned int to match the result of std::thread::hardware_concurrency().


void thread_func() {
    while(true) {
        std::unique_lock lk{pool_mutex};
        if (task_queue.empty() && stopped) {
            break;
        }
        ...

I'm not sure that we should wait for the task_queue to be empty before stopping. I'd suggest that stop() should prevent threads taking any more tasks from the queue (i.e. provide a way to "cancel" tasks that haven't started executing yet).

If we wait for the queue to be empty when calling stop(), we're waiting for all the tasks to complete, so calling stop() doesn't really do much. (The calling code is already in control of whether or not it adds more tasks and waits for futures).

        if (task_queue.empty()) {
            cond.wait(lk, [this]() { return !task_queue.empty() || stopped; });
        }

We don't really need to put cond.wait() in an if statement. It will check the condition before waiting.

I'd probably go for something more like:

void thread_func() {
    while(true) {
        
        std::unique_lock lk{pool_mutex};
        cond.wait(lk, [this]() { return !task_queue.empty() || stopped; });
        
        if (stopped)
            break;

        assert(!task_queue.empty());
        
        auto callable = std::move(task_queue.front());
        task_queue.pop();
        
        lk.unlock();
        
        callable();
    }
}
\$\endgroup\$
1
  • \$\begingroup\$ I agree with your critique. Thanks! \$\endgroup\$
    – pic11
    May 10 at 11:54

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.