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;
}
fifo_try_increment_head_atomically
called from a high priority ISR? \$\endgroup\$fifo_try_enqueue
, but yes, any interrupt can enqueue a message, and they get dequeued in the main function (lowest priority). \$\endgroup\$while (true)
loop is supposed to stop? \$\endgroup\$head
is0
at the beginning, one irq handler will get1
, the other one2
at the exit of this function.atomic_compare_exchange
loadsfifo->head
intoprevious
if it fails, so the loop starts with a "fresh" previous on each iteration, if that's what you are asking? \$\endgroup\$