MiddleCodeCommonNot answered yet
Write a thread-safe thread pool.
Implement a thread pool that pre-spawns N worker threads and runs tasks submitted to a shared queue.
Requirements:
- the task queue is shared state — protect it with a
std::mutexand astd::condition_variable - workers must wait with a predicate (
wait(lock, pred)) so spurious wakeups do not dequeue from an empty queue submit()enqueues astd::function<void()>and wakes exactly one worker (notify_one)- the destructor sets a stop flag, notifies all workers, lets them drain the queue, and
joins every thread (notdetach)
class ThreadPool {
public:
explicit ThreadPool(std::size_t threads);
~ThreadPool();
ThreadPool(const ThreadPool&) = delete;
ThreadPool& operator=(const ThreadPool&) = delete;
void submit(std::function<void()> task);
// your code here
};Write the implementation.
Pre-spawn N workers waiting on a shared task queue protected by a mutex and condition variable. submit() enqueues a std::function<void()> and notifies one worker; the destructor sets a stop flag and joins.
- ✗Not re-checking the predicate after
waitwakes up — spurious wakeups deliver the thread with an empty queue; always usewait(lock, pred)or a loop - ✗Destroying the pool while tasks are still queued — workers should drain the queue before exiting, or the destructor should decide whether to cancel or complete outstanding work
- ✗Blocking
submit()on a full queue without a size limit — unbounded queues can exhaust memory under load
- →How would you add priority support to the task queue?
- →How can you return a
std::futurefromsubmit()so callers can wait for results?
Contents
Thread-Safe Thread Pool
Implementation
#include <condition_variable>
#include <functional>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>
class ThreadPool {
public:
explicit ThreadPool(std::size_t threads) {
for (std::size_t i = 0; i < threads; ++i)
workers_.emplace_back([this] { workerLoop(); });
}
~ThreadPool() {
{
std::lock_guard lock(mu_);
stop_ = true;
}
cv_.notify_all();
for (auto& t : workers_)
t.join();
}
ThreadPool(const ThreadPool&) = delete;
ThreadPool& operator=(const ThreadPool&) = delete;
void submit(std::function<void()> task) {
{
std::lock_guard lock(mu_);
if (stop_)
throw std::runtime_error("submit on stopped ThreadPool");
tasks_.push(std::move(task));
}
cv_.notify_one();
}
private:
void workerLoop() {
while (true) {
std::function<void()> task;
{
std::unique_lock lock(mu_);
cv_.wait(lock, [this] { return stop_ || !tasks_.empty(); });
if (stop_ && tasks_.empty())
return;
task = std::move(tasks_.front());
tasks_.pop();
}
task();
}
}
std::vector<std::thread> workers_;
std::queue<std::function<void()>> tasks_;
std::mutex mu_;
std::condition_variable cv_;
bool stop_ = false;
};Extension: submit returning std::future
#include <future>
#include <type_traits>
template<typename F>
auto ThreadPool::submitFuture(F&& f) -> std::future<std::invoke_result_t<F>> {
using R = std::invoke_result_t<F>;
auto task = std::make_shared<std::packaged_task<R()>>(std::forward<F>(f));
std::future<R> fut = task->get_future();
submit([task] { (*task)(); });
return fut;
}Test
#include <atomic>
#include <cassert>
int main() {
ThreadPool pool(4);
std::atomic<int> counter{0};
const int N = 100;
for (int i = 0; i < N; ++i)
pool.submit([&counter] { ++counter; });
// Pool destructor drains queue and joins all workers
}
// After destructor: counter == N (100)Key Points
| Aspect | Solution |
|---|---|
| Queue synchronization | std::mutex + std::condition_variable |
| Safe shutdown | stop_ flag + notify_all() in destructor |
| Queue drain | worker exits only when stop_ && tasks_.empty() |
| Spurious wakeups | cv_.wait(lock, predicate) |
| Task exceptions | propagated via std::packaged_task with submitFuture |
Contents
