multithreading · advanced

Concurrency: Thread Pool — Implementation, packaged_task, Shutdown, Pitfalls

Thread pool: create N worker threads once; reuse for many tasks; avoids ~100μs thread-create overhead. Components: shared task queue (queue<function<void()>>), mutex, condition_variable, stop flag. Worker loop: cv.wait(lk, predicate) — predicate handles spurious wakeups; release lock BEFORE task() to avoid serializing all workers. submit(): wrap in packaged_task<R()> via bind; push lambda capturing shared_ptr; cv.notify_one(); return future<R>. Shutdown: set stop=true under mutex; cv.notify_all(); join all workers. Size: N = hardware_concurrency() for CPU-bound; N = cores * (1 + wait/cpu) for I/O-bound. Pitfalls: unbounded queue → OOM; task submits to same pool → deadlock; unwaited future → silent exception loss; dangling pool reference.

🔑 Key line

ThreadPool: N workers + shared queue + mutex + cv; cv.wait(lk, pred) for spurious-wakeup safety; release lock BEFORE task(); submit returns future<R> via packaged_task; ~ThreadPool: lock→stop=true; notify_all; join all.

The code

class ThreadPool {
    vector<thread> workers;
    queue<function<void()>> tasks; // shared work queue
    mutex mtx;
    condition_variable cv;
    bool stop{false};

public:
    explicit ThreadPool(size_t n) {
        for (size_t i = 0; i < n; ++i)
            workers.emplace_back([this] {
                while (true) {
                    function<void()> task;
                    {
                        unique_lock<mutex> lk(mtx);
                        cv.wait(lk, [this] { return stop || !tasks.empty(); });
                        if (stop && tasks.empty())
                            return;
                        task = move(tasks.front());
                        tasks.pop();
                    } // release lock BEFORE executing task
                    task();
                }
            });
    }

    template <typename F, typename... Args>
    auto submit(F&& f, Args&&... args) -> future<invoke_result_t<F, Args...>> {
        using R = invoke_result_t<F, Args...>;
        auto task = make_shared<packaged_task<R()>>(bind(forward<F>(f), forward<Args>(args)...));
        future<R> fut = task->get_future();
        {
            lock_guard<mutex> lk(mtx);
            if (stop)
                throw runtime_error("submit on stopped pool");
            tasks.emplace([task] { (*task)(); });
        }
        cv.notify_one();
        return fut;
    }

    ~ThreadPool() {
        {
            lock_guard<mutex> lk(mtx);
            stop = true;
        }
        cv.notify_all();
        for (auto& w : workers)
            w.join();
    }
};

// Usage:
ThreadPool pool(thread::hardware_concurrency());
auto f = pool.submit([](int x) { return x * x; }, 7);
cout << f.get(); // 49

What this lesson walks through

01The pieces: a shared queue + N workers
02Worker loop — wait, pop, UNLOCK, run
03submit() — enqueue, notify one worker, return a future
04Shutdown — set stop, notify ALL, join
05Sizing the pool
06Pitfalls to call out

A thread pool is a fixed set of worker threads pulling from one shared task queue, guarded by a mutex and a condition variable. Submit puts work in; idle workers wake and run it.

See it animated — step by step, at your own pace

Unlock the full interactive walkthrough of Concurrency: Thread Pool — Implementation, packaged_task, Shutdown, Pitfalls and 100+ animated C++ interview lessons.

Start free →Already a member? Log in →