Reusable header-only circular buffer library in C

cb.h

#pragma once

#include <stdint.h>
#include <string.h>
#include <stdbool.h>

// Defines
#define CBT_MIN(a, b) ((a) < (b) ? (a) : (b))

// Checks
#ifndef CBT_MUTEX_T
#error "Please define CBT_MUTEX_T before including cb.h"
#endif
#ifndef CBT_MUTEX_LOCK
#error "Please define CBT_MUTEX_LOCK before including cb.h, with signature `void (*lock)(CBT_MUTEX_T)`"
#endif
#ifndef CBT_MUTEX_UNLOCK
#error "Please define CBT_MUTEX_UNLOCK before including cb.h, with signature `void (*unlock)(CBT_MUTEX_T)`"
#endif

// Defines

// Typedefs
typedef struct CB_t {
    CBT_MUTEX_T lock;
    uint8_t* buffer;
    uint32_t capacity;
    uint32_t length;
    uint32_t write_index;
    uint32_t read_index;
} CB_t;

typedef struct CBQ_t {
    CBT_MUTEX_T lock;
    void* buffer;
    uint32_t capacity;
    uint32_t length;
    uint32_t element_size;
    // The back index points at the *next* element to be added. It's the traditional "write" index
    uint32_t back_index;
    // The front index points at the *next* element to be consumed. It's the traditional "read" index
    uint32_t front_index;
} CBQ_t;

// Public functions: cbt
void cbt_init(CB_t* cbt, CBT_MUTEX_T lock, uint8_t* buffer, uint32_t capacity);
void cbt_lock(CB_t* cbt);
uint32_t cbt_space_remaining(CB_t* cbt);
uint32_t cbt_contiguous_space_remaining(CB_t* cbt);
uint32_t cbt_contiguous_length(CB_t* cbt);
uint8_t* cbt_get_read_ptr(CB_t* cbt);
uint8_t* cbt_get_write_ptr(CB_t* cbt);

uint32_t cbt_write(CB_t* cbt, const uint8_t* data, uint32_t length);
uint32_t cbt_read(CB_t* cbt, uint8_t* data, uint32_t length);

// Public functions: cbq
bool cbq_init(CBQ_t* queue, CBT_MUTEX_T lock, void* buffer, uint32_t capacity, uint32_t element_size);
void cbq_lock(CBQ_t* queue);
void cbq_unlock(CBQ_t* queue);

// These functions do not require the user to explicitly take the lock. Receive functions copy the element into the user's buffer
bool cbq_send_to_back(CBQ_t* queue, const void* element);
bool cbq_send_to_front(CBQ_t* queue, const void* element);
bool cbq_receive_from_front(CBQ_t* queue, void* element);
bool cbq_receive_from_back(CBQ_t* queue, void* element);

// These functions give back a pointer to the element in the queue, and therefore care must be taken when using them
void* cbq_peek_back(CBQ_t* queue);
void* cbq_peek_front(CBQ_t* queue);

// These functions require the user to explicitly take the lock
void* cbq_get_writable_back_ptr(CBQ_t* queue);
void* cbq_get_writable_front_ptr(CBQ_t* queue);
void cbq_wrote_to_back(CBQ_t* queue);
void cbq_wrote_to_front(CBQ_t* queue);
void cbq_consume_back(CBQ_t* queue);
void cbq_consume_front(CBQ_t* queue);

// Implementation
#ifdef CBT_IMPLEMENTATION

// Public functions: cbq
void cbt_init(CB_t* cbt, CBT_MUTEX_T lock, uint8_t* buffer, uint32_t capacity) {
    cbt->lock = lock;
    cbt->buffer = buffer;
    cbt->capacity = capacity;
    cbt->length = 0;
    cbt->write_index = 0;
    cbt->read_index = 0;
}

void cbt_lock(CB_t* cbt) {
    CBT_MUTEX_LOCK(cbt->lock);
}

void cbt_unlock(CB_t* cbt) {
    CBT_MUTEX_UNLOCK(cbt->lock);
}

uint32_t cbt_space_remaining(CB_t* cbt) {
    return cbt->capacity - cbt->length;
}

uint32_t cbt_contiguous_space_remaining(CB_t* cbt) {
    uint32_t contiguous_space = cbt->capacity - cbt->write_index;
    return CBT_MIN(contiguous_space, cbt_space_remaining(cbt));
}

uint32_t cbt_contiguous_length(CB_t* cbt) {
    uint32_t contiguous = cbt->capacity - cbt->read_index;
    return CBT_MIN(contiguous, cbt->length);
}

uint8_t* cbt_get_read_ptr(CB_t* cbt) {
    return &cbt->buffer[cbt->read_index];
}

uint8_t* cbt_get_write_ptr(CB_t* cbt) {
    return &cbt->buffer[cbt->write_index];
}

uint32_t cbt_write(CB_t* cbt, const uint8_t* data, uint32_t length) {
    uint32_t space = cbt_space_remaining(cbt);
    uint32_t contiguous_space = cbt_contiguous_space_remaining(cbt);
    uint32_t write_length = CBT_MIN(space, length);
    uint32_t write_length_contiguous = CBT_MIN(contiguous_space, write_length);

    // Write to the end of the buffer
    memcpy(cbt_get_write_ptr(cbt), data, write_length_contiguous);
    cbt->write_index = (cbt->write_index + write_length_contiguous) % cbt->capacity;
    cbt->length += write_length_contiguous;

    // Wrap around if necessary
    if (write_length > write_length_contiguous) {
        uint32_t write_length_remaining = write_length - write_length_contiguous;
        memcpy(cbt->buffer, &data[write_length_contiguous], write_length_remaining);
        cbt->write_index = (cbt->write_index + write_length_remaining) % cbt->capacity;
        cbt->length += write_length_remaining;
    }

    return write_length;
}

uint32_t cbt_read(CB_t* cbt, uint8_t* data, uint32_t length) {
    uint32_t read_length = CBT_MIN(cbt->length, length);
    uint32_t contiguous_length = cbt_contiguous_length(cbt);
    uint32_t read_length_contiguous = CBT_MIN(contiguous_length, read_length);

    // Read from the beginning of the buffer
    memcpy(data, cbt_get_read_ptr(cbt), read_length_contiguous);
    cbt->read_index = (cbt->read_index + read_length_contiguous) % cbt->capacity;
    cbt->length -= read_length_contiguous;

    // Read from the end of the buffer
    if (read_length > read_length_contiguous) {
        uint32_t read_length_remaining = read_length - read_length_contiguous;
        memcpy(&data[read_length_contiguous], cbt->buffer, read_length_remaining);
        cbt->read_index = (cbt->read_index + read_length_remaining) % cbt->capacity;
        cbt->length -= read_length_remaining;
    }

    return read_length;
}

// Public functions: cbq
bool cbq_init(CBQ_t* queue, CBT_MUTEX_T lock, void* buffer, uint32_t capacity, uint32_t element_size) {
    queue->lock = lock;
    queue->buffer = buffer;
    queue->capacity = capacity;
    queue->element_size = element_size;
    queue->back_index = 0;
    queue->front_index = 0;
    queue->length = 0;

    return queue->lock != NULL;
}

void cbq_lock(CBQ_t* queue) {
    CBT_MUTEX_LOCK(queue->lock);
}

void cbq_unlock(CBQ_t* queue) {
    CBT_MUTEX_UNLOCK(queue->lock);
}


bool cbq_send_to_back(CBQ_t* queue, const void* element) {
    cbq_lock(queue);

    if (queue->length == queue->capacity) {
        cbq_unlock(queue);
        return false;
    }

    void* write_ptr = &((uint8_t*)queue->buffer)[queue->back_index * queue->element_size];
    memcpy(write_ptr, element, queue->element_size);
    queue->back_index = (queue->back_index + 1) % queue->capacity;
    queue->length++;
    cbq_unlock(queue);

    return true;
}

bool cbq_send_to_front(CBQ_t* queue, const void* element) {
    cbq_lock(queue);

    if (queue->length == queue->capacity) {
        cbq_unlock(queue);
        return false;
    }

    if (queue->front_index == 0) {
        queue->front_index = queue->capacity - 1;
    } else {
        queue->front_index--;
    }

    void* write_ptr = &((uint8_t*)queue->buffer)[queue->front_index * queue->element_size];
    memcpy(write_ptr, element, queue->element_size);
    queue->length++;
    cbq_unlock(queue);

    return true;
}

bool cbq_receive_from_front(CBQ_t* queue, void* element) {
    cbq_lock(queue);

    if (queue->length == 0) {
        cbq_unlock(queue);
        return false;
    }

    void* read_ptr = &((uint8_t*)queue->buffer)[queue->front_index * queue->element_size];
    memcpy(element, read_ptr, queue->element_size);
    queue->front_index = (queue->front_index + 1) % queue->capacity;
    queue->length--;
    cbq_unlock(queue);

    return true;
}

bool cbq_receive_from_back(CBQ_t* queue, void* element) {
    cbq_lock(queue);

    if (queue->length == 0) {
        cbq_unlock(queue);
        return false;
    }

    if (queue->back_index == 0) {
        queue->back_index = queue->capacity - 1;
    } else {
        queue->back_index--;
    }

    void* read_ptr = &((uint8_t*)queue->buffer)[queue->back_index * queue->element_size];
    memcpy(element, read_ptr, queue->element_size);
    queue->length--;
    cbq_unlock(queue);

    return true;
}

void cbq_consume_back(CBQ_t* queue) {
    if (queue->length == 0) {
        return;
    }

    if (queue->back_index == 0) {
        queue->back_index = queue->capacity - 1;
    } else {
        queue->back_index--;
    }

    queue->length--;
}

void cbq_consume_front(CBQ_t* queue) {
    if (queue->length == 0) {
        return;
    }

    queue->front_index = (queue->front_index + 1) % queue->capacity;
    queue->length--;
}


void* cbq_peek_back(CBQ_t* queue) {
    uint32_t index = queue->back_index == 0
        ? queue->capacity - 1
        : queue->back_index - 1;

    return &((uint8_t*)queue->buffer)[index * queue->element_size];
}

void* cbq_peek_front(CBQ_t* queue) {
    return &((uint8_t*)queue->buffer)[queue->front_index * queue->element_size];
}


void* cbq_get_writable_back_ptr(CBQ_t* queue) {
    return &((uint8_t*)queue->buffer)[queue->back_index * queue->element_size];
}

void* cbq_get_writable_front_ptr(CBQ_t* queue) {
    uint32_t index = queue->front_index == 0
        ? queue->capacity - 1
        : queue->front_index - 1;

    return &((uint8_t*)queue->buffer)[index * queue->element_size];
}

void cbq_wrote_to_back(CBQ_t* queue) {
    queue->back_index = (queue->back_index + 1) % queue->capacity;
    queue->length++;
}

void cbq_wrote_to_front(CBQ_t* queue) {
    queue->front_index = queue->front_index == 0
        ? queue->capacity - 1
        : queue->front_index - 1;

    queue->length++;
}

#endif // CBT_IMPLEMENTATION