2
\$\begingroup\$

I have a single threaded embedded system with nested interrupts.

  • A writer interrupt will periodically update some global data structure with data.
  • Reader interrupts / calls from main thread will read the "latest updated value" and do something with it.
  • Sometimes, a value can be written from the main thread as well.
  • The writer may or may not be faster than the readers, so a bounded queue is out of question.
  • It is okay if readers sometimes get a stale value, or if some writes fail (in case of multiple writes).
  • Dynamic memory allocation (malloc) is not an option.

I have the following implementation.

/** \file double_buffer.h
 *
 * Nested multiple-producer multiple-consumer double buffer to store the most
 * recent value with static storage.
 *
 * Usage:
 * \code{.c}
 * typedef struct {...} MyStruct;
 *
 * // Must be initialized to some sane value as this can be returned to a
 * // reader initially.
 * static MyStruct array[2];
 *
 * DoubleBuffer global_latest_value = DOUBLE_BUFFER_STATIC_INIT(
 *         sizeof(MyStruct), array);
 *
 * void handler_writer(void) {
 *     // some interrupt, eg systick, or value from some sensor
 *     MyStruct *latest = DoubleBuffer_write_acquire(&global_latest_value);
 *     if (latest != NULL) {
 *         // write value into *latest
 *     }
 *     DoubleBuffer_write_commit(&global_latest_value, latest);
 * }
 *
 * void handler2(void) {
 *     // periodically check the latest value
 *     MyStruct *latest = DoubleBuffer_read_acquire(&global_latest_value);
 *     ... // Do something with *latest
 *     DoubleBuffer_read_release(&global_latest_value, latest);
 * }
 *
 * void handler_n(void) {
 *     // Can be just like handler2
 *     handler2();
 * }
 *
 * // handler_writer, handler2, ... handler_n can have any priorities. Any of
 * // them can interrupt any other handler.
 * \endcode
 *
 * \note acquire() and commit()/release() must happen in a nested order. The
 * best way to ensure this is that each function calls acqurire() only once and
 * calls the corresponding commit()/release() before it returns.
 */

#ifndef DOUBLE_BUFFER_H
#define DOUBLE_BUFFER_H 1
#include <stdatomic.h>
#include <stddef.h>

_Static_assert(
        ATOMIC_POINTER_LOCK_FREE,
        "Your stdlib implementation does not have lock-free pointer atomics");
_Static_assert(
        ATOMIC_INT_LOCK_FREE,
        "Your stdlib implementation does not have lock-free atomic int");


/** \brief Internal data structure of the nested MPMC double buffer
 *
 * This must be initialized with #DOUBLE_BUFFER_STATIC_INIT at declaration.
 */
typedef struct DoubleBuffer {
    /** Pointer to current slot in #data being read */
    void *_Atomic selected_read;
    /** Pointer to next slot in #data that can be read from */
    void *_Atomic next_read;
    /** Data array (of length 2) forming the double buffer */
    void *const data;
    /** Size of a slot in #data */
    const size_t elem_size;
    /** Number of readers currently reading */
    _Atomic int n_readers;
    /** Write mutex that allows only one writer at a time */
    atomic_flag write_mutex;
} DoubleBuffer;


/** \brief Statically initialize a #DoubleBuffer
 *
 * \param p_elem_size  size of one element of \p p_data_array
 * \param p_data_array data array (of length 2) to use as the double buffer
 *
 * \return A #DoubleBuffer static initializer
 */
#define DOUBLE_BUFFER_STATIC_INIT(p_elem_size, p_data_array)      \
    {                                                             \
        .data = p_data_array, .elem_size = p_elem_size,           \
        .selected_read = p_data_array, .next_read = p_data_array, \
        .n_readers = 0, .write_mutex = ATOMIC_FLAG_INIT           \
    }


/** \brief Acquire a slot for writing
 *
 * \param db #DoubleBuffer to acquire the slot from
 *
 * \return Pointer to an available slot in \p db->data
 * \retval NULL if another writer has acquired the slot first
 *
 * \pre \p db must be initialized with #DOUBLE_BUFFER_STATIC_INIT
 * \post #DoubleBuffer_write_commit() must be called after writing to the
 * returned slot.
 */
void *DoubleBuffer_write_acquire(DoubleBuffer *db);


/** \brief Commit a slot previously acquired for writing
 *
 * \param db   #DoubleBuffer from which \p slot was acquired
 * \param slot slot acquired by #DoubleBuffer_write_acquire() or NULL
 *
 * \pre #DoubleBuffer_write_commit() must be called for any intermediate slots
 * acquired after the corresponding #DoubleBuffer_write_acquire() call
 */
void DoubleBuffer_write_commit(DoubleBuffer *db, void *slot);


/** \brief Acquire a slot for reading
 *
 * \param db #DoubleBuffer to acquire the slot from
 *
 * \return Pointer to an available slot in \p db->data for reading
 *
 * \pre \p db must be initialized with #DOUBLE_BUFFER_STATIC_INIT
 * \post #DoubleBuffer_read_release() must be called after using the slot
 */
void *DoubleBuffer_read_acquire(DoubleBuffer *db);


/** \brief Release a slot previously acquired for reading
 *
 * \param db   #DoubleBuffer from which \p slot was acquired
 * \param slot slot acquired by #DoubleBuffer_read_acquire() or NULL
 *
 * \pre #DoubleBuffer_read_release() must be called for any intermediate slots
 * acquired after the corresponding #DoubleBuffer_read_acquire() call
 */
void DoubleBuffer_read_release(DoubleBuffer *db, const void *slot);


#endif /* ifndef BUFFER_H */

/** \file double_buffer.c
 *
 * Nested multiple-producer multiple-consumer double buffer to store the most
 * recent value with static storage.
 */
#include "double_buffer.h"


void *DoubleBuffer_write_acquire(DoubleBuffer *db) {
    if (atomic_flag_test_and_set(&db->write_mutex)) {
        /* Another writer is writing */
        return NULL;
    } else {
        /* Now we are the exclusive writer */
        /* Write the current slot being read into the next slot to read, so
         * that that no reader will touch the remainig slot.
         * This must be done in a loop since a reader can change the current
         * slot begin read between the atomic_load and the atomic_exchange.
         * This is *NOT* async-safe, but only async-interrupt-safe */
        void *last_selected;
        do {
            last_selected = atomic_load(&db->selected_read);
        } while (last_selected
                 != atomic_exchange(&db->next_read, last_selected));
        /* Now both current read and next to read point to the same slot. We
         * will now actually acquire the other slot for writing. Readers can
         * now keep reading from the last_selected slot. */
        void *acquired = last_selected == db->data ?
                                 (char *)db->data + db->elem_size :
                                 db->data;
        return acquired;
    }
}


void DoubleBuffer_write_commit(DoubleBuffer *db, void *slot) {
    if (slot == NULL) return;
    /* It's up to the caller to ensure correct slot pointer is passed */
    atomic_store(&db->next_read, slot);
    atomic_flag_clear(&db->write_mutex);
}


void *DoubleBuffer_read_acquire(DoubleBuffer *db) {
    if (0 == atomic_fetch_add(&db->n_readers, 1)) {
        /* We are the first reader */
        /* We check if there is a new slot with updated data, and set it as the
         * slot to read from. All readers that can interrupt this reader after
         * this step will use the updated value.
         * This must be done in a loop till both values stabilize, as a writer
         * may change the next slot to read from between the atomic_load and
         * the atomic_exchange.
         * This is *NOT* async-safe, but only async-interrupt-safe */
        void *last_next_read;
        do {
            last_next_read = atomic_load(&db->next_read);
        } while (last_next_read
                 != atomic_exchange(&db->selected_read, last_next_read));
    }
    /* Now, just pick the slot selected for reading */
    return atomic_load(&db->selected_read);
}


void DoubleBuffer_read_release(DoubleBuffer *db, const void *slot) {
    if (slot == NULL) return;
    /* We don't really care about the value of slot since all readers will be
     * reading from the same slot (only the first reader changes the slot) */
    atomic_fetch_sub(&db->n_readers, 1);
}

So, are there any bugs/data races in this? Is there any better implementation?

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.