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,47 +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
+ * @brief API that an RTIO IO device should implement
  */
-struct rtio_iodev_sq {
-	struct rtio_spsc _spsc;
-	struct rtio_iodev_sqe buffer[];
+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);
 };
 
 /**
@@ -308,7 +295,7 @@ struct rtio_iodev {
 	const struct rtio_iodev_api *api;
 
 	/* Queue of RTIO contexts with requests */
-	struct rtio_iodev_sq *iodev_sq;
+	struct rtio_mpsc iodev_sq;
 
 	/* Data associated with this iodev */
 	void *data;
@@ -389,29 +376,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_sq = RTIO_MPSC_INIT((name.iodev_sq)),                                       \
 		.data = (iodev_data),                                                              \
 	}
 
@@ -449,14 +424,16 @@ 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);
+	iodev_sqe->sqe->iodev->api->submit(iodev_sqe);
 }
 
 /**
@@ -571,27 +548,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
  *

samples/subsys/rtio/sensor_batch_processing/src/vnd_sensor.c

@@ -22,7 +22,6 @@ struct vnd_sensor_data {
 	struct rtio_iodev iodev;
 	struct k_timer timer;
 	const struct device *dev;
-	struct k_msgq msgq;
 	uint32_t sample_number;
 };
 
@@ -53,9 +52,10 @@ static int vnd_sensor_iodev_read(const struct device *dev, uint8_t *buf,
 }
 
 static void vnd_sensor_iodev_execute(const struct device *dev,
-		const struct rtio_sqe *sqe, struct rtio *r)
+		struct rtio_iodev_sqe *iodev_sqe)
 {
 	int result;
+	const struct rtio_sqe *sqe = iodev_sqe->sqe;
 
 	if (sqe->op == RTIO_OP_RX) {
 		result = vnd_sensor_iodev_read(dev, sqe->buf, sqe->buf_len);
@@ -65,36 +65,28 @@ static void vnd_sensor_iodev_execute(const struct device *dev,
 	}
 
 	if (result < 0) {
-		rtio_sqe_err(r, sqe, result);
+		rtio_iodev_sqe_err(iodev_sqe, result);
 	} else {
-		rtio_sqe_ok(r, sqe, result);
+		rtio_iodev_sqe_ok(iodev_sqe, result);
 	}
 }
 
-static void vnd_sensor_iodev_submit(const struct rtio_sqe *sqe, struct rtio *r)
+static void vnd_sensor_iodev_submit(struct rtio_iodev_sqe *iodev_sqe)
 {
-	struct vnd_sensor_data *data = (struct vnd_sensor_data *) sqe->iodev;
-	const struct device *dev = data->dev;
-	struct rtio_iodev_sqe *iodev_sqe = rtio_spsc_acquire(data->iodev.iodev_sq);
-
-	if (iodev_sqe != NULL) {
-		iodev_sqe->sqe = sqe;
-		iodev_sqe->r = r;
-		rtio_spsc_produce(data->iodev.iodev_sq);
-	} else {
-		LOG_ERR("%s: Could not put a msg", dev->name);
-		rtio_sqe_err(r, sqe, -EWOULDBLOCK);
-	}
+	struct vnd_sensor_data *data = (struct vnd_sensor_data *) iodev_sqe->sqe->iodev;
+
+	rtio_mpsc_push(&data->iodev.iodev_sq, &iodev_sqe->q);
 }
 
 static void vnd_sensor_handle_int(const struct device *dev)
 {
 	struct vnd_sensor_data *data = dev->data;
-	struct rtio_iodev_sqe *iodev_sqe = rtio_spsc_consume(data->iodev.iodev_sq);
+	struct rtio_mpsc_node *node = rtio_mpsc_pop(&data->iodev.iodev_sq);
 
-	if (iodev_sqe != NULL) {
-		vnd_sensor_iodev_execute(dev, iodev_sqe->sqe, iodev_sqe->r);
-		rtio_spsc_release(data->iodev.iodev_sq);
+	if (node != NULL) {
+		struct rtio_iodev_sqe *iodev_sqe = CONTAINER_OF(node, struct rtio_iodev_sqe, q);
+
+		vnd_sensor_iodev_execute(dev, iodev_sqe);
 	} else {
 		LOG_ERR("%s: Could not get a msg", dev->name);
 	}
@@ -116,6 +108,8 @@ static int vnd_sensor_init(const struct device *dev)
 
 	data->dev = dev;
 
+	rtio_mpsc_init(&data->iodev.iodev_sq);
+
 	k_timer_init(&data->timer, vnd_sensor_timer_expiry, NULL);
 
 	k_timer_start(&data->timer, K_MSEC(sample_period),
@@ -135,13 +129,10 @@ static const struct rtio_iodev_api vnd_sensor_iodev_api = {
 		.sample_size = DT_INST_PROP(n, sample_size),                                       \
 	};                                                                                         \
                                                                                                    \
-	RTIO_IODEV_SQ_DEFINE(vnd_sensor_iodev_sq_##n, DT_INST_PROP(n, max_msgs));                  \
-                                                                                                   \
 	static struct vnd_sensor_data vnd_sensor_data_##n = {                                      \
 		.iodev =                                                                           \
 			{                                                                          \
 				.api = &vnd_sensor_iodev_api,                                      \
-				.iodev_sq = (struct rtio_iodev_sq *)&vnd_sensor_iodev_sq_##n,     \
 			},                                                                         \
 	};                                                                                         \
                                                                                                    \