Lock free single producer - Single consumer ring buffer implementation

Without Atomic operations, the ring buffer implementation

#define MAX_BUFFER_SIZE 10
#define BUFFER_FULL -1
#define BUFFER_EMPTY -2

struct Sensor {
    int id;
    float value;
};

struct ring_buffer {
    struct Sensor *buffer;
    volatile int head;
    volatile int tail; 
    int max;
    // semaphore_t CountingSem;
}

mutex_t lock;

ring_buffer *ring = {
    .buffer = malloc(sizeof(struct Sensor) * MAX_BUFFER_SIZE),
    .head = 0,
    .tail = 0,
    .max = MAX_BUFFER_SIZE
    // .CountingSem = sem_init(&ring->CountingSem, 0, 0)
}

// producer thread
static void enqueue(struct ring_buffer *ring, struct Sensor sensor_data) {
    
    // buffer full check
    next_head = ring->head + 1 % ring->max;
    if (next_head == ring->tail )
        return BUFFER_FULL;
    
    ring->buffer[ring->head] =  sensor_data;

    // move head to next position
    ring->head = (ring->head + 1) % MAX_BUFFER_SIZE;
}

// consumer thread
static void dequeue( struct ring_buffer *ring, struct Sensor *sensor_data ) {
    // buffer empty check
    if (ring->tail == ring->head)
        return BUFFER_EMPTY;

    // copy data from ring buffer to sensor_data
    *sensor_data = ring->buffer[ring->tail];

    // move tail to next position
    ring->tail = (ring->tail + 1) % MAX_BUFFER_SIZE;
}

With Atomic operations

#include <stdatomic.h>
#include <stdlib.h>

#define MAX_BUFFER_SIZE 10
#define BUFFER_FULL -1
#define BUFFER_EMPTY -2
#define SUCCESS 0

struct Sensor {
    int id;
    float value;
};

struct ring_buffer {
    struct Sensor *buffer;
    atomic_int head; // Producer exclusively writes, Consumer reads atomically
    atomic_int tail; // Consumer exclusively writes, Producer reads atomically
    int max;
};

// Helper initialization function (static initialization of pointers/atomics inline has syntax constraints in standard C)
struct ring_buffer* create_ring_buffer(void) {
    struct ring_buffer *ring = malloc(sizeof(struct ring_buffer));
    if (!ring) return NULL;

    ring->buffer = malloc(sizeof(struct Sensor) * MAX_BUFFER_SIZE);
    if (!ring->buffer) {
        free(ring);
        return NULL;
    }

    atomic_init(&ring->head, 0);
    atomic_init(&ring->tail, 0);
    ring->max = MAX_BUFFER_SIZE;

    return ring;
}

// --- Producer Thread Function ---
static int enqueue(struct ring_buffer *ring, struct Sensor sensor_data) {
    // Atomically load current head and tail indices
    int current_head = atomic_load_explicit(&ring->head, memory_order_relaxed);
    int current_tail = atomic_load_explicit(&ring->tail, memory_order_acquire);

    // Calculate next head position
    int next_head = (current_head + 1) % ring->max;

    // Check if buffer is full (next head meets tail)
    if (next_head == current_tail) {
        return BUFFER_FULL;
    }
    
    // Copy data into the buffer slot
    ring->buffer[current_head] = sensor_data;

    // Update head with release memory ordering. 
    // This ensures the data write above is globally visible BEFORE the head pointer update is published to the consumer.
    atomic_store_explicit(&ring->head, next_head, memory_order_release);

    return SUCCESS;
}

// --- Consumer Thread Function ---
static int dequeue(struct ring_buffer *ring, struct Sensor *sensor_data) {
    // Atomically load current tail and head indices
    int current_tail = atomic_load_explicit(&ring->tail, memory_order_relaxed);
    int current_head = atomic_load_explicit(&ring->head, memory_order_acquire);

    // Check if buffer is empty
    if (current_tail == current_head) {
        return BUFFER_EMPTY;
    }

    // Copy data from the ring buffer to destination pointer
    *sensor_data = ring->buffer[current_tail];

    // Calculate next tail position
    int next_tail = (current_tail + 1) % ring->max;

    // Update tail with release memory ordering.
    // This ensures data reading is complete before the tail pointer update is published to the producer.
    atomic_store_explicit(&ring->tail, next_tail, memory_order_release);

    return SUCCESS;
}

Multi Producer- MultiConsumer Ring buffer

#include <stdio.h>
#include <stdlib.h>
#include <stdatomic.h>
#include <pthread.h>
#include <semaphore.h>

#define MAX_BUFFER_SIZE 10
#define SUCCESS 0

struct Sensor {
    int id;
    float value;
};

struct ring_buffer {
    struct Sensor *buffer;
    atomic_int head;      // Write index
    atomic_int tail;      // Read index
    int max;

    // Synchronization primitives
    pthread_mutex_t mtx;  // Protects concurrent structural modifications if needed
    sem_t slots_sem;      // Counts available empty slots (for producers)
    sem_t items_sem;      // Counts available items to read (for consumers)
};

// --- Initialization ---
struct ring_buffer* create_ring_buffer(void) {
    struct ring_buffer *ring = malloc(sizeof(struct ring_buffer));
    if (!ring) return NULL;

    ring->buffer = malloc(sizeof(struct Sensor) * MAX_BUFFER_SIZE);
    if (!ring->buffer) {
        free(ring);
        return NULL;
    }

    atomic_init(&ring->head, 0);
    atomic_init(&ring->tail, 0);
    ring->max = MAX_BUFFER_SIZE;

    pthread_mutex_init(&ring->mtx, NULL);
    
    // Initialize semaphores: slots starts at MAX_BUFFER_SIZE, items starts at 0
    sem_init(&ring->slots_sem, 0, MAX_BUFFER_SIZE);
    sem_init(&ring->items_sem, 0, 0);

    return ring;
}

void destroy_ring_buffer(struct ring_buffer *ring) {
    if (!ring) return;
    pthread_mutex_destroy(&ring->mtx);
    sem_destroy(&ring->slots_sem);
    sem_destroy(&ring->items_sem);
    free(ring->buffer);
    free(ring);
}

// --- Producer Thread Function (Multi-Producer Safe) ---
int mpmc_enqueue(struct ring_buffer *ring, struct Sensor sensor_data) {
    // 1. Wait for an available empty slot (blocking if buffer is full)
    if (sem_wait(&ring->slots_sem) != 0) {
        return -1; 
    }

    int current_head;
    int next_head;

    // 2. Atomically reserve a slot using Compare-And-Swap (CAS) loop
    do {
        current_head = atomic_load_explicit(&ring->head, memory_order_relaxed);
        next_head = (current_head + 1) % ring->max;
    } while (!atomic_compare_exchange_weak_explicit(
                &ring->head, &current_head, next_head,
                memory_order_release, memory_order_relaxed));

    // 3. Write data into the reserved slot safely under mutex or exclusive index offset
    // Note: To prevent multiple producers writing to the same slot index simultaneously 
    // before head updates propagate, a fine-grained slot-guard or mutex is ideal.
    pthread_mutex_lock(&ring->mtx);
    ring->buffer[current_head] = sensor_data;
    pthread_mutex_unlock(&ring->mtx);

    // 4. Signal consumers that a new item is ready
    sem_post(&ring->items_sem);

    return SUCCESS;
}

// --- Consumer Thread Function (Multi-Consumer Safe) ---
int mpmc_dequeue(struct ring_buffer *ring, struct Sensor *sensor_data) {
    // 1. Wait for an available item (blocking if buffer is empty)
    if (sem_wait(&ring->items_sem) != 0) {
        return -1;
    }

    int current_tail;
    int next_tail;

    // 2. Atomically reserve a slot to read using Compare-And-Swap (CAS) loop
    do {
        current_tail = atomic_load_explicit(&ring->tail, memory_order_relaxed);
        next_tail = (current_tail + 1) % ring->max;
    } while (!atomic_compare_exchange_weak_explicit(
                &ring->tail, &current_tail, next_tail,
                memory_order_release, memory_order_relaxed));

    // 3. Read data from the reserved slot safely
    pthread_mutex_lock(&ring->mtx);
    *sensor_data = ring->buffer[current_tail];
    pthread_mutex_unlock(&ring->mtx);

    // 4. Signal producers that an empty slot is now available
    sem_post(&ring->slots_sem);

    return SUCCESS;
}

MPMC Ring Buffer Implementation (Mutex, CondVar, Semaphores)

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

#define MAX_BUFFER_SIZE 10
#define SUCCESS 0
#define BUFFER_FULL -1
#define BUFFER_EMPTY -2

struct Sensor {
    int id;
    float value;
};

struct ring_buffer {
    struct Sensor *buffer;
    int head;
    int tail;
    int count;          // Tracks current number of items in the buffer
    int max;

    // Synchronization primitives
    pthread_mutex_t mtx;       // Protects shared state (head, tail, count, buffer array)
    pthread_cond_t not_full;   // Condition variable signaled when space becomes available
    pthread_cond_t not_empty;  // Condition variable signaled when items become available
    sem_t counting_sem;        // Alternative/supplementary counting semaphore
};

// --- Initialization ---
struct ring_buffer* create_ring_buffer(void) {
    struct ring_buffer *ring = malloc(sizeof(struct ring_buffer));
    if (!ring) return NULL;

    ring->buffer = malloc(sizeof(struct Sensor) * MAX_BUFFER_SIZE);
    if (!ring->buffer) {
        free(ring);
        return NULL;
    }

    ring->head = 0;
    ring->tail = 0;
    ring->count = 0;
    ring->max = MAX_BUFFER_SIZE;

    pthread_mutex_init(&ring->mtx, NULL);
    pthread_cond_init(&ring->not_full, NULL);
    pthread_cond_init(&ring->not_empty, NULL);
    
    // Initialize counting semaphore to max buffer size (representing available empty slots)
    sem_init(&ring->counting_sem, 0, MAX_BUFFER_SIZE);

    return ring;
}

void destroy_ring_buffer(struct ring_buffer *ring) {
    if (!ring) return;
    pthread_mutex_destroy(&ring->mtx);
    pthread_cond_destroy(&ring->not_full);
    pthread_cond_destroy(&ring->not_empty);
    sem_destroy(&ring->counting_sem);
    free(ring->buffer);
    free(ring);
}

// --- Producer Thread Function ---
int mpmc_enqueue(struct ring_buffer *ring, struct Sensor sensor_data) {
    // 1. Wait for an available slot using the counting semaphore 
    // (Alternatively, you could use pthread_cond_wait on ring->not_full inside the mutex)
    if (sem_wait(&ring->counting_sem) != 0) {
        return BUFFER_FULL;
    }

    // 2. Lock mutex for critical section access (head, tail, buffer array)
    pthread_mutex_lock(&ring->mtx);

    // Double check full condition under lock (safety standard)
    while (ring->count >= ring->max) {
        pthread_cond_wait(&ring->not_full, &ring->mtx);
    }

    // 3. Write data
    ring->buffer[ring->head] = sensor_data;
    ring->head = (ring->head + 1) % ring->max;
    ring->count++;

    // 4. Wake up a waiting consumer
    pthread_cond_signal(&ring->not_empty);

    pthread_mutex_unlock(&ring->mtx);

    return SUCCESS;
}

// --- Consumer Thread Function ---
int mpmc_dequeue(struct ring_buffer *ring, struct Sensor *sensor_data) {
    pthread_mutex_lock(&ring->mtx);

    // 1. Wait until the buffer has items to read
    while (ring->count == 0) {
        pthread_cond_wait(&ring->not_empty, &ring->mtx);
    }

    // 2. Read data
    *sensor_data = ring->buffer[ring->tail];
    ring->tail = (ring->tail + 1) % ring->max;
    ring->count--;

    // 3. Wake up a waiting producer
    pthread_cond_signal(&ring->not_full);

    pthread_mutex_unlock(&ring->mtx);

    // 4. Post to the counting semaphore to signal an available slot
    sem_post(&ring->counting_sem);

    return SUCCESS;
}

Circular Bufer with overflow - FIFO implementation

Circular queue with overwrite policy, Implement a fixed-size byte queue:

Requirements:

#define QUEUE_SIZE 8

struct byte_queue {
    uint8_t buf[QUEUE_SIZE];
    uint32_t head;
    uint32_t tail;
    uint32_t count;
};

int queue_push(struct byte_queue *q, uint8_t data);
int queue_pop(struct byte_queue *q, uint8_t *data);

/*
`queue_push()` adds one byte.
`queue_pop()` removes one byte.
If the queue is full, queue_push() must overwrite the oldest byte.
`queue_pop()` returns:
    0 on success
    -1 if the queue is empty
queue_push() returns:
    0 on success
    -1 for invalid input
Do not dynamically allocate memory.
*/

#include <stdio.h>
#include <stdint.h>
#include <stddef.h>

#define QUEUE_SIZE 5

struct byte_queue {
    uint8_t buf[QUEUE_SIZE];
    uint32_t head;   // Next index to write into
    uint32_t tail;   // Next index to read from
    uint32_t count;  // Current number of elements
};

#define SUCCESS          0
#define QUEUE_EMPTY     -1
#define INVALID_INPUT   -1

int queue_push(struct byte_queue *q, uint8_t data) {
    if (q == NULL) {
        return INVALID_INPUT;
    }
    
    // If the queue is full, overwrite the oldest byte (advance tail)
    if (q->count == QUEUE_SIZE) {
        q->tail = (q->tail + 1) % QUEUE_SIZE;
    } else {
        // Only increment count if we are not overwriting
        q->count++;
    }

    // Write data at the current head, then advance head
    q->buf[q->head] = data;
    q->head = (q->head + 1) % QUEUE_SIZE;
    
    return SUCCESS;
}

int queue_pop(struct byte_queue *q, uint8_t *data) {
    if (q == NULL || data == NULL) {
        return INVALID_INPUT;
    }

    if (q->count == 0) {
        return QUEUE_EMPTY;
    }

    // Read data from the current tail, then advance tail
    *data = q->buf[q->tail];
    q->tail = (q->tail + 1) % QUEUE_SIZE;
    q->count--;
    
    return SUCCESS;
}

void print_queue(struct byte_queue *q) {
    if (q == NULL) return;
    for (int i = 0; i < QUEUE_SIZE; i++) {
        printf("%d ", q->buf[i]);
    }
    printf("\n");
}

int main() {
    // Correct structure initialization
    struct byte_queue q = {0}; 
    uint8_t data; // Must be uint8_t to match queue_pop signature

    printf("Pushing 1, 2, 3, 4, 5:\n");
    queue_push(&q, 1);
    queue_push(&q, 2);
    queue_push(&q, 3);
    queue_push(&q, 4);
    queue_push(&q, 5);
    print_queue(&q); // Output: 1 2 3 4 5

    printf("\nPopping 2 elements:\n");
    queue_pop(&q, &data);
    queue_pop(&q, &data);
    // Buffer still holds data elements internally, but pointers restrict access
    print_queue(&q); 

    printf("\nPushing 6:\n");
    queue_push(&q, 6);
    print_queue(&q); // Output: 6 2 3 4 5

    printf("\nPushing 7:\n");
    queue_push(&q, 7);
    print_queue(&q); // Output: 6 7 3 4 5

    printf("\nPushing 8 (Triggers overwrite of oldest active element '3'):\n");
    queue_push(&q, 8);
    print_queue(&q); // Output: 6 7 8 4 5

    printf("\nPopping all active elements:\n");
    while (queue_pop(&q, &data) == SUCCESS) {
        printf("Popped: %d\n", data);
    }

    printf("\nAttempting to pop empty queue:\n");
    if (queue_pop(&q, &data) == QUEUE_EMPTY) {
        printf("Queue is successfully verified empty!\n");
    }

    return 0;
}