Type agnostic, statically allocated, thread-safe queue in C

Question

I need a queue in C for an embedded application in which I cannot allocate memory dynamically.

I avoided using void* for the queue elements (which is not convenient lifecycle-wise) by having the user provide their underlying array.

user_code.c:

#include "generic_queue.c"

typedef struct custom_t
{
    /* whatever */
} custom_t;

custom_t underlying_array[5];
queue_t queue;
queue_init(&queue, underlying_array, 5, sizeof(custom_t));
/* now use queue to do stuff */
queue_destroy(&queue);

generic_queue.h:

#ifndef INCLUDE_GUARD_GENERIC_QUEUE_H
#define INCLUDE_GUARD_GENERIC_QUEUE_H

#include <stddef.h>
#include <stdbool.h>
#include <pthread.h>

typedef struct queue_t
{
    void* array;
    size_t capacity;
    size_t sizeof_element;
    size_t in;
    size_t out;
    pthread_mutex_t mutex;
} queue_t;


/* return true on success */
bool queue_init(queue_t* queue, void* array, size_t capacity, size_t sizeof_element);
void queue_destroy(queue_t* queue);
bool queue_front(queue_t * queue, void *dest);
bool queue_back(queue_t * queue, void *dest);
bool queue_push(queue_t *queue, void const *src);
bool queue_pop(queue_t *queue, void *dest);

/* assert on failure */
size_t queue_size(queue_t * queue);
size_t queue_capacity(queue_t * queue);
bool queue_empty(queue_t * queue);
bool queue_full(queue_t * queue);

#endif /* INCLUDE_GUARD_GENERIC_QUEUE_H */

genric_queue.c:

#include "generic_queue.h"

#include <stddef.h>
#include <stdbool.h>
#include <string.h> /* memcpy */
#include <assert.h>
#include <pthread.h>


bool queue_init(queue_t* queue, void* array, size_t capacity, size_t sizeof_element)
{
    bool success;
    
    if ((queue == NULL) || (array == NULL) || (capacity < 2) || (sizeof_element == 0))
    {
        success = false;
    }
    else
    {
        pthread_mutex_init(&queue->mutex, NULL);
        pthread_mutex_lock(&queue->mutex);
        
        queue->array = array;
        queue->capacity = capacity;
        queue->sizeof_element = sizeof_element;
        queue->in = 0;
        queue->out = 0;
        success = true;
        
        pthread_mutex_unlock(&queue->mutex);
    }
    return success;
}

void queue_destroy(queue_t* queue)
{
   pthread_mutex_destroy(&queue->mutex);
   queue->array = NULL;
   queue->capacity = 0;
   queue->sizeof_element = 0;
   queue->in = 0;
   queue->out = 0;
}

static void* index_to_address(queue_t const * queue, size_t index)
{
    void* address;
    assert(queue != NULL);
    
    /* No mutex here: this is an internal utility function */
    address = ((unsigned char*)queue->array) + (index * queue->sizeof_element);
    
    return address;
}

static bool queue_empty_internal(queue_t * queue, bool lock_mutex)
{
    bool empty;
    assert(queue != NULL);
    
    if (lock_mutex) { pthread_mutex_lock(&queue->mutex); }
    empty = (queue->in == queue->out);
    if (lock_mutex) { pthread_mutex_unlock(&queue->mutex); }
    
    return empty;
}

bool queue_empty(queue_t * queue)
{
    return queue_empty_internal(queue, true);
}

/* When used internally, mutex might be already locked, so an additional "private" `lock_mutex` argument is needed */
static bool queue_front_internal(queue_t * queue, void *dest, bool lock_mutex)
{
    bool success;
    void *src;
    
    if ((queue == NULL) || (queue_empty_internal(queue, lock_mutex)) || (dest == NULL))
    {
        success = false;
    }
    else
    {
        if (lock_mutex) { pthread_mutex_lock(&queue->mutex); }
        
        src = index_to_address(queue, queue->out);
        memcpy(dest, src, queue->sizeof_element);
        success = true;
        
        if (lock_mutex) { pthread_mutex_unlock(&queue->mutex); }
    }
    return success;
}

bool queue_front(queue_t * queue, void *dest)
{
    return queue_front_internal(queue, dest, true);
}

bool queue_back(queue_t * queue, void *dest)
{
    bool success;
    size_t index;
    void *src;
    
    if ((queue == NULL) || (queue_empty(queue)) || (dest == NULL))
    {
        success = false;
    }
    else
    {
        pthread_mutex_lock(&queue->mutex);
        
        index = (queue->in == 0) ? queue->capacity - 1 : queue->in - 1;
        src = index_to_address(queue, index);
        memcpy(dest, src, queue->sizeof_element);
        success = true;
        
        pthread_mutex_unlock(&queue->mutex);
    }
    return success;
}

bool queue_push(queue_t *queue, void const *src)
{
    bool success;
    void *dest;
    
    if ((queue == NULL) || (queue_full(queue)) || (src == NULL))
    {
        success = false;
    }
    else
    {
        pthread_mutex_lock(&queue->mutex);
        
        dest = index_to_address(queue, queue->in);
        memcpy(dest, src, queue->sizeof_element);
        queue->in = (queue->in + 1) % queue->capacity;
        success = true;
        
        pthread_mutex_unlock(&queue->mutex);
    }
    return success;
}

bool queue_pop(queue_t *queue, void *dest)
{
    bool success;
    
    if ((queue == NULL) || (queue_empty(queue)))
    {
        success = false;
    }
    else
    {
        pthread_mutex_lock(&queue->mutex);
        
        if (dest != NULL)
        {
            queue_front_internal(queue, dest, false);
        }
        queue->out = (queue->out + 1) % queue->capacity;
        success = true;
        
        pthread_mutex_unlock(&queue->mutex);
    }
    return success;
}

size_t queue_size(queue_t * queue)
{
    size_t size;
    assert(queue != NULL);
    
    pthread_mutex_lock(&queue->mutex);
    
    if (queue->out <= queue->in)
    {
        size = queue->in - queue->out;
    }
    else
    {
        size = queue->in + queue->capacity - queue->out;
    }
    
    pthread_mutex_unlock(&queue->mutex);
    
    return size;
}

size_t queue_capacity(queue_t * queue)
{
    size_t capacity;
    assert(queue != NULL);
    
    pthread_mutex_lock(&queue->mutex);
    capacity = queue->capacity - 1;
    pthread_mutex_unlock(&queue->mutex);
    
    return capacity;
}

bool queue_full(queue_t * queue)
{
    bool full;
    assert(queue != NULL);
    
    pthread_mutex_lock(&queue->mutex);
    full = (((queue->in + 1) % queue->capacity) == queue->out);
    pthread_mutex_unlock(&queue->mutex);
    
    return full;
}

Answer 1

Your code looks quite nice. I especially like that you have a very consistent API with clear names that follow the conventions of the C++ STL, and the pointer to the queue always as the first parameter. That said, there are some possible improvements.

Naming things

I would use queue_init() and queue_exit() here. Alternatively, use queue_create() and queue_destroy(), however I would use those names only if those actually allocated/freed a queue_t.

Instead of sizeof_element I would write write element_size, or perhaps just size, as the latter is already used in a similar way in standard C functions like qsort().

I would also use head and tail instead of in and out.

Be consistent using if or assert()

Some functions use an if-statement to check if parameters are valid, others just use assert(). Be consistent and always use the same method for a given parameter, like queue. And also make sure that if one function checks it, all of them. For example, queue_destroy() fails to check whether queue is NULL.

I advice you to use assert() for catching programming errors, and if-statements to catch runtime errors. I would consider queue being NULL a programming error. However, trying to insert into a full queue would be a runtime error, since what we put into a queue is most likely something we did not know up front when writing the program. The exception would be queue_destroy(), where I would use an if-statement to check for NULL, as this matches what free() does.

Did you consider checking the return value of pthread_mutex_init() and pthread_mutex_lock()? It so happens that in your code, you can avoid checking that, since you are not using recursive mutexes, but if you didn't know that, you should have either written if-statements to check whether these functions succeeded, or have read their manual pages to check whether they could fail.

Handle corner cases if possible

While it seems rather silly, there is no reason not to allow a queue of size 0 or 1, and elements of size 0. Handling corner cases correctly makes your code more robust.

Related to this:

Allow the whole queue to be used

You allow one less element in the queue than the capacity you pass to queue_init(). In fact:

int array[10];
queue_t queue;
queue_init(&queue, array, 10, sizeof(int));
int capacity = queue_capacity(queue); // will be 9!

Most likely you did this to work around the ambiguity when in == out. However, you can work around this by not having two indices into the array, but rather one offset and one count of the number of elements in the queue.

Use const whereever it is possible

You should mark parameters as const whenever it is possible to do so. For example, queue_empty() does not modify the queue at all, so the parameter queue should be a const queue_t *.

Avoid locking the mutex unnecessarily

There is no need to lock the mutex in queue_capacity(), as the capacity should never change between queue_init() and queue_destroy().

Make a clear distinction between locked and unlocked functions

Instead of having a queue_empty_internal() that takes a parameter whether to lock the mutex or not, I would write a queue_empty_unlocked() that never locks, and then make sure queue_empty() takes the lock itself before calling queue_empty_unlocked(). I would make it so that all user visible API functions take the lock, and all internal functions not take locks.

Consider returning a pointer instead of copying an element

Instead of queue_front() returning a bool and copying the contents to dest, I would rather have it return a pointer to the front element, or return NULL if the queue is empty. This is especially useful if the elements are very large, or if you want to modify an element that is in the queue.

Answer 2

Not much to add to the good code and the good review of @G. Sliepen

No error checking of pthread_mutex...() calls

I'd expect return values checked.

Consider an apply() function

Consider a function that applies a function to each queue element. Maybe ending early if an error detected.

bool queue_apply(queue_t * queue, int (*apply)(void *state, void *ellement), void * state);

Useful for printing and others tasks.

Some functional documentation in the *.h

Consider a user should get an idea of code use just by looking at the .h file. Example: It is unclear what the below do. I recommend at least a one line descripting for each function and an over description for the set.

bool queue_front(queue_t * queue, void *dest);
bool queue_back(queue_t * queue, void *dest);

Alternative style

Consider this style order of parameters in function declarations should be arranged such that the size of an array appears before the array

// bool queue_init(queue_t* queue, void* array, size_t capacity, size_t sizeof_element);
bool queue_init(queue_t* queue, size_t capacity, void* array, size_t sizeof_element);