Mutex-Protected Stack Implementation

The Problem: Race Conditions in a Simple Stack

Let's consider a basic stack. In a single-threaded program, it's a trivial data structure. But what happens when multiple threads try to push and pop from the same stack simultaneously?

The operations are not atomic. A single call to push, for example, involves at least two steps: placing the value in an array and incrementing the stack pointer/count. A thread can be preempted by the OS between these steps, leading to a race condition.

To make this concrete, let's start with a non-thread-safe stack implementation.

The code for Version 1 is:

// A simple fixed-sized stack (version 1)
#define STACK_SIZE 20
int count;
double values[STACK_SIZE];

void push(double v) {
    values[count++] = v;
}

double pop() {
    return values[--count];
}

int is_empty() {
    return count == 0;
}

Consider two threads, T1 and T2, calling push() concurrently when count is 5.

1. T1 executes values[5] = value1;
2. The OS preempts T1 before it can execute count++. count is still 5.
3. T2 executes values[5] = value2;. This overwrites the value from T1.
4. T2 executes count++. count is now 6.
5. T1 resumes and executes count++. count is now 7.

The result is a disaster: T1's data is lost, and the count is incorrect, leaving a garbage value at index 6. The push, pop, and is_empty functions are all critical sections because they read from or write to the shared values and count variables.

An Attempt to Fix It (With Flaws)

To fix this, we need to ensure mutual exclusion. Only one thread should be allowed inside a critical section at any given time. A mutex is the perfect tool for this.

Let's look at a first attempt to add mutexes.

A Correct, Encapsulated Implementation

The solution is to use a single mutex to protect all operations on a given stack instance. Furthermore, good software design dictates that the data and the synchronization primitive that protects it should be bundled together. Instead of using global variables, we'll create a stack_t struct that encapsulates the data, the count, and the mutex.

This final, correct version is also detailed in the Synchronization resource.

Let's break down the key aspects of this superior design:

• Encapsulation: The stack_t struct holds the count, the values pointer, and the pthread_mutex_t m. Any instance of the stack carries its own data and its own lock. This allows you to have multiple independent thread-safe stacks.
• Lifecycle Management:
  • stack_create: Allocates memory for the struct and its internal values array. Crucially, it calls pthread_mutex_init() to initialize the mutex for that specific stack instance.
  • stack_destroy: Frees the values array and the stack struct itself. It properly cleans up by calling pthread_mutex_destroy().
• Mutual Exclusion:
  • Every function (push, pop, is_empty) that accesses shared data first calls pthread_mutex_lock() on the stack's mutex.
  • Just before returning, it calls pthread_mutex_unlock(). This ensures that all operations on the stack's internal state are atomic from the perspective of other threads.

Here is a complete, runnable example based on that design. Note that this implementation does not yet handle pushing to a full stack or popping from an empty one. We'll address that limitation in the next lesson.

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

#define STACK_CAPACITY 10
#define NUM_PRODUCERS 2
#define NUM_CONSUMERS 2
#define ITEMS_PER_PRODUCER 5

// The encapsulated, thread-safe stack
typedef struct {
    int count;
    int capacity;
    pthread_mutex_t mutex;
    int* values;
} stack_t;

// Function prototypes for our stack
stack_t* stack_create(int capacity);
void stack_destroy(stack_t* s);
void stack_push(stack_t* s, int value);
int stack_pop(stack_t* s);

// Thread functions
void* producer(void* arg);
void* consumer(void* arg);

int main() {
    stack_t* s = stack_create(STACK_CAPACITY);
    pthread_t producers[NUM_PRODUCERS];
    pthread_t consumers[NUM_CONSUMERS];

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

    // Join all threads
    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("All threads finished. Final stack count: %d\n", s->count);

    stack_destroy(s);
    return 0;
}

// --- Stack Implementation ---

stack_t* stack_create(int capacity) {
    stack_t* s = malloc(sizeof(stack_t));
    if (!s) return NULL;
    
    s->capacity = capacity;
    s->count = 0;
    s->values = malloc(sizeof(int) * capacity);
    if (!s->values) {
        free(s);
        return NULL;
    }
    pthread_mutex_init(&s->mutex, NULL);
    return s;
}

void stack_destroy(stack_t* s) {
    pthread_mutex_destroy(&s->mutex);
    free(s->values);
    free(s);
}

void stack_push(stack_t* s, int value) {
    pthread_mutex_lock(&s->mutex);
    
    // For this example, we assume the stack won't overflow
    if (s->count < s->capacity) {
        s->values[s->count++] = value;
    }
    
    pthread_mutex_unlock(&s->mutex);
}

int stack_pop(stack_t* s) {
    pthread_mutex_lock(&s->mutex);
    
    int value = -1; // Return -1 if stack is empty
    // For this example, we assume the stack won't underflow in the test
    if (s->count > 0) {
        value = s->values[--s->count];
    }
    
    pthread_mutex_unlock(&s->mutex);
    return value;
}

// --- Thread Logic ---

void* producer(void* arg) {
    stack_t* s = (stack_t*)arg;
    for (int i = 0; i < ITEMS_PER_PRODUCER; i++) {
        int item = (int)(pthread_self() % 100) * 100 + i;
        stack_push(s, item);
        printf("Producer %ld pushed %d\n", pthread_self() % 100, item);
        usleep((rand() % 100) * 1000); // sleep for 0-99 ms
    }
    return NULL;
}

void* consumer(void* arg) {
    stack_t* s = (stack_t*)arg;
    // Each consumer will try to pop the same number of items a producer makes
    for (int i = 0; i < ITEMS_PER_PRODUCER; i++) {
        // Simple busy-wait loop if stack is empty. We'll improve this in the next lesson.
        int item = -1;
        while(item == -1) {
             // Lock, check, pop, unlock all happens in stack_pop
             item = stack_pop(s);
             if (item != -1) {
                 printf("Consumer %ld popped %d\n", pthread_self() % 100, item);
             } else {
                 usleep(10 * 1000); // wait 10ms if stack was empty
             }
        }
    }
    return NULL;
}

int stack_peek(stack_t* s) {
    pthread_mutex_lock(&s->mutex);

    int value = -1; // Value to return if stack is empty
    if (s->count > 0) {
        value = s->values[s->count - 1]; // Get value at the top index
    }

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

Conclusion

In this lesson, we successfully designed and implemented a thread-safe stack. We saw how seemingly simple data structure operations can fail catastrophically under concurrent access and how a mutex can be used to enforce atomicity.

Key Takeaways:

• Any data structure that will be shared between threads must be explicitly designed for thread safety.
• The first step is to identify the critical sections—code that accesses shared, mutable state.
• A single mutex should be used to protect all critical sections associated with a single instance of a data structure.
• The best practice is to encapsulate the data and its corresponding mutex within a struct, managing their lifecycle together with create and destroy functions.

Preview of the Next Lesson:
Our current stack implementation has a significant inefficiency. In the consumer code, if the stack is empty, the thread spins in a loop, repeatedly calling stack_pop, burning CPU cycles just to check for work. This is called "busy-waiting."

In our next lesson, we will design and implement a thread-safe queue. As part of this, we will introduce a new, more powerful synchronization primitive: the condition variable. Condition variables allow threads to wait efficiently for a specific condition (like "the queue is no longer empty") without burning CPU, solving the busy-waiting problem.