2
\$\begingroup\$

Implemented a simple thread pool that executes task asyncly, and returns a simple future you can wait on. Can you please suggest improvements?

#pragma once

#include <iostream>
#include <atomic>
#include <cstdint>
#include <queue>
#include <thread>
#include <functional>
#include <mutex>
#include <memory>
#include <cstdio>


class thread_pool {
public:
    using task_t = std::function<int()>;

private:
    struct pool_task {
        pool_task(task_t callable) : m_callable(std::move(callable)), m_task_value(0) {}

        void wait() {
            std::unique_lock<std::mutex> lock(m_value_lock);
            m_value_cv.wait(lock, [this]() {
                return m_task_done.load();
            });
        }

        int value() {
            return m_task_value;
        }

        task_t m_callable;
        int m_task_value;
        std::mutex m_value_lock;
        std::condition_variable m_value_cv;
        std::atomic_bool m_task_done = false;
    };

public:
    /**
     *  Represents a future a client can use to receive
     *  the value returned from a submitted task.
     */
    struct future {
        future(std::shared_ptr<pool_task> task) : m_pool_task(task) {}

        int get() {
            m_pool_task->wait();
            return m_pool_task->value();
        }
    private:
        std::shared_ptr<pool_task> m_pool_task;
    };

    thread_pool(std::int32_t nthreads) : m_stopped(false) {
        for (int i = 0; i < nthreads; i++) {
            m_pool.emplace_back(std::thread([this]() {
                while (!m_stopped) {
                    execute_task();
                }
            }));
        }
    }

    thread_pool(const thread_pool& other) = default;
    thread_pool(thread_pool&& other) = default;

    ~thread_pool() {
        m_stopped = true;
        m_tasks_cv.notify_all();

        for (auto & thread : m_pool) {
            thread.join();
        }

    }

    future submit(task_t callable) {
        std::unique_lock<std::mutex> lock(m_tasks_lock);
        auto task = std::make_shared<pool_task>(callable);
        m_pool_tasks.emplace(task);

        return future(task);
    }

    /**
     * Stops all tasks.
     */
    void stop() {
        m_stopped = true;
        m_tasks_cv.notify_all();
    }

private:
    /**
     *  Execute a task fetched from the tasks queue.
     */
    void execute_task() {
        std::shared_ptr<pool_task> task = nullptr;
        {
            std::unique_lock<std::mutex> lock(m_tasks_lock);
            m_tasks_cv.wait(lock, [this]() {
                return m_stopped || !m_pool_tasks.empty();
            });

            if (m_stopped) {
                return;
            }

            task = m_pool_tasks.front();
            m_pool_tasks.pop();
        }

        /**
         *  Shouldn't happen, but just in case..
         */
        if (task == nullptr) {
            return;
        }

        task->m_task_value = task->m_callable();
        task->m_task_done = true;
        task->m_value_cv.notify_one();
    }

    std::atomic_bool m_stopped = false;
    std::vector<std::thread> m_pool;
    std::queue<std::shared_ptr<pool_task>> m_pool_tasks;
    std::mutex m_tasks_lock;
    std::condition_variable m_tasks_cv;
}; 

Code works and tested, although not sure I've utilized C++17's full power.

\$\endgroup\$

1 Answer 1

1
\$\begingroup\$

Incorrect #includes

You have lots of unnecessary #include statements, for example <atomic>, <cstdio> and some others, but you are missing the required <condition_variable>.

Make better use of the standard library

Why implement your own class pool_task and class future when the standard library already comes with std::packaged_task and std::future? It could look like:

class thread_pool {
    …
    std::future<int> submit(std::function<int()> callable) {
        std::lock_guard lock(m_tasks_lock);
        m_pool_tasks.emplace(callable);
        return m_pool_tasks.back().get_future();
    }
    …
    void execute_task() {
        std::packaged_task<int()> task;
        {
            …
            task = std::move(m_pool_tasks.front());
            m_pool_tasks.pop();
        }
        task();
    }
    …
    std::queue<std::packaged_task<int()>> m_pool_tasks;
    …
};

Don't mix atomics and mutexes

Make m_stopped a regular bool and only access it while holding m_tasks_lock. Otherwise, the problem is that m_stopped and m_pool_tasks are part of different synchronization scopes, and you will have subtle corner cases where things go wrong.

Separate the queue from the thread pool

Your class thread_pool does several things: it implements a thread-safe queue for tasks, and it manages a pool of threads that take tasks from that queue. Consider splitting this up in two smaller, more maintable classes.

\$\endgroup\$
4
  • \$\begingroup\$ It seems like I did not the since C++11 part correctly. Sorry about that. \$\endgroup\$ Dec 5, 2023 at 12:38
  • \$\begingroup\$ I dont think std::packaged_tasks will work here. When invoked it creates it own new thread. If I understand OP's intentions correctly he wants to reuse threads to remove thread creation overhead. One can argue that std::packaged_tasks like std::async might use threadpools under the hood but the standard does not guarantee it. \$\endgroup\$ Dec 7, 2023 at 19:58
  • \$\begingroup\$ @AjinkyaKamat std::packaged_task does not create a thread. It just wraps a callable so that its return value can be waited for and accessed in a thread-safe way. \$\endgroup\$
    – G. Sliepen
    Dec 7, 2023 at 23:11
  • \$\begingroup\$ @G.Sliepen Oh, so its like std::function but with std::future? That explains why my code is slow. Doh. \$\endgroup\$ Dec 7, 2023 at 23:53

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

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