SeniorCodeCommonNot answered yet
Implement producer-consumer using condition variables.
Implement a bounded blocking queue for the producer-consumer pattern, plus a graceful shutdown path.
Requirements:
- one
std::mutexand two condition variables:not_full(producers wait while the queue is full) andnot_empty(consumers wait while it is empty) close()sets adone_flag under the lock and wakes all waiters; mutatedone_only while holding the mutex- consumers must drain remaining items before exiting — the consumer predicate is
done_ && queue_.empty(), not justdone_; signal end-of-stream (e.g.std::nullopt) once drained
template<typename T>
class BoundedQueue {
public:
explicit BoundedQueue(std::size_t capacity);
void push(T item); // blocks while full
std::optional<T> pop(); // blocks while empty; nullopt when drained+closed
void close();
// your code here
};Write the implementation.
A bounded queue under a mutex with two condition variables: not_full (producer waits if full) and not_empty (consumer waits if empty). Graceful shutdown uses a done_ flag — consumers drain on done_ && queue_.empty().
- ✗Using a single condition variable for both 'not full' and 'not empty' — requires
notify_all()and wastes CPU; two CVs withnotify_one()are more efficient - ✗Calling
notify_one()while holding the lock — technically correct but releases the woken thread only to block immediately on the mutex; consider notifying after releasing the lock - ✗Not handling the shutdown race — if
done_is set before consumers check it they may miss the last items; the predicate must bedone_ && queue_.empty()not justdone_
- →How would you implement a lock-free bounded queue for producer-consumer?
- →What is back-pressure and how do you implement it when the consumer is slower than the producer?
Contents
Producer-Consumer with Condition Variables
Implementation
#include <condition_variable>
#include <mutex>
#include <optional>
#include <queue>
#include <stdexcept>
template<typename T>
class BoundedQueue {
public:
explicit BoundedQueue(std::size_t capacity) : capacity_(capacity) {}
// Blocks the producer when the queue is full
void push(T item) {
std::unique_lock lock(mu_);
not_full_.wait(lock, [this] {
return queue_.size() < capacity_ || done_;
});
if (done_)
throw std::runtime_error("push on closed queue");
queue_.push(std::move(item));
not_empty_.notify_one();
}
// Blocks the consumer when the queue is empty; returns nullopt when done
std::optional<T> pop() {
std::unique_lock lock(mu_);
not_empty_.wait(lock, [this] {
return !queue_.empty() || done_;
});
if (queue_.empty())
return std::nullopt; // drained and done_
T item = std::move(queue_.front());
queue_.pop();
not_full_.notify_one();
return item;
}
// Signal completion; consumers finish after draining remaining items
void close() {
{
std::lock_guard lock(mu_);
done_ = true;
}
not_full_.notify_all();
not_empty_.notify_all();
}
private:
std::size_t capacity_;
std::queue<T> queue_;
mutable std::mutex mu_;
std::condition_variable not_full_;
std::condition_variable not_empty_;
bool done_ = false;
};Usage: multiple producers and consumers
#include <atomic>
#include <cassert>
#include <thread>
#include <vector>
int main() {
BoundedQueue<int> q(10);
const int ITEMS = 100;
const int PRODUCERS = 4;
const int CONSUMERS = 3;
std::atomic<int> total_consumed{0};
std::vector<std::thread> consumers;
for (int i = 0; i < CONSUMERS; ++i) {
consumers.emplace_back([&q, &total_consumed] {
while (auto item = q.pop())
++total_consumed;
});
}
std::vector<std::thread> producers;
std::atomic<int> produced{0};
for (int i = 0; i < PRODUCERS; ++i) {
producers.emplace_back([&q, &produced, ITEMS, PRODUCERS] {
int per = ITEMS / PRODUCERS;
for (int j = 0; j < per; ++j)
q.push(produced.fetch_add(1));
});
}
for (auto& t : producers) t.join();
q.close();
for (auto& t : consumers) t.join();
assert(total_consumed.load() == ITEMS);
}Edge Case Analysis
| Scenario | Behaviour |
|---|---|
| Full queue | Producer waits on not_full_ |
| Empty queue | Consumer waits on not_empty_ |
close() with full queue | Producer wakes, throws exception |
close() with empty queue | Consumer wakes, returns nullopt |
close() with items in queue | Consumers drain, then return nullopt |
Contents
