Challenge: Thread-safe bounded queue

Implement a thread-safe bounded queue in C using mutexes and condition variables — supporting multiple producers and multiple consumers.

A bounded queue (also called a blocking queue or channel) is the fundamental inter-thread communication primitive. It is the foundation of thread pools, work queues, and pipeline architectures. This challenge asks you to implement one that is correct under concurrent access by multiple producers and multiple consumers.

Specification

Implement bqueue.h and bqueue.c with this API:

typedef struct BQueue BQueue;

/* Create a bounded queue with a maximum capacity of `capacity` items. */
BQueue *bqueue_create(int capacity);

/* Push an item. Blocks if the queue is full. */
void bqueue_push(BQueue *q, int item);

/* Pop an item. Blocks if the queue is empty.
   Returns the item. */
int bqueue_pop(BQueue *q);

/* Push an item, but do not block. Returns 1 on success, 0 if full. */
int bqueue_try_push(BQueue *q, int item);

/* Destroy the queue. Undefined behaviour if threads are still using it. */
void bqueue_destroy(BQueue *q);

Requirements

Correct under simultaneous use by N producers and M consumers.
Verified race-free by ThreadSanitizer (-fsanitize=thread).
Blocking bqueue_push must wake when space becomes available.
Blocking bqueue_pop must wake when an item becomes available.
Zero memory leaks (valgrind clean).

Suggested internal structure

struct BQueue {
    int           *data;
    int            capacity;
    int            count;
    int            head;      /* index to read next item from */
    int            tail;      /* index to write next item to */
    pthread_mutex_t lock;
    pthread_cond_t  not_full;
    pthread_cond_t  not_empty;
};

Worked solution

/* bqueue.c */
#include "bqueue.h"
#include <stdlib.h>
#include <stdio.h>

BQueue *bqueue_create(int capacity) {
    BQueue *q = malloc(sizeof(BQueue));
    if (!q) { return NULL; }
    q->data = malloc(capacity * sizeof(int));
    if (!q->data) { free(q); return NULL; }
    q->capacity = capacity;
    q->count = q->head = q->tail = 0;
    pthread_mutex_init(&q->lock, NULL);
    pthread_cond_init(&q->not_full, NULL);
    pthread_cond_init(&q->not_empty, NULL);
    return q;
}

void bqueue_push(BQueue *q, int item) {
    pthread_mutex_lock(&q->lock);
    while (q->count == q->capacity) {
        pthread_cond_wait(&q->not_full, &q->lock);
    }
    q->data[q->tail] = item;
    q->tail = (q->tail + 1) % q->capacity;
    q->count++;
    pthread_cond_signal(&q->not_empty);
    pthread_mutex_unlock(&q->lock);
}

int bqueue_pop(BQueue *q) {
    pthread_mutex_lock(&q->lock);
    while (q->count == 0) {
        pthread_cond_wait(&q->not_empty, &q->lock);
    }
    int item = q->data[q->head];
    q->head = (q->head + 1) % q->capacity;
    q->count--;
    pthread_cond_signal(&q->not_full);
    pthread_mutex_unlock(&q->lock);
    return item;
}

int bqueue_try_push(BQueue *q, int item) {
    pthread_mutex_lock(&q->lock);
    if (q->count == q->capacity) {
        pthread_mutex_unlock(&q->lock);
        return 0;
    }
    q->data[q->tail] = item;
    q->tail = (q->tail + 1) % q->capacity;
    q->count++;
    pthread_cond_signal(&q->not_empty);
    pthread_mutex_unlock(&q->lock);
    return 1;
}

void bqueue_destroy(BQueue *q) {
    pthread_mutex_destroy(&q->lock);
    pthread_cond_destroy(&q->not_full);
    pthread_cond_destroy(&q->not_empty);
    free(q->data);
    free(q);
}

Stress test

/* test_bqueue.c */
#include <stdio.h>
#include <pthread.h>
#include "bqueue.h"

#define NUM_PRODUCERS 3
#define NUM_CONSUMERS 3
#define ITEMS_PER_PRODUCER 10000
#define QUEUE_CAPACITY 20

static BQueue *q;
static int total_consumed = 0;
static pthread_mutex_t total_mtx = PTHREAD_MUTEX_INITIALIZER;

void *producer(void *arg) {
    int id = *(int *)arg;
    for (int i = 0; i < ITEMS_PER_PRODUCER; i++) {
        bqueue_push(q, id * ITEMS_PER_PRODUCER + i);
    }
    return NULL;
}

void *consumer(void *arg) {
    (void)arg;
    int local_count = 0;
    for (int i = 0; i < ITEMS_PER_PRODUCER; i++) {
        bqueue_pop(q);
        local_count++;
    }
    pthread_mutex_lock(&total_mtx);
    total_consumed += local_count;
    pthread_mutex_unlock(&total_mtx);
    return NULL;
}

int main(void) {
    q = bqueue_create(QUEUE_CAPACITY);
    pthread_t producers[NUM_PRODUCERS], consumers[NUM_CONSUMERS];
    int ids[NUM_PRODUCERS];

    for (int i = 0; i < NUM_PRODUCERS; i++) {
        ids[i] = i;
        pthread_create(&producers[i], NULL, producer, &ids[i]);
    }
    for (int i = 0; i < NUM_CONSUMERS; i++) {
        pthread_create(&consumers[i], NULL, consumer, NULL);
    }
    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); }

    printf("Total consumed: %d (expected %d)\n",
           total_consumed, NUM_PRODUCERS * ITEMS_PER_PRODUCER);

    bqueue_destroy(q);
    return 0;
}

Compile and run under TSan:

gcc -Wall -Wextra -pthread -fsanitize=thread -g \
    test_bqueue.c bqueue.c -o test_bqueue
./test_bqueue

Zero TSan errors and the correct total required.

What you practised

Mutex + two condition variables for a blocking circular buffer
The while (!condition) { cond_wait } pattern
Stress testing a concurrent data structure
ThreadSanitizer as a correctness verification tool

The next module is Systems programming patterns — signals, fork/exec, pipes, and writing portable C.

Finished reading? Mark it complete to track your progress.