rtio: Shareable lock-free iodevs

By using an mpsc queue for each iodev, the iodev itself is shareable across
contexts. Since its lock free, submits may occur even from an ISR context.

Rather than a fixed size queue, and with it the possibility of running
out of pre-allocated spots, each iodev now holds a wait-free mpsc
queue head.

This changes the parameter of iodev submit to be a struct containing 4
pointers for the rtio context, the submission queue entry, and the mpsc
node for the iodevs submission queue.

This solves the problem involving busy iodevs working with real
devices. For example a busy SPI bus driver could enqueue, without locking,
a request to start once the current request is done.

The queue entries are expected to be owned and allocated by the
executor rather than the iodev. This helps simplify potential
tuning knobs to one place, the RTIO context and its executor an
application directly uses.

As the test case shows iodevs can operate effectively lock free
with the mpsc queue and a single atomic denoting the current task.

Signed-off-by: Tom Burdick <[email protected]>

Author: teburd

include/zephyr/rtio/rtio.h

@@ -33,6 +33,7 @@
 #define ZEPHYR_INCLUDE_RTIO_RTIO_H_
 
 #include <zephyr/rtio/rtio_spsc.h>
+#include <zephyr/rtio/rtio_mpsc.h>
 #include <zephyr/sys/__assert.h>
 #include <zephyr/sys/atomic.h>
 #include <zephyr/device.h>
@@ -164,6 +165,7 @@ struct rtio_cq {
 };
 
 struct rtio;
+struct rtio_iodev_sqe;
 
 struct rtio_executor_api {
 	/**
@@ -179,12 +181,12 @@ struct rtio_executor_api {
 	/**
 	 * @brief SQE completes successfully
 	 */
-	void (*ok)(struct rtio *r, const struct rtio_sqe *sqe, int result);
+	void (*ok)(struct rtio_iodev_sqe *iodev_sqe, int result);
 
 	/**
-	 * @brief SQE fails to complete
+	 * @brief SQE fails to complete successfully
 	 */
-	void (*err)(struct rtio *r, const struct rtio_sqe *sqe, int result);
+	void (*err)(struct rtio_iodev_sqe *iodev_sqe, int result);
 };
 
 /**
@@ -257,61 +259,32 @@ struct rtio {
 	struct rtio_cq *cq;
 };
 
+
 /**
- * @brief API that an RTIO IO device should implement
+ * @brief IO device submission queue entry
  */
-struct rtio_iodev_api {
-	/**
-	 * @brief Submission function for a request to the iodev
-	 *
-	 * The iodev is responsible for doing the operation described
-	 * as a submission queue entry and reporting results using using
-	 * `rtio_sqe_ok` or `rtio_sqe_err` once done.
-	 */
-	void (*submit)(const struct rtio_sqe *sqe,
-		       struct rtio *r);
-
-	/**
-	 * TODO some form of transactional piece is missing here
-	 * where we wish to "transact" on an iodev with multiple requests
-	 * over a chain.
-	 *
-	 * Only once explicitly released or the chain fails do we want
-	 * to release. Once released any pending iodevs in the queue
-	 * should be done.
-	 *
-	 * Something akin to a lock/unlock pair.
-	 */
-};
-
-/* IO device submission queue entry */
 struct rtio_iodev_sqe {
+	struct rtio_mpsc_node q;
 	const struct rtio_sqe *sqe;
 	struct rtio *r;
 };
 
 /**
- * @brief IO device submission queue
- *
- * This is used for reifying the member of the rtio_iodev struct
- */
-struct rtio_iodev_sq {
-	struct rtio_spsc _spsc;
-	struct rtio_iodev_sqe buffer[];
-};
-
-/**
- * @brief An IO device with a function table for submitting requests
+ * @brief API that an RTIO IO device should implement
  */
-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;
+struct rtio_iodev_api {
+	/**
+	 * @brief Submit to the iodev an entry to work on
+	 *
+	 * This call should be short in duration and most likely
+	 * either enqueue or kick off an entry with the hardware.
+	 *
+	 * If polling is required the iodev should add itself to the execution
+	 * context (@see rtio_add_pollable())
+	 *
+	 * @param iodev_sqe Submission queue entry
+	 */
+	void (*submit)(struct rtio_iodev_sqe *iodev_sqe);
 };
 
 /** An operation that does nothing and will complete immediately */
@@ -389,29 +362,17 @@ static inline void rtio_sqe_prep_write(struct rtio_sqe *sqe,
 #define RTIO_CQ_DEFINE(name, len)			\
 	RTIO_SPSC_DEFINE(name, struct rtio_cqe, len)
 
-
-/**
- * @brief Statically define and initialize a fixed length iodev submission queue
- *
- * @param name Name of the queue.
- * @param len Queue length, power of 2 required
- */
-#define RTIO_IODEV_SQ_DEFINE(name, len) \
-	RTIO_SPSC_DEFINE(name, struct rtio_iodev_sqe, len)
-
 /**
  * @brief Statically define and initialize an RTIO IODev
  *
  * @param name Name of the iodev
  * @param iodev_api Pointer to struct rtio_iodev_api
- * @param qsize Size of the submission queue, must be power of 2
  * @param iodev_data Data pointer
  */
-#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) = {                                        \
+#define RTIO_IODEV_DEFINE(name, iodev_api, iodev_data)                                             \
+	STRUCT_SECTION_ITERABLE(rtio_iodev, name) = {                                              \
 		.api = (iodev_api),                                                                \
-		.iodev_sq = (struct rtio_iodev_sq *const)&_iodev_sq_##name,                        \
+		.iodev_q = RTIO_MPSC_INIT((name.iodev_q)),                                         \
 		.data = (iodev_data),                                                              \
 	}
 
@@ -449,14 +410,32 @@ static inline void rtio_set_executor(struct rtio *r, struct rtio_executor *exc)
 }
 
 /**
- * @brief Perform a submitted operation with an iodev
+ * @brief Submit to an iodev a submission to work on
  *
- * @param sqe Submission to work on
- * @param r RTIO context
+ * Should be called by the executor when it wishes to submit work
+ * to an iodev.
+ *
+ * @param iodev_sqe Submission to work on
  */
-static inline void rtio_iodev_submit(const struct rtio_sqe *sqe, struct rtio *r)
+static inline void rtio_iodev_submit(struct rtio_iodev_sqe *iodev_sqe)
 {
-	sqe->iodev->api->submit(sqe, r);
+	return iodev_sqe->sqe->iodev->api->submit(iodev_sqe);
+}
+
+/**
+ * @brief Poll an iodev for completion of a submission
+ *
+ * Should be called by the executor when rtio_submit or rtio_cqe is called
+ * and polling the device is required.
+ *
+ * @param iodev IODev to poll
+ *
+ * @retval RTIO_POLL_COMPLETED Polling is no longer required
+ * @retval RTIO_POLL_PENDING Polling is still required (task is pending)
+ */
+static inline enum rtio_poll_status rtio_iodev_poll(struct rtio_iodev *iodev)
+{
+	return iodev->api->poll(iodev);
 }
 
 /**
@@ -571,27 +550,25 @@ static inline void rtio_cqe_release_all(struct rtio *r)
  *
  * This may start the next asynchronous request if one is available.
  *
- * @param r RTIO context
- * @param sqe Submission that has succeeded
+ * @param iodev_sqe IODev Submission that has succeeded
  * @param result Result of the request
  */
-static inline void rtio_sqe_ok(struct rtio *r, const struct rtio_sqe *sqe, int result)
+static inline void rtio_iodev_sqe_ok(struct rtio_iodev_sqe *iodev_sqe, int result)
 {
-	r->executor->api->ok(r, sqe, result);
+	iodev_sqe->r->executor->api->ok(iodev_sqe, result);
 }
 
 /**
  * @brief Inform the executor of a submissions completion with error
  *
  * This SHALL fail the remaining submissions in the chain.
  *
- * @param r RTIO context
- * @param sqe Submission that has failed
+ * @param iodev_sqe Submission that has failed
  * @param result Result of the request
  */
-static inline void rtio_sqe_err(struct rtio *r, const struct rtio_sqe *sqe, int result)
+static inline void rtio_iodev_sqe_err(struct rtio_iodev_sqe *iodev_sqe, int result)
 {
-	r->executor->api->err(r, sqe, result);
+	iodev_sqe->r->executor->api->err(iodev_sqe, result);
 }
 
 /**

include/zephyr/rtio/rtio_executor_concurrent.h

@@ -37,20 +37,18 @@ int rtio_concurrent_submit(struct rtio *r);
 /**
  * @brief Report a SQE has completed successfully
  *
- * @param r RTIO context to use
- * @param sqe RTIO SQE to report success
+ * @param sqe RTIO IODev SQE to report success
  * @param result Result of the SQE
  */
-void rtio_concurrent_ok(struct rtio *r, const struct rtio_sqe *sqe, int result);
+void rtio_concurrent_ok(struct rtio_iodev_sqe *sqe, int result);
 
 /**
  * @brief Report a SQE has completed with error
  *
- * @param r RTIO context to use
- * @param sqe RTIO SQE to report success
+ * @param sqe RTIO IODev SQE to report success
  * @param result Result of the SQE
  */
-void rtio_concurrent_err(struct rtio *r, const struct rtio_sqe *sqe, int result);
+void rtio_concurrent_err(struct rtio_iodev_sqe *sqe, int result);
 
 /**
  * @brief Concurrent Executor
@@ -76,8 +74,8 @@ struct rtio_concurrent_executor {
 	/* Array of task statuses */
 	uint8_t *task_status;
 
-	/* Array of struct rtio_sqe *'s one per task' */
-	struct rtio_sqe **task_cur;
+	/* Array of struct rtio_iodev_sqe *'s one per task' */
+	struct rtio_iodev_sqe *task_cur;
 };
 
 /**
@@ -101,7 +99,7 @@ static const struct rtio_executor_api z_rtio_concurrent_api = {
  * @param concurrency Allowed concurrency (number of concurrent tasks).
  */
 #define RTIO_EXECUTOR_CONCURRENT_DEFINE(name, concurrency)                                         \
-	static struct rtio_sqe *_task_cur_##name[(concurrency)];                                   \
+	static struct rtio_iodev_sqe _task_cur_##name[(concurrency)];                              \
 	uint8_t _task_status_##name[(concurrency)];                                                \
 	static struct rtio_concurrent_executor name = {                                            \
 		.ctx = { .api = &z_rtio_concurrent_api },                                          \

include/zephyr/rtio/rtio_executor_simple.h

@@ -36,26 +36,25 @@ int rtio_simple_submit(struct rtio *r);
 /**
  * @brief Report a SQE has completed successfully
  *
- * @param r RTIO context to use
- * @param sqe RTIO SQE to report success
+ * @param iodev_sqe RTIO IODEV SQE to report success
  * @param result Result of the SQE
  */
-void rtio_simple_ok(struct rtio *r, const struct rtio_sqe *sqe, int result);
+void rtio_simple_ok(struct rtio_iodev_sqe *iodev_sqe, int result);
 
 /**
  * @brief Report a SQE has completed with error
  *
- * @param r RTIO context to use
- * @param sqe RTIO SQE to report success
+ * @param iodev_sqe RTIO IODEV SQE to report success
  * @param result Result of the SQE
  */
-void rtio_simple_err(struct rtio *r, const struct rtio_sqe *sqe, int result);
+void rtio_simple_err(struct rtio_iodev_sqe *iodev_sqe, int result);
 
 /**
  * @brief Simple Executor
  */
 struct rtio_simple_executor {
 	struct rtio_executor ctx;
+	struct rtio_iodev_sqe task;
 };
 
 /**

include/zephyr/rtio/rtio_mpsc.h

@@ -12,6 +12,7 @@
 #include <stdint.h>
 #include <stdbool.h>
 #include <zephyr/sys/atomic.h>
+#include <zephyr/kernel.h>
 
 /**
  * @brief RTIO Multiple Producer Single Consumer (MPSC) Queue API
@@ -52,6 +53,21 @@ struct rtio_mpsc {
 	struct rtio_mpsc_node stub;
 };
 
+
+/**
+ * @brief Static initializer for a mpsc queue
+ *
+ * Since the queue is
+ *
+ * @param symbol name of the queue
+ */
+#define RTIO_MPSC_INIT(symbol)                                 \
+	{                                                      \
+		.head = (struct rtio_mpsc_node *)&symbol.stub, \
+		.tail = (struct rtio_mpsc_node *)&symbol.stub, \
+		.stub.next = NULL,                             \
+	}
+
 /**
  * @brief Initialize queue
  *