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/sys/atomic.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(name, type, sz)     \
        { ._spsc = {                              \
                  .acquire = 0,                   \
                  .consume = 0,                   \
                  .in = ATOMIC_INIT(0),           \
                  .out = ATOMIC_INIT(0),          \
                  .mask = sz - 1,                 \
          }                                       \
        }
 
#define RTIO_SPSC_DECLARE(name, type, sz) \
        struct rtio_spsc_ ## name {       \
                struct rtio_spsc _spsc;   \
                type buffer[sz];          \
        }
 
#define RTIO_SPSC_DEFINE(name, type, sz)                                                           \
        RTIO_SPSC_DECLARE(name, type, sz) name = RTIO_SPSC_INITIALIZER(name, type, sz);
 
#define rtio_spsc_size(spsc) ((spsc)->_spsc.mask + 1)
 
#define z_rtio_spsc_mask(spsc, i) ((i) & (spsc)->_spsc.mask)
 
#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 = atomic_get(&(spsc)->_spsc.in) + (spsc)->_spsc.acquire;         \
                bool acq = (idx - atomic_get(&(spsc)->_spsc.out)) < rtio_spsc_size(spsc);          \
                if (acq) {                                                                         \
                        (spsc)->_spsc.acquire += 1;                                                \
                }                                                                                  \
                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 = atomic_get(&(spsc)->_spsc.out) + (spsc)->_spsc.consume;        \
                bool has_consumable = (idx != atomic_get(&(spsc)->_spsc.in));                      \
                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 = atomic_get(&(spsc)->_spsc.out) + (spsc)->_spsc.consume;        \
                bool has_consumable = (idx != atomic_get(&(spsc)->_spsc.in));                      \
                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, atomic_get(&(spsc)->_spsc.in)));           \
                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, atomic_get(&(spsc)->_spsc.out));     \
                has_prev ? &((spsc)->buffer[z_rtio_spsc_mask(spsc, idx - 1)]) : NULL;              \
        })
 
#endif /* ZEPHYR_RTIO_SPSC_H_ */