10
\$\begingroup\$

I have written a small and simple ThreadPool class for rather simple multithreading applications.

  • The Threadpool class manages a vector with the actual threads.
  • The Thread struct is used to store the thread's function.
  • ThreadPoolElement keeps track of the thread and its state.
#pragma once

#include <vector>
#include <thread>
#include <mutex>

namespace frm { namespace util {

template <typename F, typename ... Args>
struct Thread {
    Thread(F&& f) 
    : func(f) {
    }
    std::function<void(Args...)> func;
};

class Threadpool {

public:

    Threadpool(const size_t& capasity = std::thread::hardware_concurrency())
        : m_capacity(capasity) {
        m_threads.reserve(m_capacity);
    }

    virtual ~Threadpool() {
        for (auto& t : m_threads) {
            t.thread.join();
        }
    }

    template <typename Thread, typename ... Args>
    void getThread(Thread& t, Args... args) {

        while (m_threads.size() >= m_capacity) {
            cleanup();
        }

        std::lock_guard<std::mutex> lock(m_mutex);
        m_threads.emplace_back(ThreadPoolElement());
        ThreadPoolElement& elem = m_threads[m_threads.size() - 1];
        elem.thread = std::thread([&] (ThreadPoolElement* e, Thread& t, Args... args) {
            t.func(args...);
            e->is_done = true;
        }, &elem, t, args...);
    }

    size_t activeThreads() const {
        return m_threads.size();
    }

    size_t reservedThreads() const {
        return m_capacity;
    }

    void cleanup() {
        for (int i = 0; i < m_threads.size(); i++) {
            if (m_threads[i].is_done) {
                std::lock_guard<std::mutex> lock(m_mutex);
                m_threads[i].thread.join();
                m_threads.erase(m_threads.begin() + i--);
            }
        }
    }

private:

    struct ThreadPoolElement {
        std::thread thread;
        volatile bool is_done = false;
    };

    size_t m_capacity;
    std::mutex m_mutex;
    std::vector<ThreadPoolElement> m_threads;

};

} }

My questions are pretty simple:

  • Has this implementation any significant downsides that would make it a 'Please do not use this' implementation?

  • What are the most important parts I'm missing?

\$\endgroup\$
  • 2
    \$\begingroup\$ Welcome to Code Review! \$\endgroup\$ – L. F. Sep 25 at 10:17
12
\$\begingroup\$

I found a few issues, here they are (in order in which they were found). The list may seem long, but this does not mean that you did a bad job. Instead, it means that the feedback is detailed ;)

If some issue starts with [major], it means that it would stop me from using your class.

  • [major] There is zero documentation.
  • The Thread struct is not really a thread. In fact, it just stores a single std::function member. Why did you name it like that?
  • The Threadpool has a virtual destructor. This is not necessary, because it does not make sense to inherit from it. I suggest you make it non-virtual.
  • The prefix get usually indicates that you return something from it. However, getThread does not return anything. Even though t is passed by reference, you do not change it. Maybe addTask is a better name.
  • [major] If the thread pool is full, getThread blocks until one of the threads is free. This is bad. In addition, you use an active loop to block, which means that waiting takes 100% of one CPU core. This is really bad. The whole point of threads and thread pools is that the user does not need to wait. I suggest that you copy or move the task (your std::function) into the pool and execute it as soon as some thread is free. This way, getThread can return immediately.
  • [major] The thread pool currently creates one std::thread for each task that is added. However, creating threads has some overhead and this will make a big difference if you have many small tasks. I suggest that you create m_capacity worker threads that grab the tasks from some container. This transforms your problem into a typical producer-consumer problem (adding a task is producing work, which is then consumed by the threads).
  • [major] You are using std::function<void(Args...)> to store a single task. This means that the thread pool does not support return values. A common pattern is that addTask returns a std::future which will receive the return value after some thread completed the task. Additionally, thread pool users can use the std::future to check whether the task completed and they can use it to wait for completion.
  • [major] The getThread function uses references of t. This means that problems will occur if t is destroyed while the thread is still running. This is bad, because t is passed by the user, so the pool has no control over its lifetime. The non-existing documentation would have been a good place to mention that t must not be destroyed before the task is finished :) Typically, you would std::move the task, which transfers the control to the pool.
  • If I understand m_mutex correctly, it guards m_threads against race conditions. This means that you want to lock the mutex whenever you use m_threads. Since reading from a variable while another thread modifies it is undefined behavior, you need to lock the mutex in some more places. There are some usages of m_thread or of its elements where the mutex is not locked (and I do not mean constructor/destructor).
  • The ThreadPoolElement has a volatile bool. Depending on compiler and platform, volatile may do what you think it does. However, you can not rely on that. You should use std::atomic<bool> instead.
  • Threadpool has a lower case p, but ThreadPoolElement has an upper case P.

I suggest that you refactor the Threadpool such that you do not need to create a new std::thread object for each task. Typically, a thread pool has a queue with pending tasks and a fixed number of worker threads. The worker threads grab tasks from the queue and complete them. This way, it is possible to add many tasks without the overhead of creating new threads.

You may want to look into the std::packaged_task and std::future classes, since they can simplify your implementation. You can probably get rid of the Thread and ThreadPoolElement classes.

\$\endgroup\$
  • 2
    \$\begingroup\$ Welcome to Code Review! \$\endgroup\$ – L. F. Sep 25 at 10:16
  • 2
    \$\begingroup\$ You have done a great job. Thank you! \$\endgroup\$ – Alphastrick Sep 25 at 10:52
  • \$\begingroup\$ I'm unable to figure out, how to queue the task's. I tried to push them to a deque, obviously, this is not possible if the parent class shall not be a template class. Is there any way, to put the queue at some place, there the threads can access it? \$\endgroup\$ – Alphastrick Sep 25 at 12:43
  • 1
    \$\begingroup\$ @Alphastrick That part is tricky if you have never seen it :) Say your task is std::function<R(Args...)> task. Independent of return and argument types, the queue always stores std::function<void()>. You dont push task to the queue, but a lambda [=]() { task(args...); }. You could also use std::packaged_task<R(Args...)> as task, which easily allows you to create a std::future<R> that can be given to the user, but you need C++14 to std::move the packaged_task into the lambda. \$\endgroup\$ – pschill Sep 25 at 13:42
  • 1
    \$\begingroup\$ @Alphastrick Another solution would be the following: Add a base class TaskBase with a pure virtual void execute() method. Add a derived class template <typename R, typename... Args> class Task : public TaskBase. The Task constructor takes a std::function<R(Args...)> and the arguments and stores them. The overridden execute method calls the function with its arguments. Then the addTask method creates a Task and adds it to a TaskBase queue. The workers grab a TaskBase and call execute. \$\endgroup\$ – pschill Sep 25 at 13:48

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.