7
\$\begingroup\$

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;
}
\$\endgroup\$
5
  • \$\begingroup\$ Is fifo_try_increment_head_atomically called from a high priority ISR? \$\endgroup\$
    – vnp
    Jun 11, 2018 at 3:29
  • \$\begingroup\$ @vnp it's called from fifo_try_enqueue, but yes, any interrupt can enqueue a message, and they get dequeued in the main function (lowest priority). \$\endgroup\$
    – vgru
    Jun 11, 2018 at 5:31
  • \$\begingroup\$ How then the while (true) loop is supposed to stop? \$\endgroup\$
    – vnp
    Jun 11, 2018 at 5:57
  • \$\begingroup\$ @vnp: It's likely there is a problem here I am not seeing, but can you clarify? The way I see it, if head is 0 at the beginning, one irq handler will get 1, the other one 2 at the exit of this function. atomic_compare_exchange loads fifo->head into previous if it fails, so the loop starts with a "fresh" previous on each iteration, if that's what you are asking? \$\endgroup\$
    – vgru
    Jun 11, 2018 at 6:47
  • \$\begingroup\$ @Groo Can you provide the latest implementation ? I'm facing the exact issues in terms of scenarios that you have faced. \$\endgroup\$ Oct 6, 2021 at 11:31

1 Answer 1

3
\$\begingroup\$

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.

\$\endgroup\$
4
  • \$\begingroup\$ Thanks for the suggestion! But if something "continually prevents the lower priority producer from executing", does it make sense to say that consumer will have nothing to consume because of the algorithm? I.e. if you are starving the producer of time needed to produce an item, then the item won't be actually produced, regardless of the algorithm? The consumer interrupt doesn't block when no data is available (if (!entry->written) return false;). \$\endgroup\$
    – vgru
    Jun 13, 2018 at 8:39
  • \$\begingroup\$ Suppose the producers run on one core, and the consumer runs on a separate core. If the producers get into a situation where the low priority producer has increased head without setting written, then the whole system becomes blocked. A proper lock-free algorithm would be able to make progress no matter what happened to any thread of execution, even if one thread is killed permanently. \$\endgroup\$
    – JS1
    Jun 14, 2018 at 4:24
  • \$\begingroup\$ I agree, that's a good point, in this case we have a specific situation that it's a single core system no RTOS, i.e. no threading. Since timing requirements are rather strict, we are also not allowed to disable interrupts while mutating shared data, so I was looking for ways to make some kind of a pool which would work for this specific case. I wasn't able to find any working implementation of a multiple-producer/multiple-consumer pool which would work without disabling interrupts, without dynamic allocation, and still work if being written to from multiple level interrupts. \$\endgroup\$
    – vgru
    Jun 14, 2018 at 8:49
  • \$\begingroup\$ @Groo I added a section with recommendations based on your specific setup. \$\endgroup\$
    – JS1
    Jun 14, 2018 at 19:48

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.