rtio_spsc.h

Go to the documentation of this file.

/*
 * Copyright (c) 2022 Intel Corporation
 *
 * SPDX-License-Identifier: Apache-2.0
 */
 
 
#ifndef ZEPHYR_RTIO_SPSC_H_
#define ZEPHYR_RTIO_SPSC_H_
 
#include <stdint.h>
#include <stdbool.h>
#include <zephyr/toolchain/common.h>
#include <zephyr/sys/atomic.h>
#include <zephyr/sys/util_macro.h>
 
struct rtio_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 RTIO_SPSC_INITIALIZER(sz, buf)          \
        {                                       \
                ._spsc = {                      \
                        .acquire = 0,           \
                        .consume = 0,           \
                        .in = ATOMIC_INIT(0),   \
                        .out = ATOMIC_INIT(0),  \
                        .mask = sz - 1,         \
                },                              \
                .buffer = buf,                  \
        }
 
#define RTIO_SPSC_DECLARE(name, type)           \
        static struct rtio_spsc_##name {        \
                struct rtio_spsc _spsc; \
                type * const buffer;            \
        }
 
#define RTIO_SPSC_DEFINE(name, type, sz)                                \
        BUILD_ASSERT(IS_POWER_OF_TWO(sz));                              \
        static type __spsc_buf_##name[sz];                              \
        RTIO_SPSC_DECLARE(name, type) name = RTIO_SPSC_INITIALIZER(sz, __spsc_buf_##name);
 
#define rtio_spsc_size(spsc) ((spsc)->_spsc.mask + 1)
 
#define z_rtio_spsc_mask(spsc, i) ((i) & (spsc)->_spsc.mask)
 
 
#define z_rtio_spsc_in(spsc) (unsigned long)atomic_get(&(spsc)->_spsc.in)
 
#define z_rtio_spsc_out(spsc) (unsigned long)atomic_get(&(spsc)->_spsc.out)
 
#define rtio_spsc_reset(spsc)                                                                      \
        ({                                                                                         \
                (spsc)->_spsc.consume = 0;                                                         \
                (spsc)->_spsc.acquire = 0;                                                         \
                atomic_set(&(spsc)->_spsc.in, 0);                                                  \
                atomic_set(&(spsc)->_spsc.out, 0);                                                 \
        })
 
#define rtio_spsc_acquire(spsc)                                                                    \
        ({                                                                                         \
                unsigned long idx = z_rtio_spsc_in(spsc) + (spsc)->_spsc.acquire;                  \
                bool spsc_acq = (idx - z_rtio_spsc_out(spsc)) < rtio_spsc_size(spsc);              \
                if (spsc_acq) {                                                                    \
                        (spsc)->_spsc.acquire += 1;                                                \
                }                                                                                  \
                spsc_acq ? &((spsc)->buffer[z_rtio_spsc_mask(spsc, idx)]) : NULL;                  \
        })
 
#define rtio_spsc_produce(spsc)                                                                    \
        ({                                                                                         \
                if ((spsc)->_spsc.acquire > 0) {                                                   \
                        (spsc)->_spsc.acquire -= 1;                                                \
                        atomic_add(&(spsc)->_spsc.in, 1);                                          \
                }                                                                                  \
        })
 
#define rtio_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 rtio_spsc_drop_all(spsc)                \
        do {                                    \
                (spsc)->_spsc.acquire = 0;      \
        } while (false)
 
#define rtio_spsc_consume(spsc)                                                                    \
        ({                                                                                         \
                unsigned long idx = z_rtio_spsc_out(spsc) + (spsc)->_spsc.consume;                 \
                bool has_consumable = (idx != z_rtio_spsc_in(spsc));                               \
                if (has_consumable) {                                                              \
                        (spsc)->_spsc.consume += 1;                                                \
                }                                                                                  \
                has_consumable ? &((spsc)->buffer[z_rtio_spsc_mask(spsc, idx)]) : NULL;            \
        })
 
#define rtio_spsc_release(spsc)                                                                    \
        ({                                                                                         \
                if ((spsc)->_spsc.consume > 0) {                                                   \
                        (spsc)->_spsc.consume -= 1;                                                \
                        atomic_add(&(spsc)->_spsc.out, 1);                                         \
                }                                                                                  \
        })
 
 
#define rtio_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 rtio_spsc_acquirable(spsc)                                                                 \
        ({                                                                                         \
                (((spsc)->_spsc.in + (spsc)->_spsc.acquire) - (spsc)->_spsc.out) -                 \
                        rtio_spsc_size(spsc);                                                      \
        })
 
#define rtio_spsc_consumable(spsc)                                                                 \
        ({ (spsc)->_spsc.in - (spsc)->_spsc.out - (spsc)->_spsc.consume; })
 
#define rtio_spsc_peek(spsc)                                                                       \
        ({                                                                                         \
                unsigned long idx = z_rtio_spsc_out(spsc) + (spsc)->_spsc.consume;                 \
                bool has_consumable = (idx != z_rtio_spsc_in(spsc));                               \
                has_consumable ? &((spsc)->buffer[z_rtio_spsc_mask(spsc, idx)]) : NULL;            \
        })
 
#define rtio_spsc_next(spsc, item)                                                                 \
        ({                                                                                         \
                unsigned long idx = ((item) - (spsc)->buffer);                                     \
                bool has_next = z_rtio_spsc_mask(spsc, (idx + 1)) !=                               \
                                (z_rtio_spsc_mask(spsc, z_rtio_spsc_in(spsc)));                    \
                has_next ? &((spsc)->buffer[z_rtio_spsc_mask((spsc), idx + 1)]) : NULL;            \
        })
 
#define rtio_spsc_prev(spsc, item)                                                                 \
        ({                                                                                         \
                unsigned long idx = ((item) - &(spsc)->buffer[0]) / sizeof((spsc)->buffer[0]);     \
                bool has_prev = idx != z_rtio_spsc_mask(spsc, z_rtio_spsc_out(spsc));              \
                has_prev ? &((spsc)->buffer[z_rtio_spsc_mask(spsc, idx - 1)]) : NULL;              \
        })
 
#endif /* ZEPHYR_RTIO_SPSC_H_ */