Building a Thread-Safe Blocking Queue

The Journey to a Thread-Safe Queue

Let's start by watching a video that perfectly demonstrates the step-by-step process of building a thread-safe queue. It begins with a naive, unsafe implementation and progressively adds synchronization layers, fixing problems at each stage. This will be a practical recap of the concepts we discussed theoretically in the last lesson.

As the video demonstrates, simply wrapping our queue operations with a std::mutex solves the data corruption problem but introduces a significant performance issue. The consumer thread spins in a tight loop, constantly locking the mutex just to check if the queue is empty. This "busy-waiting" wastes CPU cycles, generates heat, and is the hallmark of an amateurish concurrency design.

The Efficient Solution: Signaling with std::condition_variable

To eliminate busy-waiting, we need to move from a polling model ("Is there work yet?") to a signaling or event-driven model ("Wake me up when there is work."). This is precisely the job of a std::condition_variable.

The core logic, which we'll soon implement, follows this pattern:

Let's continue with the video to see how std::condition_variable and its partner, std::unique_lock, solve the busy-waiting problem.

The wait Operation and its Internals

The cv.wait() call is the heart of this mechanism. As the video explained, its behavior is critical to understand:

1. It checks the predicate. If the predicate is true, wait() returns immediately, and the thread continues, still holding the lock.
2. If the predicate is false, wait() atomically releases the lock and puts the thread into a sleeping/blocked state.
3. When the thread is woken up (by a notify call or spuriously), it atomically re-acquires the lock and goes back to step 1 to re-check the predicate.

This atomic release-and-sleep is fundamental. If the thread went to sleep while holding the lock, the producer thread could never acquire the lock to add an item, leading to a deadlock.

Since you're interested in OS internals, it's worth knowing that on Linux, this sleep/wakeup mechanism is typically implemented using futexes (Fast Userspace Mutexes). The cv.wait() call, when it needs to sleep, makes a sys_futex(FUTEX_WAIT, ...) system call. The cv.notify_one() call results in a sys_futex(FUTEX_WAKE, ...) call. This is how the C++ standard library interfaces with the kernel's scheduler to efficiently block and unblock threads.

The resource "Mutexes and Condition Variables using Futexes" shows a low-level C implementation. Notice in the cond_wait function how it calls mutex_unlock before making the sys_futex call to wait.

// From the locklessinc.com article
int cond_wait(cv *c, mutex *m) {
    int seq = c->seq;
    // ... setup code ...
    mutex_unlock(m); // The lock is released!
    sys_futex(&c->seq, FUTEX_WAIT_PRIVATE, seq, NULL, NULL, 0); // Go to sleep.
    while (xchg_32(m, 2)) { // Re-acquire the lock upon waking.
        sys_futex(m, FUTEX_WAIT_PRIVATE, 2, NULL, NULL, 0);
    }
    return 0;
}

This C implementation mirrors the logic inside std::condition_variable::wait and connects the high-level C++ feature directly to the underlying system call.

Encapsulating into a Robust Class

Now, let's follow best practices and encapsulate this logic into a clean, reusable, templated class. We will also address a subtle but critical bug related to exception safety that distinguishes production-quality code from a simple demo.

The article highlights two major issues, but the second one is particularly insightful:

• Problem: Imagine a pop() function that returns the value: T pop(). Inside, it might look like this:
  1. Lock mutex.
  2. Wait for data.
  3. T value = myQueue.front();
  4. myQueue.pop();
  5. Unlock mutex.
  6. return value;
• The Danger: What if the return value operation (which involves a copy or move) throws an exception (e.g., out of memory)? The element has already been removed from myQueue (step 4), but it never reached the caller. The data is lost forever.

The solution is to change the function signature. Instead of returning the value, we pass an output parameter by reference.

Let's watch the final part of the DeepDiveDev video, which implements exactly this robust, exception-safe design.

The Final Implementation

By combining all these concepts, we arrive at our final ThreadSafeQueue implementation. This class is the core component we'll use in our thread pool.

#include <queue>
#include <mutex>
#include <condition_variable>

template<typename T>
class ThreadSafeQueue {
public:
    ThreadSafeQueue() = default;
    ~ThreadSafeQueue() = default;

    // We disable copy and move semantics for simplicity.
    // A queue managing threads' work shouldn't be copied.
    ThreadSafeQueue(const ThreadSafeQueue&) = delete;
    ThreadSafeQueue& operator=(const ThreadSafeQueue&) = delete;

    /**
     * @brief Pushes a new value onto the queue.
     * @param value The value to be moved into the queue.
     */
    void push(T value) {
        // Use a simple lock_guard, as we just need to lock, push, and unlock.
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(std::move(value));

        // Notify one waiting thread that a new item is available.
        cond_var_.notify_one();
    }

    /**
     * @brief Waits until an element is available and then pops it into the provided reference.
     * This is an exception-safe design.
     * @param value A reference to store the popped value.
     */
    void wait_and_pop(T& value) {
        // unique_lock is required for condition_variable::wait
        std::unique_lock<std::mutex> lock(mutex_);

        // The wait call will atomically unlock 'lock' and wait.
        // Upon wakeup, it re-acquires the lock. The predicate protects
        // against spurious wakeups.
        cond_var_.wait(lock, [this] { return !queue_.empty(); });

        // By now, we hold the lock and the queue is not empty.
        // Move the value to the output reference *before* popping from the queue.
        // If this move throws, the queue remains unchanged and the data is not lost.
        value = std::move(queue_.front());
        queue_.pop();
    }

private:
    std::mutex mutex_;
    std::queue<T> queue_;
    std::condition_variable cond_var_;
};

This implementation is:

• Thread-Safe: The mutex_ protects all access to the internal queue_.
• Blocking: The wait_and_pop method will block efficiently (without busy-waiting) until an element is available.
• Generic: It's a class template that can hold any type T.
• Exception-Safe: It correctly handles exceptions during the copy/move of an element, preventing data loss.
• Encapsulated: The synchronization details are hidden from the user, providing a clean API.

void push(T value) {
    {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(std::move(value));
    } // lock is released here
    cond_var_.notify_one();
}

Conclusion

In this lesson, we have successfully built a robust, thread-safe, blocking queue from first principles. We went on a journey from a naive, incorrect implementation to a professional-grade component, understanding not just the "how" but also the "why" at each step.

Key Takeaways:

• A thread-safe queue requires a combination of a container (std::queue), a mutex (std::mutex), and a condition variable (std::condition_variable).
• std::mutex alone leads to inefficient busy-waiting. std::condition_variable provides an efficient, event-driven mechanism for threads to wait for work.
• The cv.wait(lock, predicate) function is the core of the blocking mechanism, and the predicate is mandatory to handle spurious wakeups.
• For a robust implementation, exception safety is paramount. Designing methods like wait_and_pop to use output reference parameters (void pop(T& value)) prevents data loss if a copy/move operation throws.
• Encapsulating this logic into a class provides a clean, reusable, and maintainable component for concurrent systems.

Next Lesson Preview:

We now have the backbone of our thread pool. In the next lesson, we will integrate this ThreadSafeQueue into our ThreadPool class. We will implement the worker thread's main loop to consume tasks from our new queue and explore how to handle tasks that have return values by using std::packaged_task and std::future.