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; 
        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;
};
 
 
#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,
                                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,
                                      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,
                                       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)                       \
        static RTIO_SPSC_DEFINE(name, struct rtio_sqe, len)
 
#define RTIO_CQ_DEFINE(name, len)                       \
        static RTIO_SPSC_DEFINE(name, struct rtio_cqe, len)
 
#define RTIO_DEFINE(name, exec, sq_sz, cq_sz)                                                      \
        IF_ENABLED(CONFIG_RTIO_SUBMIT_SEM, (K_SEM_DEFINE(_submit_sem_##name, 0, K_SEM_MAX_LIMIT))) \
        IF_ENABLED(CONFIG_RTIO_CONSUME_SEM, (K_SEM_DEFINE(_consume_sem_##name, 0, 1)))             \
        RTIO_SQ_DEFINE(_sq_##name, sq_sz);                                                         \
        RTIO_CQ_DEFINE(_cq_##name, cq_sz);                                                         \
        static struct 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 int 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;
}
 
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_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
}
 
/* TODO add rtio_sqe_suspend() for suspending a submission chain that must
 * wait on other in progress submissions or submission chains.
 */
 
#ifdef __cplusplus
}
#endif
 
#endif /* ZEPHYR_INCLUDE_RTIO_RTIO_H_ */