Thread-Safe Queue with Mutex and Condition Variables

The Producer-Consumer Pattern

The producer-consumer pattern involves two types of threads:

• Producers: Threads that create data or tasks and add them to a shared buffer.
• Consumers: Threads that remove data or tasks from the buffer and process them.

A queue is the natural data structure for this, ensuring that items are processed in a First-In, First-Out (FIFO) order.

The Problem with Polling

Just as with the stack, our first step to making a queue thread-safe is to add a mutex to protect its shared state (the data array, head/tail pointers, and item count). However, this leads us back to the busy-waiting problem.

Imagine a dequeue function:

1. Lock the mutex.
2. Check if the queue is empty.
3. If it is, what can the thread do? It cannot hold the lock and wait, as that would block producers from adding items, causing a deadlock.
4. It must unlock the mutex and loop back to try again later.

This polling is inefficient. The CPU is consumed by threads that are doing no useful work.

The Solution: Condition Variables

Condition variables (pthread_cond_t) are synchronization primitives that allow threads to block until a specific condition becomes true. They work in tandem with a mutex to provide a mechanism for efficient waiting.

The core functions are:

• pthread_cond_init(): Initializes a condition variable.
• pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex): This is the key function. It is called while a thread holds a mutex. It atomically:
  1. Unlocks the mutex.
  2. Puts the calling thread to sleep (waiting on cond).
     When the thread is woken up, it will re-acquire the mutex before pthread_cond_wait returns. This atomic operation is crucial for preventing race conditions.
• pthread_cond_signal(pthread_cond_t *cond): Wakes up at least one thread that is currently waiting on the condition variable cond. If no threads are waiting, the signal is lost (it has no effect).
• pthread_cond_destroy(): Destroys a condition variable.

Let's watch a short video that explains the mechanics of these functions with a clear analogy.

Implementing a Thread-Safe Queue

Now that we understand the basics of condition variables, let's apply them to our queue. A naive approach might use a single condition variable. However, a more robust and correct design for a bounded buffer uses two condition variables:

1. not_full: Used to signal that the queue is no longer full. Producers will wait on this condition if they find the queue is full. Consumers will signal this condition after they remove an item.
2. not_empty: Used to signal that the queue is no longer empty. Consumers will wait on this condition if they find the queue is empty. Producers will signal this condition after they add an item.

The following video provides a complete walkthrough of building a task queue for a thread pool, which is precisely the producer-consumer pattern we're implementing.

Why a while Loop? And Why Two Conditions?

The video and our implementation use two critical patterns. It's important to understand the theory behind them.

1. The "Spurious Wakeup" and the while Loop:
   A thread woken by pthread_cond_signal is not guaranteed that the condition it was waiting for is still true. Another thread might have acquired the lock first and changed the state. Also, on some systems, threads can wake up for other reasons (a "spurious wakeup"). Therefore, you must always re-check the condition in a while loop after pthread_cond_wait returns. An if statement is incorrect.

2. The Deadlock Risk with One Condition Variable:
   If you use only one condition variable for both "full" and "empty" states, you can create a deadlock. A consumer might signal, intending to wake up a producer, but instead wake up another consumer. That second consumer would see the queue is still empty and go back to sleep. If all producers are asleep waiting for space, and all consumers are asleep waiting for items, the system is deadlocked. Using two distinct conditions (not_full and not_empty) prevents this ambiguity.

Let's read from the classic OSTEP book to solidify these two points.

The Final Implementation

Here is a complete, runnable C program that implements a thread-safe queue using one mutex and two condition variables. It brings together all the concepts we've discussed.

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

#define QUEUE_CAPACITY 10
#define NUM_PRODUCERS 3
#define NUM_CONSUMERS 3
#define ITEMS_PER_PRODUCER 8

// Encapsulated, thread-safe queue
typedef struct {
    int* values;
    int capacity;
    int count;
    int head;
    int tail;
    
    pthread_mutex_t mutex;
    pthread_cond_t not_full;  // Signaled when an item is removed
    pthread_cond_t not_empty; // Signaled when an item is added
} queue_t;

// Function prototypes
queue_t* queue_create(int capacity);
void queue_destroy(queue_t* q);
void queue_enqueue(queue_t* q, int value);
int queue_dequeue(queue_t* q);

void* producer(void* arg);
void* consumer(void* arg);

int main() {
    queue_t* q = queue_create(QUEUE_CAPACITY);
    pthread_t producers[NUM_PRODUCERS];
    pthread_t consumers[NUM_CONSUMERS];

    for (int i = 0; i < NUM_PRODUCERS; i++) {
        pthread_create(&producers[i], NULL, producer, q);
    }
    for (int i = 0; i < NUM_CONSUMERS; i++) {
        pthread_create(&consumers[i], NULL, consumer, q);
    }

    // Wait for all threads to finish
    for (int i = 0; i < NUM_PRODUCERS; i++) {
        pthread_join(producers[i], NULL);
    }
    for (int i = 0; i < NUM_CONSUMERS; i++) {
        pthread_join(consumers[i], NULL);
    }

    queue_destroy(q);
    return 0;
}

// --- Queue Implementation ---

queue_t* queue_create(int capacity) {
    queue_t* q = malloc(sizeof(queue_t));
    if (!q) return NULL;

    q->capacity = capacity;
    q->values = malloc(sizeof(int) * capacity);
    q->count = 0;
    q->head = 0;
    q->tail = 0;
    
    pthread_mutex_init(&q->mutex, NULL);
    pthread_cond_init(&q->not_full, NULL);
    pthread_cond_init(&q->not_empty, NULL);

    return q;
}

void queue_destroy(queue_t* q) {
    pthread_mutex_destroy(&q->mutex);
    pthread_cond_destroy(&q->not_full);
    pthread_cond_destroy(&q->not_empty);
    free(q->values);
    free(q);
}

void queue_enqueue(queue_t* q, int value) {
    pthread_mutex_lock(&q->mutex);

    // Wait while the queue is full
    while (q->count == q->capacity) {
        pthread_cond_wait(&q->not_full, &q->mutex);
    }

    // Add the item
    q->values[q->tail] = value;
    q->tail = (q->tail + 1) % q->capacity;
    q->count++;

    // Signal that the queue is no longer empty
    pthread_cond_signal(&q->not_empty);

    pthread_mutex_unlock(&q->mutex);
}

int queue_dequeue(queue_t* q) {
    pthread_mutex_lock(&q->mutex);

    // Wait while the queue is empty
    while (q->count == 0) {
        pthread_cond_wait(&q->not_empty, &q->mutex);
    }

    // Remove the item
    int value = q->values[q->head];
    q->head = (q->head + 1) % q->capacity;
    q->count--;

    // Signal that the queue is no longer full
    pthread_cond_signal(&q->not_full);

    pthread_mutex_unlock(&q->mutex);
    return value;
}

// --- Thread Logic ---

void* producer(void* arg) {
    queue_t* q = (queue_t*)arg;
    for (int i = 0; i < ITEMS_PER_PRODUCER; i++) {
        int item = (int)(pthread_self() % 100) * 1000 + i;
        queue_enqueue(q, item);
        printf("Producer %ld enqueued %d\n", pthread_self() % 100, item);
        usleep((rand() % 100) * 100); 
    }
    return NULL;
}

void* consumer(void* arg) {
    queue_t* q = (queue_t*)arg;
    // Each consumer will consume a proportional share of total items
    int items_to_consume = (NUM_PRODUCERS * ITEMS_PER_PRODUCER) / NUM_CONSUMERS;
    for (int i = 0; i < items_to_consume; i++) {
        int item = queue_dequeue(q);
        printf("Consumer %ld dequeued %d\n", pthread_self() % 100, item);
        usleep((rand() % 200) * 100);
    }
    return NULL;
}

Conclusion

In this lesson, we built a robust and efficient thread-safe queue, solving the classic producer-consumer problem. This is a significant step up from our simple mutex-based stack.

Key Takeaways:

• Condition variables solve the busy-waiting problem by allowing threads to sleep until a condition is met.
• pthread_cond_wait is the core function, which atomically unlocks the associated mutex and waits, then re-acquires the lock before returning.
• You must always re-check the condition in a while loop after pthread_cond_wait returns to handle spurious wakeups and race conditions.
• For the bounded buffer problem, using two condition variables (not_full and not_empty) provides a clear and deadlock-free design by directing signals to the correct type of waiting thread.

Preview of the Next Lesson:
Our stack and queue implementations both use a single, coarse-grained lock that protects the entire data structure. While simple and correct, this can become a bottleneck. If one thread is adding an item, no other thread can do anything with the structure, even a non-conflicting operation.

In the next lesson, we will begin exploring more advanced locking strategies by analyzing the performance limitations of using a single lock for a linked list. This will set the stage for implementing fine-grained locking, a technique to improve concurrency.