7
\$\begingroup\$

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
    }
}
\$\endgroup\$
14
  • \$\begingroup\$ Using previous implementations of thread pools as a reference” Can you add attribution for code you have reused? \$\endgroup\$
    – G. Sliepen
    Commented Apr 2, 2023 at 13:12
  • \$\begingroup\$ I added the refernce, which I modified based on some implemenations of my own, and removed some parts which I thought redundent, \$\endgroup\$
    – G. Nass
    Commented Apr 2, 2023 at 13:28
  • \$\begingroup\$ @G.Sliepen, changed the implemenation as it doesn't need to be complicated, \$\endgroup\$
    – G. Nass
    Commented Apr 2, 2023 at 16:27
  • \$\begingroup\$ To me it looks like there is a race condition in your tests, you are potentially modifying count on several threads at the same time \$\endgroup\$
    – Alex
    Commented Apr 2, 2023 at 16:35
  • 2
    \$\begingroup\$ Please do not edit the question, especially the code, after an answer has been posted. Changing the question may cause answer invalidation. Everyone needs to be able to see what the reviewer was referring to. What to do after the question has been answered. \$\endgroup\$
    – pacmaninbw
    Commented Apr 2, 2023 at 17:48

2 Answers 2

7
\$\begingroup\$

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.

\$\endgroup\$
4
  • \$\begingroup\$ I appreciate your feedback. The thread_id is used as a reference to lookup data. By adding suspender_id_, we are being more cautious without disrupting the execution. If isSuspended is true, the call will wait until all threads are in an idle state. I am planning to use this thread pool with casablanca httprest frame work, should be safe to use? \$\endgroup\$
    – G. Nass
    Commented Apr 2, 2023 at 17:18
  • \$\begingroup\$ regarding the bysy loop, I fixed it like this, << condition_.wait(lock, [this] { return !tasks_.empty() || stop_ || !suspend_; }) >> \$\endgroup\$
    – G. Nass
    Commented Apr 2, 2023 at 17:27
  • 1
    \$\begingroup\$ @G.Adam isSuspended returning true does not mean it is true by the time you make a decision based on it. About the only use for lock-and-return-without-mutate is debugging. \$\endgroup\$
    – Yakk
    Commented Apr 2, 2023 at 23:38
  • \$\begingroup\$ In some cases would be hundreds of core threads \$\endgroup\$
    – G. Nass
    Commented Apr 3, 2023 at 4:08
6
\$\begingroup\$
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.

\$\endgroup\$
9
  • \$\begingroup\$ thanks for your answer, I tried your idea here, but it is not working with my main, godbolt.org/z/d6vsrf7b8, can you pls have a look \$\endgroup\$
    – G. Nass
    Commented Apr 3, 2023 at 11:32
  • \$\begingroup\$ You changed decrement. And you tried to partial suspend. After fixing that and some minor issues: godbolt.org/z/doq36qKab @G.Nass \$\endgroup\$
    – Yakk
    Commented Apr 3, 2023 at 13:23
  • \$\begingroup\$ I'll look into it, FYI, some times it complete and sometimes it hangs., @Yakk \$\endgroup\$
    – G. Nass
    Commented Apr 3, 2023 at 13:33
  • 2
    \$\begingroup\$ @G.Nass Aha! Use-after-free (well, use after destruction, or use not-happens-before destruction). The gate object is freed in ~Suspender after gate.wait(), but the thread_suspender tasks are still possibly gate.wait()ing. Any attempt to synchronize "they all finished waiting" runs into the same problem. Might need a shared pointer/heap allocation to keep the gate around long enough. Move State to a shared_ptr, pass that shared_ptr to the suspension functions, and that should fix that problem. \$\endgroup\$
    – Yakk
    Commented Apr 3, 2023 at 13:41
  • 3
    \$\begingroup\$ @G.Nass In decades of programming, I have never seen any multithreaded code with that many low level variables without any structure keeping it sane without concurrency bugs. That is why I say it has too much state; I am unaware of a software developer or a software developer team capable of writing that code and maintaining it and modifying it exists. Maybe it does, and I just haven't met the people and team, and maybe you are that person! In some cases, I've had to fix concurrency bugs in decade+ complex code like that, only found because we got a stack dump off a user's system. \$\endgroup\$
    – Yakk
    Commented Apr 3, 2023 at 13:50

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.