spsc_lockfree.h

Go to the documentation of this file.

/*
 * Copyright (c) 2023 Intel Corporation
 *
 * SPDX-License-Identifier: Apache-2.0
 */
 
#ifndef ZEPHYR_SYS_SPSC_LOCKFREE_H_
#define ZEPHYR_SYS_SPSC_LOCKFREE_H_
 
#include <stdint.h>
#include <stdbool.h>
#include <zephyr/toolchain/common.h>
#include <zephyr/sys/atomic.h>
#include <zephyr/sys/util_macro.h>
 
struct spsc {
        /* private value only the producer thread should mutate */
        unsigned long acquire;
 
        /* private value only the consumer thread should mutate */
        unsigned long consume;
 
        /* producer mutable, consumer readable */
        atomic_t in;
 
        /* consumer mutable, producer readable */
        atomic_t out;
 
        /* mask used to automatically wrap values */
        const unsigned long mask;
};
 
#define SPSC_INITIALIZER(sz, buf)                                                                  \
        {                                                                                          \
                ._spsc =                                                                           \
                        {                                                                          \
                                .acquire = 0,                                                      \
                                .consume = 0,                                                      \
                                .in = ATOMIC_INIT(0),                                              \
                                .out = ATOMIC_INIT(0),                                             \
                                .mask = sz - 1,                                                    \
                        },                                                                         \
                .buffer = buf,                                                                     \
        }
 
#define SPSC_DECLARE(name, type)                                                                   \
        static struct spsc_##name {                                                                \
                struct spsc _spsc;                                                                 \
                type * const buffer;                                                               \
        }
 
#define SPSC_DEFINE(name, type, sz)                                                                \
        BUILD_ASSERT(IS_POWER_OF_TWO(sz));                                                         \
        static type __spsc_buf_##name[sz];                                                         \
        SPSC_DECLARE(name, type) name = SPSC_INITIALIZER(sz, __spsc_buf_##name);
 
#define spsc_size(spsc) ((spsc)->_spsc.mask + 1)
 
#define z_spsc_mask(spsc, i) ((i) & (spsc)->_spsc.mask)
 
#define z_spsc_in(spsc) (unsigned long)atomic_get(&(spsc)->_spsc.in)
 
#define z_spsc_out(spsc) (unsigned long)atomic_get(&(spsc)->_spsc.out)
 
#define spsc_reset(spsc)                                                                           \
        ({                                                                                         \
                (spsc)->_spsc.consume = 0;                                                         \
                (spsc)->_spsc.acquire = 0;                                                         \
                atomic_set(&(spsc)->_spsc.in, 0);                                                  \
                atomic_set(&(spsc)->_spsc.out, 0);                                                 \
        })
 
#define spsc_acquire(spsc)                                                                         \
        ({                                                                                         \
                unsigned long idx = z_spsc_in(spsc) + (spsc)->_spsc.acquire;                       \
                bool spsc_acq = (idx - z_spsc_out(spsc)) < spsc_size(spsc);                        \
                if (spsc_acq) {                                                                    \
                        (spsc)->_spsc.acquire += 1;                                                \
                }                                                                                  \
                spsc_acq ? &((spsc)->buffer[z_spsc_mask(spsc, idx)]) : NULL;                       \
        })
 
#define spsc_produce(spsc)                                                                         \
        ({                                                                                         \
                if ((spsc)->_spsc.acquire > 0) {                                                   \
                        (spsc)->_spsc.acquire -= 1;                                                \
                        atomic_add(&(spsc)->_spsc.in, 1);                                          \
                }                                                                                  \
        })
 
#define spsc_produce_all(spsc)                                                                     \
        ({                                                                                         \
                if ((spsc)->_spsc.acquire > 0) {                                                   \
                        unsigned long acquired = (spsc)->_spsc.acquire;                            \
                        (spsc)->_spsc.acquire = 0;                                                 \
                        atomic_add(&(spsc)->_spsc.in, acquired);                                   \
                }                                                                                  \
        })
 
#define spsc_drop_all(spsc)                                                                        \
        do {                                                                                       \
                (spsc)->_spsc.acquire = 0;                                                         \
        } while (false)
 
#define spsc_consume(spsc)                                                                         \
        ({                                                                                         \
                unsigned long idx = z_spsc_out(spsc) + (spsc)->_spsc.consume;                      \
                bool has_consumable = (idx != z_spsc_in(spsc));                                    \
                if (has_consumable) {                                                              \
                        (spsc)->_spsc.consume += 1;                                                \
                }                                                                                  \
                has_consumable ? &((spsc)->buffer[z_spsc_mask(spsc, idx)]) : NULL;                 \
        })
 
#define spsc_release(spsc)                                                                         \
        ({                                                                                         \
                if ((spsc)->_spsc.consume > 0) {                                                   \
                        (spsc)->_spsc.consume -= 1;                                                \
                        atomic_add(&(spsc)->_spsc.out, 1);                                         \
                }                                                                                  \
        })
 
#define spsc_release_all(spsc)                                                                     \
        ({                                                                                         \
                if ((spsc)->_spsc.consume > 0) {                                                   \
                        unsigned long consumed = (spsc)->_spsc.consume;                            \
                        (spsc)->_spsc.consume = 0;                                                 \
                        atomic_add(&(spsc)->_spsc.out, consumed);                                  \
                }                                                                                  \
        })
 
#define spsc_acquirable(spsc)                                                                      \
        ({ (((spsc)->_spsc.in + (spsc)->_spsc.acquire) - (spsc)->_spsc.out) - spsc_size(spsc); })
 
#define spsc_consumable(spsc) ({ (spsc)->_spsc.in - (spsc)->_spsc.out - (spsc)->_spsc.consume; })
 
#define spsc_peek(spsc)                                                                            \
        ({                                                                                         \
                unsigned long idx = z_spsc_out(spsc) + (spsc)->_spsc.consume;                      \
                bool has_consumable = (idx != z_spsc_in(spsc));                                    \
                has_consumable ? &((spsc)->buffer[z_spsc_mask(spsc, idx)]) : NULL;                 \
        })
 
#define spsc_next(spsc, item)                                                                      \
        ({                                                                                         \
                unsigned long idx = ((item) - (spsc)->buffer);                                     \
                bool has_next =                                                                    \
                        z_spsc_mask(spsc, (idx + 1)) != (z_spsc_mask(spsc, z_spsc_in(spsc)));      \
                has_next ? &((spsc)->buffer[z_spsc_mask((spsc), idx + 1)]) : NULL;                 \
        })
 
#define spsc_prev(spsc, item)                                                                      \
        ({                                                                                         \
                unsigned long idx = ((item) - &(spsc)->buffer[0]) / sizeof((spsc)->buffer[0]);     \
                bool has_prev = idx != z_spsc_mask(spsc, z_spsc_out(spsc));                        \
                has_prev ? &((spsc)->buffer[z_spsc_mask(spsc, idx - 1)]) : NULL;                   \
        })
 
#endif /* ZEPHYR_SYS_SPSC_LOCKFREE_H_ */