Multiple-producer, single-consumer lock-free circular queue

Question

I'm working on a firmware on a bare-metal ARM single core processor (Cortex M4), with no RTOS. I need to add a multiple-producer, single-consumer, lockless fixed-size circular (ring) buffer to the firmware. It must handle multiple interrupts writing to the queue, but will be dequeued from a single, lowest level interrupt. However, since there is no threading in this code (i.e. no interrupt can "yield" to lower level interrupts), I cannot use spinlocks or mutexes; the only thing I can do is spin on CAS operations.

Anyway, after spending several hours writing this code, I've gotten to a point where I've simplified it to a point where I am starting to feel it's too simple:

The idea is basically that each producer increments the head pointer (modulo length) atomically, and once it does, no other producer can acquire this particular pointer (since the increment is atomic). So, as soon as the head pointer has been acquired, the producer needs to write the data to the slot and mark it as 'written'. Consumer, on the other hand, simply checks if the tail points to a 'written' entry.

So, these are the structs:

// using a multiple of 2 will allow compiler
// to optimize head/tail modulo operations
#define FIFO_LEN 128

// each fifo entry has a 'written' flag,
// which indicates it is ready to be consumed
typedef struct
{
    bool written;
    void * data;
}
entry_t;

// fifo contains an array of entries
// and two pointers
typedef struct
{
    entry_t entries[FIFO_LEN];
    int head;
    int tail;
}
fifo_t;

And here are the enqueue/dequeue functions:

// fifo is empty when head == tail
void fifo_init(fifo_t * fifo)
{
    fifo->head = 0;
    fifo->tail = 0;
}

// internal helper function -- returns false if the
// fifo is already full, otherwise increments the 
// head (modulo LEN) and returns true
bool fifo_try_increment_head_atomically(fifo_t *fifo, int *acquired_head)
{
    int previous = atomic_load(&fifo->head);

    while (true)
    {
        *acquired_head = (previous + 1) % FIFO_LEN;

        // if moving the head will overflow, 
        // it means there is no more space 
        if (*acquired_head == atomic_load(&fifo->tail))
            return false;

        // if we are the ones that succeeded, return true
        if (atomic_compare_exchange(&fifo->head, &previous, *acquired_head))
            return true;
    }
}

// if the fifo is full, returns false, otherwise
// enqueues the data into the fifo and returns true
bool fifo_try_enqueue(fifo_t *fifo, void * input_data)
{
    // if we cannot move the head, this means
    // the fifo is full
    int acquired_head;
    if (!fifo_try_increment_head_atomically(fifo, &acquired_head))
        return false;

    // at this point, we have moved the head atomically, 
    // so we simply need to make sure the entry is ready
    // and then set the 'written' flag
    entry_t * entry = &fifo->entries[acquired_head];
    entry->data = input_data;

    atomic_thread_fence(memory_order_release);
    entry->written = true;

    return true;
}

// if the fifo is empty, or the first available entry
// doesn't have the 'written' flag set, returns false.
// otherwise returns the data from the first entry, 
// removes the entry from the fifo, and returns true.
bool fifo_try_dequeue(fifo_t * fifo, void ** output_data)
{
    int head = atomic_load(&fifo->head);
    int tail = atomic_load(&fifo->tail);

    // fifo empty?
    if (head == tail)
        return false;

    // if we are here, there is at least one 
    // entry written (or being written)
    tail = (tail + 1) % FIFO_LEN;
    entry_t * entry = &fifo->entries[tail];

    // producer in the middle of writing?
    // return false without actually moving
    // fifo->tail
    if (!entry->written)
        return false;

    // if we are here, data is ready
    atomic_thread_fence(memory_order_acquire);
    *output_data = entry->data;
    entry->written = false;

    atomic_thread_fence(memory_order_release);
    atomic_store(&fifo->tail, tail);

    return true;
}

Answer 1

Lock-free guarantees

Although your program doesn't contain locks, I'm not sure it can really be considered "lock-free". Lock-free algorithms should guarantee to make system wide progress. For example according to this article:

In particular, if one thread is suspended, then a lock-free algorithm guarantees that the remaining threads can still make progress.

Your program looks to be correct in terms of enqueueing and dequeueing data. However, there are ways in which your system can fail to make progress.

For example, suppose a producer moves head forward and then stops executing before writing written. Once that happens, the consumer can no longer consume any data because it will be waiting for written to become true. In your system this could happen if a higher priority interrupt/producer interrupts a lower priority producer, and then continually prevents the lower priority producer from executing.

Perhaps in your actual system setup the above situation could ever occur. But from a theoretical standpoint, the code couldn't really be considered "lock-free".

Your specific situation

With regards to your specific setup:

• One core
• Consumer running on lowest priority interrupt
• Producers running on higher priority interrupts

I would recommend the following changes to simplify your code:

1. There is no need for memory barriers (aka thread fences) on a single core system.

2. You don't need the written field at all. As long as the consumer is running at a lower priority than the producers, each producer will completely finish enqueuing its data before the consumer will notice that head has moved forward.