A Bounded Blocking Queue in C++: Predicates, Backpressure, and Shutdown

A queue can make a thread pool look simpler while hiding its hardest question: what happens when producers outpace consumers? An unbounded container answers by retaining more memory. A bounded queue needs a deliberate policy.

This example blocks producers when capacity is exhausted and supports graceful close. It is a small teaching implementation for integer values, not the portfolio’s proposed C++ thread-pool project or a production-ready generic container.

Define the contract before the locks

push waits for space and returns false if the queue closes. pop waits for an item, drains existing items after close, then returns false when the closed queue is empty. close is idempotent and wakes both kinds of waiter.

The caller must ensure every thread has stopped using the queue before destroying it. Closing is not joining. A blocked event-loop thread may be the wrong backpressure policy; the same queue can be correct and unsuitable for that caller.

The predicate is the state

Condition variables do not store a count of notifications. A waiter needs a shared-state predicate protected by the mutex. Checking the predicate inside wait handles spurious wakeups and rechecks state after competing threads run.

#include <cassert>
#include <condition_variable>
#include <cstddef>
#include <deque>
#include <exception>
#include <mutex>
#include <stdexcept>
#include <thread>

class IntQueue {
 public:
  explicit IntQueue(std::size_t capacity) : capacity_(capacity) {
    if (capacity == 0) throw std::invalid_argument("capacity must be positive");
  }

  bool push(int value) {
    std::unique_lock<std::mutex> lock(mutex_);
    writable_.wait(lock, [this] { return closed_ || values_.size() < capacity_; });
    if (closed_) return false;
    values_.push_back(value);
    lock.unlock();
    readable_.notify_one();
    return true;
  }

  bool pop(int& value) {
    std::unique_lock<std::mutex> lock(mutex_);
    readable_.wait(lock, [this] { return closed_ || !values_.empty(); });
    if (values_.empty()) return false;
    value = values_.front();
    values_.pop_front();
    lock.unlock();
    writable_.notify_one();
    return true;
  }

  void close() {
    {
      std::lock_guard<std::mutex> lock(mutex_);
      closed_ = true;
    }
    readable_.notify_all();
    writable_.notify_all();
  }

 private:
  const std::size_t capacity_;
  std::deque<int> values_;
  bool closed_ = false;
  std::mutex mutex_;
  std::condition_variable readable_;
  std::condition_variable writable_;
};

int main() {
  IntQueue queue(1);
  std::thread producer([&] {
    if (!queue.push(1) || !queue.push(2)) std::terminate();
    queue.close();
  });
  int value = 0;
  int sum = 0;
  while (queue.pop(value)) sum += value;
  producer.join();
  assert(sum == 3);
  assert(!queue.push(3));
  assert(!queue.pop(value));
  queue.close();
}

Save this complete example as queue.cpp in a disposable local directory. With a thread-capable GCC toolchain:

g++ -std=c++11 -O2 -Wall -Wextra -pthread queue.cpp -o queue-demo
./queue-demo

The executable exits without output when its checks pass. Keep assertions enabled for this smoke test. It is a functional example, not a throughput measurement or an exhaustive concurrency proof.

Why close wakes everyone

Closing changes the condition for both producers and consumers. A producer must stop waiting for capacity that is no longer usable. A consumer must either drain a remaining item or learn that no future item will arrive.

The mutex serializes changes to the queue and closed flag. Notifications occur after unlocking so a woken thread can acquire the lock without immediately competing with the notifier. Correctness still comes from the predicate and lock, not from notification timing alone.

Test boundaries before benchmarking

Exercise capacity one, an empty consumer during close, a full-queue producer during close, repeated close, and multiple producers/consumers with a counted set of values. Use a race detector on a supported platform, but do not confuse a clean run with a proof.

A generic queue adds exception and ownership questions: moving an item can throw, allocation can fail, and cancellation may abandon pending work. This integer example intentionally avoids promising a generic exception guarantee. Deadline-aware waits and nonblocking try_push need separately documented return states.

A queue limit is not a service limit

Queued items can own large buffers, and active workers can hold additional memory. Bound bytes or resource cost where an item count is insufficient. Measure queue wait separately from execution time and report rejection or cancellation outcomes.

Start with clear shutdown semantics and measure contention before choosing a more complex algorithm. The example is a synchronized producer-consumer queue, with its behavior visible in the predicates.

Further reading

Primary references behind this explanation. Consult documentation for the version and environment you actually run.

1. C++ working draft: condition variables
2. C++ working draft: mutex requirements
3. cppreference: condition_variable