Multi-threaded circular buffer library in C that supports simultaneous reading and writing operations

concurrent_circular_buffer.c

/*
	Multi-threaded circular buffer library in C that supports simultaneous
	reading and writing operations. It provides options to overwrite the oldest data
	when the buffer is full or to skip new data based on a flag. The library includes
	functions to initialize, write, read, and clean up the circular buffer, and also
	demonstrates concurrent access with multiple writer and reader threads.

	The code is written in standard C and can be compiled and executed on various platforms, including Linux, Windows, and macOS. The code does not inherently depend on any specific platform-dependent features, libraries, or APIs. It uses standard C libraries and pthreads for multithreading, which are commonly available on many systems.

    Author : Eray Ozturk | [email protected]
*/

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <stdbool.h>
#include <string.h>
#include <unistd.h> // for sleep function

#define MAX_WRITE_SIZE 10

typedef struct {
    char *buffer;
    int size;   // Size of the buffer
    int head;
    int tail;
    bool overwrite; // Flag to enable overwriting or skipping new data
    pthread_mutex_t mutex;
    pthread_cond_t not_full; // Condition variable for buffer not being full
    pthread_cond_t not_empty; // Condition variable for buffer not being empty
} CircularBuffer;

// Initialize the circular buffer
void init_buffer(CircularBuffer *cb, int size, bool overwrite) {
    cb->size = size;
    cb->buffer = (char *)malloc(size * sizeof(char));
    cb->head = 0;
    cb->tail = 0;
    cb->overwrite = overwrite;
    pthread_mutex_init(&cb->mutex, NULL);
    pthread_cond_init(&cb->not_full, NULL);
    pthread_cond_init(&cb->not_empty, NULL);
}

// Write data to the circular buffer
void write_buffer(CircularBuffer *cb, const char *data, int size) {
    pthread_mutex_lock(&cb->mutex);

    // Wait until there's enough space in the buffer
    while ((cb->head + size) % BUFFER_SIZE == cb->tail) {
        if (cb->overwrite) {
            cb->tail = (cb->tail + 1) % BUFFER_SIZE; // Overwrite oldest data
        } else {
            pthread_mutex_unlock(&cb->mutex);
            return; // Skip new data when buffer is full and overwrite is disabled
        }
    }

    // Copy data into the buffer
    for (int i = 0; i < size; i++) {
        cb->buffer[cb->head] = data[i];
        cb->head = (cb->head + 1) % BUFFER_SIZE;
    }

    // Signal that buffer is not empty anymore
    pthread_cond_signal(&cb->not_empty);
    pthread_mutex_unlock(&cb->mutex);
}

// Read data from the circular buffer
void read_buffer(CircularBuffer *cb, char *data, int size) {
    pthread_mutex_lock(&cb->mutex);

    // Wait until there's data available in the buffer
    while (cb->head == cb->tail) {
        pthread_cond_wait(&cb->not_empty, &cb->mutex);
    }

    // Copy data from the buffer
    for (int i = 0; i < size; i++) {
        data[i] = cb->buffer[cb->tail];
        cb->tail = (cb->tail + 1) % BUFFER_SIZE;
    }

    // Signal that buffer is not full anymore
    pthread_cond_signal(&cb->not_full);
    pthread_mutex_unlock(&cb->mutex);
}

// Check if the circular buffer is full
bool is_buffer_full(CircularBuffer *cb) {
    return ((cb->head + 1) % cb->size == cb->tail);
}

// Check if the circular buffer is empty
bool is_buffer_empty(CircularBuffer *cb) {
    return (cb->head == cb->tail);
}

// Calculate available space in the circular buffer
int available_space(CircularBuffer *cb) {
    return (cb->size + cb->tail - cb->head - 1) % cb->size;
}

// Check if there is enough space to write data of given size
bool can_write(CircularBuffer *cb, int data_size) {
    return (available_space(cb) >= data_size);
}

// Cleanup resources of the circular buffer
void cleanup_buffer(CircularBuffer *cb) {
	free(cb->buffer);
    pthread_mutex_destroy(&cb->mutex);
    pthread_cond_destroy(&cb->not_full);
    pthread_cond_destroy(&cb->not_empty);
}

// Writer thread function
void *writer_thread(void *arg) {
    CircularBuffer *cb = (CircularBuffer *)arg;
    for (int i = 0; i < 20; i++) {
        char data[MAX_WRITE_SIZE];
        snprintf(data, MAX_WRITE_SIZE, "Data%d", i);
        int data_size = strlen(data);
        write_buffer(cb, data, data_size);
        printf("Thread %lu wrote: %s\n", pthread_self(), data);
        usleep(100000); // Simulate some delay
    }
    return NULL;
}

// Reader thread function
void *reader_thread(void *arg) {
    CircularBuffer *cb = (CircularBuffer *)arg;
    for (int i = 0; i < 20; i++) {
        char data[MAX_WRITE_SIZE];
        read_buffer(cb, data, MAX_WRITE_SIZE);
        printf("Thread %lu read: %s\n", pthread_self(), data);
        usleep(200000); // Simulate some delay
    }
    return NULL;
}

int main() {
    // Initialize circular buffers with different overwrite settings
    CircularBuffer cb_overwrite, cb_skip;
    init_buffer(&cb_overwrite, 50, true); // Enable overwrite
    init_buffer(&cb_skip, 80, false);     // Disable overwrite

    // Create writer and reader threads
    pthread_t writers[5], readers[2];
    for (int i = 0; i < 5; i++) {
        pthread_create(&writers[i], NULL, writer_thread, &cb_overwrite);
    }
    for (int i = 0; i < 2; i++) {
        pthread_create(&readers[i], NULL, reader_thread, &cb_skip);
    }

    // Wait for threads to finish
    for (int i = 0; i < 5; i++) {
        pthread_join(writers[i], NULL);
    }
    for (int i = 0; i < 2; i++) {
        pthread_join(readers[i], NULL);
    }

    // Clean up resources
    cleanup_buffer(&cb_overwrite);
    cleanup_buffer(&cb_skip);

    return 0;
}