rtio.h

Go to the documentation of this file.

/*
 * Copyright (c) 2022 Intel Corporation
 *
 * SPDX-License-Identifier: Apache-2.0
 */
 
#ifndef ZEPHYR_INCLUDE_RTIO_RTIO_H_
#define ZEPHYR_INCLUDE_RTIO_RTIO_H_
 
#include <zephyr/rtio/rtio_spsc.h>
#include <zephyr/sys/__assert.h>
#include <zephyr/sys/atomic.h>
#include <zephyr/device.h>
#include <zephyr/kernel.h>
 
#ifdef __cplusplus
extern "C" {
#endif
 
 
struct rtio_iodev;
 
#define RTIO_PRIO_LOW 0U
 
#define RTIO_PRIO_NORM 127U
 
#define RTIO_PRIO_HIGH 255U
 
#define RTIO_SQE_CHAINED BIT(0)
 
struct rtio_sqe {
        uint8_t op; 
        uint8_t prio; 
        uint16_t flags; 
        const struct rtio_iodev *iodev; 
        void *userdata;
 
        union {
                struct {
                        uint32_t buf_len; 
                        uint8_t *buf; 
                };
        };
};
 
struct rtio_sq {
        struct rtio_spsc _spsc;
        struct rtio_sqe buffer[];
};
 
struct rtio_cqe {
        int32_t result; 
        void *userdata; 
};
 
struct rtio_cq {
        struct rtio_spsc _spsc;
        struct rtio_cqe buffer[];
};
 
struct rtio;
 
struct rtio_executor_api {
        int (*submit)(struct rtio *r);
 
        void (*ok)(struct rtio *r, const struct rtio_sqe *sqe, int result);
 
        void (*err)(struct rtio *r, const struct rtio_sqe *sqe, int result);
};
 
struct rtio_executor {
        const struct rtio_executor_api *api;
};
 
struct rtio {
 
        /*
         * An executor which does the job of working through the submission
         * queue.
         */
        struct rtio_executor *executor;
 
 
#ifdef CONFIG_RTIO_SUBMIT_SEM
        /* A wait semaphore which may suspend the calling thread
         * to wait for some number of completions when calling submit
         */
        struct k_sem *submit_sem;
 
        uint32_t submit_count;
#endif
 
#ifdef CONFIG_RTIO_CONSUME_SEM
        /* A wait semaphore which may suspend the calling thread
         * to wait for some number of completions while consuming
         * them from the completion queue
         */
        struct k_sem *consume_sem;
#endif
 
        /* Number of completions that were unable to be submitted with results
         * due to the cq spsc being full
         */
        atomic_t xcqcnt;
 
        /* Submission queue */
        struct rtio_sq *sq;
 
        /* Completion queue */
        struct rtio_cq *cq;
};
 
struct rtio_iodev_api {
        void (*submit)(const struct rtio_sqe *sqe,
                       struct rtio *r);
 
};
 
/* IO device submission queue entry */
struct rtio_iodev_sqe {
        const struct rtio_sqe *sqe;
        struct rtio *r;
};
 
struct rtio_iodev_sq {
        struct rtio_spsc _spsc;
        struct rtio_iodev_sqe buffer[];
};
 
struct rtio_iodev {
        /* Function pointer table */
        const struct rtio_iodev_api *api;
 
        /* Queue of RTIO contexts with requests */
        struct rtio_iodev_sq *iodev_sq;
 
        /* Data associated with this iodev */
        void *data;
};
 
#define RTIO_OP_NOP 0
 
#define RTIO_OP_RX 1
 
#define RTIO_OP_TX 2
 
static inline void rtio_sqe_prep_nop(struct rtio_sqe *sqe,
                                const struct rtio_iodev *iodev,
                                void *userdata)
{
        sqe->op = RTIO_OP_NOP;
        sqe->iodev = iodev;
        sqe->userdata = userdata;
}
 
static inline void rtio_sqe_prep_read(struct rtio_sqe *sqe,
                                      const struct rtio_iodev *iodev,
                                      int8_t prio,
                                      uint8_t *buf,
                                      uint32_t len,
                                      void *userdata)
{
        sqe->op = RTIO_OP_RX;
        sqe->prio = prio;
        sqe->iodev = iodev;
        sqe->buf_len = len;
        sqe->buf = buf;
        sqe->userdata = userdata;
}
 
static inline void rtio_sqe_prep_write(struct rtio_sqe *sqe,
                                       const struct rtio_iodev *iodev,
                                       int8_t prio,
                                       uint8_t *buf,
                                       uint32_t len,
                                       void *userdata)
{
        sqe->op = RTIO_OP_TX;
        sqe->prio = prio;
        sqe->iodev = iodev;
        sqe->buf_len = len;
        sqe->buf = buf;
        sqe->userdata = userdata;
}
 
#define RTIO_SQ_DEFINE(name, len)                       \
        RTIO_SPSC_DEFINE(name, struct rtio_sqe, len)
 
#define RTIO_CQ_DEFINE(name, len)                       \
        RTIO_SPSC_DEFINE(name, struct rtio_cqe, len)
 
 
#define RTIO_IODEV_SQ_DEFINE(name, len) \
        RTIO_SPSC_DEFINE(name, struct rtio_iodev_sqe, len)
 
#define RTIO_IODEV_DEFINE(name, iodev_api, qsize, iodev_data)                                      \
        static RTIO_IODEV_SQ_DEFINE(_iodev_sq_##name, qsize);                                      \
        const STRUCT_SECTION_ITERABLE(rtio_iodev, name) = {                                        \
                .api = (iodev_api),                                                                \
                .iodev_sq = (struct rtio_iodev_sq *const)&_iodev_sq_##name,                        \
                .data = (iodev_data),                                                              \
        }
 
#define RTIO_DEFINE(name, exec, sq_sz, cq_sz)   \
        IF_ENABLED(CONFIG_RTIO_SUBMIT_SEM,                                                         \
                   (static K_SEM_DEFINE(_submit_sem_##name, 0, K_SEM_MAX_LIMIT)))                  \
        IF_ENABLED(CONFIG_RTIO_CONSUME_SEM,                                                        \
                   (static K_SEM_DEFINE(_consume_sem_##name, 0, 1)))                               \
        static RTIO_SQ_DEFINE(_sq_##name, sq_sz);                                                  \
        static RTIO_CQ_DEFINE(_cq_##name, cq_sz);                                                  \
        STRUCT_SECTION_ITERABLE(rtio, name) = {                                                    \
                .executor = (exec),                                                                \
                .xcqcnt = ATOMIC_INIT(0),                                                          \
                IF_ENABLED(CONFIG_RTIO_SUBMIT_SEM, (.submit_sem = &_submit_sem_##name,))           \
                IF_ENABLED(CONFIG_RTIO_SUBMIT_SEM, (.submit_count = 0,))                           \
                IF_ENABLED(CONFIG_RTIO_CONSUME_SEM, (.consume_sem = &_consume_sem_##name,))        \
                .sq = (struct rtio_sq *const)&_sq_##name,                                          \
                .cq = (struct rtio_cq *const)&_cq_##name,                                          \
        };
 
static inline void rtio_set_executor(struct rtio *r, struct rtio_executor *exc)
{
        r->executor = exc;
}
 
static inline void rtio_iodev_submit(const struct rtio_sqe *sqe, struct rtio *r)
{
        sqe->iodev->api->submit(sqe, r);
}
 
static inline uint32_t rtio_sqe_acquirable(struct rtio *r)
{
        return rtio_spsc_acquirable(r->sq);
}
 
static inline struct rtio_sqe *rtio_sqe_acquire(struct rtio *r)
{
        return rtio_spsc_acquire(r->sq);
}
 
static inline void rtio_sqe_produce_all(struct rtio *r)
{
        rtio_spsc_produce_all(r->sq);
}
 
 
static inline void rtio_sqe_drop_all(struct rtio *r)
{
        rtio_spsc_drop_all(r->sq);
}
 
 
static inline struct rtio_cqe *rtio_cqe_consume(struct rtio *r)
{
        return rtio_spsc_consume(r->cq);
}
 
static inline struct rtio_cqe *rtio_cqe_consume_block(struct rtio *r)
{
        struct rtio_cqe *cqe;
 
        /* TODO is there a better way? reset this in submit? */
#ifdef CONFIG_RTIO_CONSUME_SEM
        k_sem_reset(r->consume_sem);
#endif
        cqe = rtio_spsc_consume(r->cq);
 
        while (cqe == NULL) {
                cqe = rtio_spsc_consume(r->cq);
 
#ifdef CONFIG_RTIO_CONSUME_SEM
                k_sem_take(r->consume_sem, K_FOREVER);
#else
                k_yield();
#endif
        }
 
        return cqe;
}
 
static inline void rtio_cqe_release_all(struct rtio *r)
{
        rtio_spsc_release_all(r->cq);
}
 
 
static inline void rtio_sqe_ok(struct rtio *r, const struct rtio_sqe *sqe, int result)
{
        r->executor->api->ok(r, sqe, result);
}
 
static inline void rtio_sqe_err(struct rtio *r, const struct rtio_sqe *sqe, int result)
{
        r->executor->api->err(r, sqe, result);
}
 
static inline void rtio_cqe_submit(struct rtio *r, int result, void *userdata)
{
        struct rtio_cqe *cqe = rtio_spsc_acquire(r->cq);
 
        if (cqe == NULL) {
                atomic_inc(&r->xcqcnt);
        } else {
                cqe->result = result;
                cqe->userdata = userdata;
                rtio_spsc_produce(r->cq);
        }
#ifdef CONFIG_RTIO_SUBMIT_SEM
        if (r->submit_count > 0) {
                r->submit_count--;
                if (r->submit_count == 0) {
                        k_sem_give(r->submit_sem);
                }
        }
#endif
#ifdef CONFIG_RTIO_CONSUME_SEM
        k_sem_give(r->consume_sem);
#endif
}
 
static inline void rtio_access_grant(struct rtio *r, struct k_thread *t)
{
        k_object_access_grant(r, t);
 
#ifdef CONFIG_RTIO_SUBMIT_SEM
        k_object_access_grant(r->submit_sem, t);
#endif
 
#ifdef CONFIG_RTIO_CONSUME_SEM
        k_object_access_grant(r->consume_sem, t);
#endif
}
 
__syscall int rtio_sqe_copy_in(struct rtio *r,
                               const struct rtio_sqe *sqes,
                               size_t sqe_count);
static inline int z_impl_rtio_sqe_copy_in(struct rtio *r,
                                          const struct rtio_sqe *sqes,
                                          size_t sqe_count)
{
        struct rtio_sqe *sqe;
        uint32_t acquirable = rtio_sqe_acquirable(r);
 
        if (acquirable < sqe_count) {
                return -ENOMEM;
        }
 
        for (int i = 0; i < sqe_count; i++) {
                sqe = rtio_sqe_acquire(r);
                __ASSERT_NO_MSG(sqe != NULL);
                *sqe = sqes[i];
        }
 
        rtio_sqe_produce_all(r);
 
        return 0;
}
 
__syscall int rtio_cqe_copy_out(struct rtio *r,
                                struct rtio_cqe *cqes,
                                size_t cqe_count,
                                k_timeout_t timeout);
static inline int z_impl_rtio_cqe_copy_out(struct rtio *r,
                                           struct rtio_cqe *cqes,
                                           size_t cqe_count,
                                           k_timeout_t timeout)
{
        size_t copied;
        struct rtio_cqe *cqe;
 
        for (copied = 0; copied < cqe_count; copied++) {
                cqe = rtio_cqe_consume_block(r);
                if (cqe == NULL) {
                        break;
                }
                cqes[copied] = *cqe;
        }
 
 
        rtio_cqe_release_all(r);
 
        return copied;
}
 
__syscall int rtio_submit(struct rtio *r, uint32_t wait_count);
 
static inline int z_impl_rtio_submit(struct rtio *r, uint32_t wait_count)
{
        int res;
 
        __ASSERT(r->executor != NULL, "expected rtio submit context to have an executor");
 
#ifdef CONFIG_RTIO_SUBMIT_SEM
        /* TODO undefined behavior if another thread calls submit of course
         */
        if (wait_count > 0) {
                __ASSERT(!k_is_in_isr(),
                         "expected rtio submit with wait count to be called from a thread");
 
                k_sem_reset(r->submit_sem);
                r->submit_count = wait_count;
        }
#endif
 
        /* Enqueue all prepared submissions */
        rtio_spsc_produce_all(r->sq);
 
        /* Submit the queue to the executor which consumes submissions
         * and produces completions through ISR chains or other means.
         */
        res = r->executor->api->submit(r);
        if (res != 0) {
                return res;
        }
 
        /* TODO could be nicer if we could suspend the thread and not
         * wake up on each completion here.
         */
#ifdef CONFIG_RTIO_SUBMIT_SEM
 
        if (wait_count > 0) {
                res = k_sem_take(r->submit_sem, K_FOREVER);
                __ASSERT(res == 0,
                         "semaphore was reset or timed out while waiting on completions!");
        }
#else
        while (rtio_spsc_consumable(r->cq) < wait_count) {
#ifdef CONFIG_BOARD_NATIVE_POSIX
                k_busy_wait(1);
#else
                k_yield();
#endif /* CONFIG_BOARD_NATIVE_POSIX */
        }
#endif
 
        return res;
}
 
#ifdef __cplusplus
}
#endif
 
#include <syscalls/rtio.h>
 
#endif /* ZEPHYR_INCLUDE_RTIO_RTIO_H_ */