C++ / CAPSTONE PROJECTS
Project: a thread-safe task queue with futures
Build a mutex- and condition-variable-guarded task pool whose submit() returns a std::future, and shut it down without dropping queued work.
What you will learn
- Wrap a callable in std::packaged_task so submit() can hand the caller a std::future
- Wait with cv.wait(lock, predicate) so spurious wakeups and early notifies stay harmless
- Drain the queue before a worker returns, so no future is left with a broken promise
- Move the job out of the queue and unlock before running it, or workers serialize
Understanding Project: a thread-safe task queue with futures
A task queue and a set of futures are the two halves of one handoff. The queue moves ownership of work forward: the caller builds a callable, the mutex-protected queue holds it until some worker pops it, and the caller stops touching it. The future moves the answer back, because std::packaged_task binds a callable to a shared state that can hold either a return value or a caught exception, and std::future is the read end of that state. Once submit() reads as "give away the work, keep the receipt", the rest follows: no shared result vector, no result mutex, no polling flag.
The condition variable is what stops idle workers from burning a core on queue_.empty(). Its predicate form, ready_.wait(lock, pred), rechecks the predicate with the mutex held, which is why a spurious wakeup costs nothing and why a notify that arrives before the wait cannot be lost, provided the state it announces was changed under the same mutex. The predicate is closed_ || !queue_.empty() rather than just !queue_.empty(), because a worker needs two reasons to wake up: there is work, or there will never be work again. The job is then moved out and the lock released before it runs; hold the mutex across job() and three workers execute strictly one at a time.
Shutdown is where task queues usually leak bugs, and futures make the requirement exact: every submitted task owns a promise, so a task discarded at shutdown leaves its future to throw std::future_error with broken_promise. Returning only when the queue is empty and closed_ is set gives drain-then-exit, and joining inside the destructor guarantees no thread is still reading the queue while the members are destroyed. The shared state is reference counted, which is why the vector of futures in main outlives the pool and its get() calls return immediately; get() also moves the value out and invalidates the future, so a value needed twice belongs in a std::shared_future.
<condition_variable>
<cstddef>
<functional>
<future>
<iostream>
<memory>
<mutex>
<queue>
<stdexcept>
<thread>
<utility>
<vector>
class TaskPool {
public:
explicit TaskPool(unsigned workers) {
for (unsigned i = 0; i < workers; ++i)
threads_.emplace_back([this] { loop(); });
}
~TaskPool() {
{
std::lock_guard<std::mutex> lock(mutex_);
closed_ = true;
}
ready_.notify_all();
for (std::thread& t : threads_)
t.join();
}
template <class F>
auto submit(F f) -> std::future<decltype(f())> {
using R = decltype(f());
auto task = std::make_shared<std::packaged_task<R()>>(std::move(f));
std::future<R> result = task->get_future();
{
std::lock_guard<std::mutex> lock(mutex_);
if (closed_)
throw std::runtime_error("pool is closed");
queue_.push([task] { (*task)(); });
}
ready_.notify_one();
return result;
}
private:
void loop() {
for (;;) {
std::function<void()> job;
{
std::unique_lock<std::mutex> lock(mutex_);
ready_.wait(lock, [this] { return closed_ || !queue_.empty(); });
if (queue_.empty())
return; // closed and drained
job = std::move(queue_.front());
queue_.pop();
}
job(); // run with the mutex released
}
}
std::mutex mutex_;
std::condition_variable ready_;
std::queue<std::function<void()>> queue_;
std::vector<std::thread> threads_;
bool closed_ = false;
};
long long triangular(int n) {
long long sum = 0;
for (int k = 1; k <= n; ++k)
sum += k;
return sum;
}
int main() {
std::vector<std::future<long long>> results;
{
TaskPool pool(3);
for (int i = 1; i <= 6; ++i)
results.push_back(pool.submit([i] { return triangular(i * 1000); }));
} // destructor closes the queue, lets the workers drain it, then joins
long long total = 0;
for (std::size_t i = 0; i < results.size(); ++i) {
long long value = results[i].get();
std::cout << "task " << (i + 1) << " -> " << value << '\n';
total += value;
}
std::cout << "total = " << total << '\n';
}
A future is the return channel for work you no longer own: ownership travels forward through the mutex-guarded queue, and the result or exception travels back through the task's shared state.
Worked examples
An exception rides the future home
Shows that a throwing task does not kill the worker thread; the exception is stored and rethrown at get().
<future>
<iostream>
<stdexcept>
<thread>
<utility>
int main() {
std::packaged_task<int(int)> task([](int n) {
if (n < 0)
throw std::domain_error("negative input");
return n * 2;
});
std::future<int> result = task.get_future();
std::thread worker(std::move(task), -5);
worker.join();
try {
std::cout << "value " << result.get() << '\n';
} catch (const std::domain_error& e) {
std::cout << "caught: " << e.what() << '\n';
}
std::cout << std::boolalpha << "still usable: " << result.valid() << '\n';
}
Example explained
Line 1task.get_future() must run before the task is moved into the thread, since the moved-from packaged_task is left empty.
Line 2std::thread worker(std::move(task), -5) invokes the task with -5 on the new thread; the move is required because packaged_task is not copyable.
Line 3packaged_task::operator() catches the domain_error and stores it in the shared state, so it never escapes the thread function and std::terminate is not called.
Line 4result.get() rethrows that stored exception in main, and because get() releases the shared state, result.valid() is false afterwards.
A queue of move-only tasks
Stores std::packaged_task in the queue directly, avoiding the shared_ptr that std::function forces on you.
<condition_variable>
<cstddef>
<future>
<iostream>
<mutex>
<queue>
<thread>
<utility>
<vector>
std::mutex m;
std::condition_variable ready;
std::queue<std::packaged_task<void()>> jobs;
bool closed = false;
template <class F>
auto submit(F f) -> std::future<decltype(f())> {
using R = decltype(f());
std::packaged_task<R()> inner(std::move(f));
std::future<R> result = inner.get_future();
std::packaged_task<void()> outer([t = std::move(inner)]() mutable { t(); });
{
std::lock_guard<std::mutex> lock(m);
jobs.push(std::move(outer));
}
ready.notify_one();
return result;
}
void worker() {
for (;;) {
std::packaged_task<void()> job;
{
std::unique_lock<std::mutex> lock(m);
ready.wait(lock, [] { return closed || !jobs.empty(); });
if (jobs.empty())
return;
job = std::move(jobs.front());
jobs.pop();
}
job();
}
}
int main() {
std::thread w(worker);
std::vector<std::future<int>> fs;
for (int i = 1; i <= 3; ++i)
fs.push_back(submit([i] { return i * i; }));
{
std::lock_guard<std::mutex> lock(m);
closed = true;
}
ready.notify_all();
for (std::size_t k = 0; k < fs.size(); ++k)
std::cout << "job " << (k + 1) << " result " << fs[k].get() << '\n';
w.join();
}
Example explained
Line 1std::queue only ever moves its elements, so it can hold move-only std::packaged_task<void()> objects that std::function<void()> would reject.
Line 2The wrapper captures inner by move, and packaged_task accepts that lambda because its constructor only requires a move-constructible callable.
Line 3The predicate closed || !jobs.empty(), followed by the empty check, makes the worker run all three jobs before returning, so no future is abandoned.
Line 4main sets closed before calling get(), so it blocks on the futures rather than on join(), and results print in submission order.
Important notes
On GNU/Linux compile with -pthread; without it the program links but std::thread construction throws std::system_error at runtime.
future::get() is valid exactly once and moves the result out, so if two places need the same value call fut.share() and hold a std::shared_future.
Common mistakes
Pushing a std::packaged_task straight into queue<std::function<void()>>: std::function needs a copyable target, so the build dies in overload resolution; wrap the task in a shared_ptr or store std::packaged_task<void()> instead.
Letting a worker return as soon as closed_ is true without checking the queue: still-queued tasks are silently dropped and each abandoned future throws std::future_error broken_promise at get().
Calling job() while still holding the mutex: results are correct but the workers run one task at a time, and a task that submits more work deadlocks on the lock its own worker holds.
Try it yourself
Change, predict, then run
Change the pool above so the task for i == 4 throws std::runtime_error("bad input"), then wrap each get() in try/catch and print either the value or the message. Confirm the other five results still arrive and the destructor still joins cleanly.
Open the C++ workspaceCheck your understanding
Six tasks go to a pool of three workers, their futures are stored in a vector, and the caller loops over that vector calling get() and printing. Why does the program print the same six lines in the same order on every run even though the tasks finish in an unpredictable order?
- The condition variable hands queued tasks to workers in FIFO order, so the tasks also finish in FIFO order.
- The mutex around the queue makes the tasks run one at a time, so submission order is execution order.
- Each future refers to one task's own shared state, so get() waits for that task's result no matter which task finished first.
- get() returns whichever result became ready first, and the vector reorders the values to match submission order.
Show answer
Every submitted task has its own shared state and its own future, so blocking on the third future says nothing about tasks four to six; the print order is simply the order in which you visit the futures. Option 0 is tempting because dequeuing really is FIFO, but three workers start tasks concurrently and task durations differ, so finishing order is not FIFO. Option 1 confuses the scope of the mutex: it guards the queue only, and the task bodies run with the lock released.