C++ Thread Pool with suspend functionality,

Question

This implementation of thread pool that includes support for suspend functionality, which is necessary for refreshing reference data periodically.

I would appreciate your review to ensure that there are no bugs and that the implementation is well-structured. Although everything seems to be working fine, I am still a novice in this area and might have overlooked some details that could be improved.

class ThreadPool
{
public:
    ThreadPool(size_t size = std::thread::hardware_concurrency()) 
    {
        threads_.reserve(size);
        for (size_t i = 0; i < size; ++i)
        {
            threads_.emplace_back([this, tid = i]()
                                  { thread_func(tid); });
        }
    }
    template <typename F, typename... Args>
    void push(F &&f, Args &&...args)
    {
        {
            std::unique_lock<std::mutex> lock(mutex_);
            tasks_.emplace([=]() { f(args...); });
        }
        condition_.notify_one();
    }

    ~ThreadPool()
    {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            stop_ = true;
        }
        condition_.notify_all();
        for (auto &thread : threads_)
        {
            thread.join();
        }
    }
    void suspend()
    {
        suspend_mutex_.lock();
        {
            std::lock_guard<std::mutex> lock(mutex_);
            suspender_id_ = std::this_thread::get_id();
            suspend_ = true;
        }
    }

    void resume()
    {
        std::lock_guard<std::mutex> lock(mutex_);
        if (suspender_id_ != std::this_thread::get_id()) { return; } 
        suspend_ = false;
        suspend_mutex_.unlock();
    }
    bool isSuspended()
    {
        std::unique_lock<std::mutex> lock(mutex_);
        return active_threads_ == 0;
    }

private:
    void thread_func(size_t thread_id)
    {
        {
            std::unique_lock<std::mutex> lock(mutex_);
            std::cout << "starting: " << thread_id << '\n';
        }
        while (true)
        {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lock(mutex_);

                condition_.wait(lock, [this]
                                { return !tasks_.empty() || stop_; });
                if (stop_ && tasks_.empty())
                {
                    --active_threads_;
                    return;
                }
                if (suspend_) 
                {
                    continue;
                }

                task = std::move(tasks_.front());
                tasks_.pop();
               ++active_threads_;
            }
            if (task) task();
            std::unique_lock<std::mutex> lock(mutex_);
            --active_threads_;
        }
    }
    std::vector<std::thread> threads_;
    std::queue<std::function<void()>> tasks_;
    std::mutex mutex_;
    std::mutex suspend_mutex_;
    std::condition_variable condition_;
    std::thread::id suspender_id_;
    bool stop_ {false};
    bool suspend_ {false};
    std::size_t active_threads_ {0};
};

Usage example

void thread_func(ThreadPool& pool)
{
     int count = 0;
    // Push some tasks to the thread pool
    pool.push([&count]() { count += 1; });
    pool.push([&count]() { count += 2; });
    pool.push([&count]() { count += 3; });

    // Suspend the thread pool and wait for a short period of time
    pool.suspend();
    std::cout << "Thread pool suspended\n";
    std::this_thread::sleep_for(std::chrono::seconds(1));

    // Push some more tasks and verify that they're not executed
    pool.push([&count]() { count += 4; });
    pool.push([&count]() { count += 5; });
    pool.push([&count]() { count += 6; });
    std::cout << "More tasks pushed, count = " << count << "\n";

    // Resume the thread pool and wait for another short period of time
    pool.resume();
    std::cout << "Thread pool resumed\n";
    std::this_thread::sleep_for(std::chrono::seconds(1));

    // Verify that all tasks have executed
    std::cout << "Final count = " << count << "\n";
    if (count == 21) {
        std::cout << "Test passed\n";
    } else {
        std::cout << "Test failed\n";
    }

}
int main()
{
    ThreadPool pool(8);
    while(1)
    for (int i = 0 ; i < 2 ; i++)
    {
        std::jthread t(thread_func, std::ref(pool));  // Create a new thread and pass in the thread pool by reference
    }
}

Answer 1

Remove thread_id

If there is no good reason why the threads in the pool should know their ID, I would omit this. This simplifies construction.

Different threads calling suspend() and resume()

If a thread calls resume() but it hasn't called suspend(), that is most definitely a programming error. Instead of just ignoring this, throw or assert(). If you do ignore it anyway, I don't see the point of keeping track of suspender_id_.

Also, if multiple thread call suspend(), then all but one of those threads will be suspended themselves. Instead of doing that, consider allowing multiple calls to suspend() to succeed, but keep track of how many threads called it, and decrement that count in resume().

Remove isSuspended()

Any function which takes a lock, and then only returns some state but does not modify the queue, is most likely wrong. The reason is that the lock is released right before returning, so when the caller looks at the return value, it might no longer reflect the current state of the queue. Any caller relying on the return value will therefore most likely have a race condition.

Even if it is used in a safe way (only after calling suspend() in the same thread), what should the caller do if it return false? Busy-loop until it returns true? I think it would be much better if suspend() waits (using condition variables to avoid busy-looping) until there are no active threads anymore before returning.

Busy loop when suspending threads

If the queue is not empty but suspend_ is true, the worker threads will start busy-looping. This is undesirable. You might want to use condition_ to signal changes in suspension state.

Your main() doesn't run multiple threads

It looks like you wanted to create multiple threads that push jobs to the job queue in main(), but it only creates one thread at a time. You have to write:

while (true) {
    std::jthread t1(thread_func, std::ref(pool));
    std::jthread t2(thread_func, std::ref(pool));
}

Or if you want even more concurrency, create a std::vector<std::jthread>.

Naming things

The name of the function isSuspended() is not correct. It doesn't return whether the thread pool is suspended, it returns whether there are no active threads. The thread pool might not be suspended, but if the queue is empty your function might return true. Conversely, suspend() could have been called, but if one of the threads is still executing a task, it will return false. A better name would be noActiveTasks().

While most functions and variables have good names, thread_func() is very generic, and doesn't describe what that function does at all. Furthermore, your test code also has a thread_func(), making it more confusing. A better name for ThreadPool::thread_func() might be consume_tasks(). Prefer using verbs for function names and nouns for variables.

Related, while threads_ is a reasonable name for the thing keeping track of the threads in a thread pool, it is also a bit generic. You could call it task_consumers, which is more specific.

Answer 2

std::vector<std::thread> threads_;
std::queue<std::function<void()>> tasks_;
std::mutex mutex_;
std::mutex suspend_mutex_;
std::condition_variable condition_;
std::thread::id suspender_id_;
bool stop_ {false};
bool suspend_ {false};
std::size_t active_threads_ {0};

that is way, way, way, way too much state in a single type to be concurrency correct.

So

Split the queue from the pool.

There is a queue of tasks to work on. This can be its own class with its own guarantees.

There is a pool of threads that consume tasks from the queue. This can be a different class.

The next part is working out what properties each needs to have in order to do your suspension operation.

From what I can tell, suspension requires that no threads are currently working on tasks. You need to be able to enter this suspension state, wait for it to occur, then do the stuff you need to do while suspended. Next, you need to restart the processing of threads.

TaskQueue:
  pop task
  push task(task)
  wait until suspended(count)
  resume processing

you may also need some APIs for shutdown.

Now, amusingly, if we add in a push front, we can implement the suspend outside of task queue.

TaskQueue:
  pop task
  push task(task)
  push front task(task)

then create a counting semaphore and a gate. The suspend task counts down the semaphore, then blocks on the gate. We initialize the counting semaphore with the number of threads we want to suspend.

Then wait until suspended (count) becomes:

 push front task (suspend task) x N
 wait on counting semaphore reaching 0

and resume becomes:

 open gate

Writing a gate and a counting semaphore is easier than writing a more complex bit of logic. You should bundle up their use into one package, however.

template <class T>
struct ThreadsafeQueue
{
    std::optional<T> pop()
    {
        auto l = lock();
        cv.wait(l, [&]
                { return !q.empty() || aborted; });
        if (aborted)
        {
            return std::nullopt;
        }
        auto front = std::move(q.front());
        q.pop_front();
        return front;
    }

    void push(T value)
    {
        auto l = lock();
        q.push_back(std::move(value));
        cv.notify_one();
    }

    void push_front(T value)
    {
        auto l = lock();
        q.push_front(std::move(value));
        cv.notify_one();
    }

    void abort()
    {
        auto l = lock();
        aborted = true;
        q.clear();
        cv.notify_all();
    }

private:
    auto lock() const
    {
        return std::unique_lock(m);
    }

    mutable std::mutex m;
    mutable std::condition_variable cv;
    std::deque<T> q;
    bool aborted = false;
};

using TaskQueue = ThreadsafeQueue<std::function<void()>>;

struct ThreadPool
{
    explicit ThreadPool(std::size_t thread_count)
        : threads(thread_count)
    {
        for (auto &thread : threads)
        {
            thread = std::jthread(task_worker, this);
        }
    }

    void abort()
    {
        queue.abort();
    }

    void add_task(std::function<void()> task)
    {
        queue.push(std::move(task));
    }

    TaskQueue &get_queue()
    {
        return queue;
    }

    std::size_t thread_count()
    {
        return threads.size();
    }

private:
    TaskQueue queue;
    std::vector<std::jthread> threads;

    static void task_worker(ThreadPool *self)
    {
        while (auto task = self->queue.pop())
        {
            (*task)();
        }
    }
};

struct Gate
{
    void open_gate()
    {
        auto l = lock();
        is_open = true;
        cv.notify_all();
    }

    void wait() const
    {
        auto l = lock();
        cv.wait(l, [&]
                { return is_open; });
    }

private:
    std::unique_lock<std::mutex> lock() const
    {
        return std::unique_lock<std::mutex>(m);
    }

    mutable std::mutex m;
    mutable std::condition_variable cv;
    bool is_open = false;
};

struct Semaphore
{
    explicit Semaphore(std::size_t count)
        : counter(count)
    {
    }

    void decrement(std::size_t n)
    {
        auto l = lock();
        if (counter < n) counter = 0;
        else counter -= n;
        cv.notify_all();
    }

    void wait_on_zero() const
    {
        auto l = lock();
        cv.wait(l, [&] { return counter == 0; });
    }

private:
    std::unique_lock<std::mutex> lock() const
    {
        return std::unique_lock<std::mutex>(m);
    }

    mutable std::mutex m;
    mutable std::condition_variable cv;
    std::size_t counter;
};


   
struct Suspender
{
    Suspender(ThreadPool& pool)
        : state(std::make_shared<helper>(pool.thread_count()))
    {
        for (std::size_t i = 0; i < pool.thread_count(); ++i)
        {
            pool.get_queue().push_front(
              [state=state]{
                thread_suspender(state);
              }
            );
        }
        state->sem.wait_on_zero();
    }

    ~Suspender()
    {
        state->gate.open_gate();
    }

    Suspender(Suspender const&)=delete;
    Suspender(Suspender&&)=default;
    Suspender& operator=(Suspender const&)=delete;
    Suspender& operator=(Suspender&&)=default;
private:
    struct helper
    {
        Semaphore sem;
        Gate gate;
        explicit helper(std::size_t count):sem(count){}
    };

    std::shared_ptr<helper> state;

    static void thread_suspender(std::shared_ptr<helper> state)
    {
        state->sem.decrement(1);
        state->gate.wait();
    }
};

(Code above is a mixture of my original code, and OP's attempt to implement the missing methods, and my fixing of that attempt to implement it.)

Live example

Now you just create a Suspender s(pool) on the stack and RAII when it exists the thread pool does not process tasks.

The point of this is each piece can be individually tested, instead of a pile of interconnected state. I can write unit tests that Gate acts like it should. I can do the same for each piece.

Then I have some confidence that ThreadPool and Suspender might work. But I can also test those.

You'll note that neither of them have a mutex or the like. They are built on top of tested primitives with clear semantics. Interacting directly with std primitives in complex code is something I seek to avoid; especially more than one.

ThreadPool already has jthread it has to manage. That is hard enough without worrying about mutexes and condition variables!

Also, I might make get_queue a private member, and make Suspender a friend. There are few valid reasons to directly mess with the queue in ThreadPool, and making each a friend is worth the extra care I think.

The point is that multithreaded code is hard. Even if it passes integration tests, you don't know it is correct unless you individually unit test and then produce a proof that the logic is sound. Multithreaded code is well known for having bugs that require strange scheduling luck to occur.

And while you cannot safely compose multithreaded code, building multithreaded code out of smaller pieces with clear semantics in the smaller pieces is easier than building a giant structure all entangled.