rtio: Add transactional submissions

Transactional submissions treat a sequence of sqes as a single atomic
submission given to a single iodev and expect as a reply a single
completion.

This is useful for scatter gather like APIs that exist in Zephyr already
for I2C and SPI.

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

Author: teburd

include/zephyr/rtio/rtio.h

@@ -97,9 +97,25 @@ struct rtio_iodev;
 
 /**
  * @brief The next request in the queue should wait on this one.
+ *
+ * Chained SQEs are individual units of work describing patterns of
+ * ordering and failure cascading. A chained SQE must be started only
+ * after the one before it. They are given to the iodevs one after another.
  */
 #define RTIO_SQE_CHAINED BIT(0)
 
+/**
+ * @brief The next request in the queue is part of a transaction.
+ *
+ * Transactional SQEs are sequential parts of a unit of work.
+ * Only the first transactional SQE is submitted to an iodev, the
+ * remaining SQEs are never individually submitted but instead considered
+ * to be part of the transaction to the single iodev. The first sqe in the
+ * sequence holds the iodev that will be used and the last holds the userdata
+ * that will be returned in a single completion on failure/success.
+ */
+#define RTIO_SQE_TRANSACTION BIT(1)
+
 /**
  * @}
  */

subsys/rtio/rtio_executor_concurrent.c

@@ -49,7 +49,7 @@ static uint16_t conex_task_next(struct rtio_concurrent_executor *exc)
 	uint16_t task_id = exc->task_in;
 
 	exc->task_in++;
-	return task_id;
+	return task_id & exc->task_mask;
 }
 
 static inline uint16_t conex_task_id(struct rtio_concurrent_executor *exc,
@@ -62,16 +62,25 @@ static inline uint16_t conex_task_id(struct rtio_concurrent_executor *exc,
 
 static void conex_sweep_task(struct rtio *r, struct rtio_concurrent_executor *exc)
 {
+	uint32_t swept = 1;
 	struct rtio_sqe *sqe = rtio_spsc_consume(r->sq);
 
-	while (sqe != NULL && sqe->flags & RTIO_SQE_CHAINED) {
+	while (sqe != NULL && (sqe->flags & (RTIO_SQE_CHAINED | RTIO_SQE_TRANSACTION))) {
 		rtio_spsc_release(r->sq);
 		sqe = rtio_spsc_consume(r->sq);
+		swept++;
 	}
 
 	rtio_spsc_release(r->sq);
 }
 
+/**
+ * @brief Sweep like a GC of sorts old tasks that are completed in order
+ *
+ * Will only sweep tasks in the order they arrived in the submission queue.
+ * Meaning there might be completed tasks that could be freed but are not yet
+ * because something before it has not yet completed.
+ */
 static void conex_sweep(struct rtio *r, struct rtio_concurrent_executor *exc)
 {
 	/* In order sweep up */
@@ -86,21 +95,77 @@ static void conex_sweep(struct rtio *r, struct rtio_concurrent_executor *exc)
 	}
 }
 
+/**
+ * @brief Prepare tasks to run by iterating through the submission queue
+ *
+ * For each submission in the queue that begins a chain or transaction
+ * start a task if possible. Concurrency is limited by the allocated concurrency
+ * per executor instance.
+ */
+static void conex_prepare(struct rtio *r, struct rtio_concurrent_executor *exc)
+{
+	struct rtio_sqe *sqe;
+
+	/* If never submitted before peek at the first item
+	 * otherwise start back up where the last submit call
+	 * left off
+	 */
+	if (exc->pending_sqe == NULL) {
+		sqe = rtio_spsc_peek(r->sq);
+	} else {
+		sqe = exc->pending_sqe;
+	}
+
+	while (sqe != NULL && conex_task_free(exc)) {
+		/* Get the next free task id */
+		uint16_t task_idx = conex_task_next(exc);
+
+		/* Setup task */
+		exc->task_cur[task_idx].sqe = sqe;
+		exc->task_cur[task_idx].r = r;
+		exc->task_status[task_idx] = CONEX_TASK_SUSPENDED;
+
+		/* Go to the next sqe not in the current chain or transaction */
+		while (sqe != NULL && (sqe->flags & (RTIO_SQE_CHAINED | RTIO_SQE_TRANSACTION))) {
+			sqe = rtio_spsc_next(r->sq, sqe);
+		}
+
+		/* SQE is the end of the previous chain or transaction so skip it */
+		sqe = rtio_spsc_next(r->sq, sqe);
+	}
+
+	/* Out of available tasks so remember where we left off to begin again once tasks free up */
+	exc->pending_sqe = sqe;
+}
+
+
+/**
+ * @brief Resume tasks that are suspended
+ *
+ * All tasks begin as suspended tasks. This kicks them off to the submissions
+ * associated iodev.
+ */
 static void conex_resume(struct rtio *r, struct rtio_concurrent_executor *exc)
 {
 	/* In order resume tasks */
 	for (uint16_t task_id = exc->task_out; task_id < exc->task_in; task_id++) {
 		if (exc->task_status[task_id & exc->task_mask] & CONEX_TASK_SUSPENDED) {
 			LOG_INF("resuming suspended task %d", task_id);
-			exc->task_status[task_id] &= ~CONEX_TASK_SUSPENDED;
-			rtio_iodev_submit(&exc->task_cur[task_id]);
+			exc->task_status[task_id & exc->task_mask] &= ~CONEX_TASK_SUSPENDED;
+			rtio_iodev_submit(&exc->task_cur[task_id & exc->task_mask]);
 		}
 	}
 }
 
+/**
+ * @brief Sweep, Prepare, and Resume in one go
+ *
+ * Called after a completion to continue doing more work if needed.
+ */
 static void conex_sweep_resume(struct rtio *r, struct rtio_concurrent_executor *exc)
 {
 	conex_sweep(r, exc);
+	conex_prepare(r, exc);
 	conex_resume(r, exc);
 }
 
@@ -114,71 +179,14 @@ static void conex_sweep_resume(struct rtio *r, struct rtio_concurrent_executor *
 int rtio_concurrent_submit(struct rtio *r)
 {
 
-	LOG_INF("submit");
-
 	struct rtio_concurrent_executor *exc =
 		(struct rtio_concurrent_executor *)r->executor;
-	struct rtio_sqe *sqe;
-	struct rtio_sqe *last_sqe;
 	k_spinlock_key_t key;
 
 	key = k_spin_lock(&exc->lock);
 
-	/* If never submitted before peek at the first item
-	 * otherwise start back up where the last submit call
-	 * left off
-	 */
-	if (exc->last_sqe == NULL) {
-		sqe = rtio_spsc_peek(r->sq);
-	} else {
-		/* Pickup from last submit call */
-		sqe = rtio_spsc_next(r->sq, exc->last_sqe);
-	}
-
-	last_sqe = sqe;
-	while (sqe != NULL && conex_task_free(exc)) {
-		LOG_INF("head SQE in chain %p", sqe);
-
-		/* Get the next task id if one exists */
-		uint16_t task_idx = conex_task_next(exc);
-
-		LOG_INF("setting up task %d", task_idx);
-
-		/* Setup task (yes this is it) */
-		exc->task_cur[task_idx].sqe = sqe;
-		exc->task_cur[task_idx].r = r;
-		exc->task_status[task_idx] = CONEX_TASK_SUSPENDED;
-
-		LOG_INF("submitted sqe %p", sqe);
-		/* Go to the next sqe not in the current chain */
-		while (sqe != NULL && (sqe->flags & RTIO_SQE_CHAINED)) {
-			sqe = rtio_spsc_next(r->sq, sqe);
-		}
-
-		LOG_INF("tail SQE in chain %p", sqe);
-
-		last_sqe = sqe;
-
-		/* SQE is the end of the previous chain */
-		sqe = rtio_spsc_next(r->sq, sqe);
-	}
-
-	/* Out of available pointers, wait til others complete, note the
-	 * first pending submission queue. May be NULL if nothing is pending.
-	 */
-	exc->pending_sqe = sqe;
-
-	/**
-	 * Run through the queue until the last item
-	 * and take not of it
-	 */
-	while (sqe != NULL) {
-		last_sqe = sqe;
-		sqe = rtio_spsc_next(r->sq, sqe);
-	}
-
-	/* Note the last sqe for the next submit call */
-	exc->last_sqe = last_sqe;
+	/* Prepare tasks to run, they start in a suspended state */
+	conex_prepare(r, exc);
 
 	/* Resume all suspended tasks */
 	conex_resume(r, exc);
@@ -202,14 +210,9 @@ void rtio_concurrent_ok(struct rtio_iodev_sqe *iodev_sqe, int result)
 	/* Interrupt may occur in spsc_acquire, breaking the contract
 	 * so spin around it effectively preventing another interrupt on
 	 * this core, and another core trying to concurrently work in here.
-	 *
-	 * This can and should be broken up into a few sections with a try
-	 * lock around the sweep and resume.
 	 */
 	key = k_spin_lock(&exc->lock);
 
-	rtio_cqe_submit(r, result, sqe->userdata);
-
 	/* Determine the task id by memory offset O(1) */
 	uint16_t task_id = conex_task_id(exc, iodev_sqe);
 
@@ -218,16 +221,19 @@ void rtio_concurrent_ok(struct rtio_iodev_sqe *iodev_sqe, int result)
 
 		exc->task_cur[task_id].sqe = next_sqe;
 		rtio_iodev_submit(&exc->task_cur[task_id]);
-
 	} else {
 		exc->task_status[task_id]  |= CONEX_TASK_COMPLETE;
 	}
 
+	bool transaction = sqe->flags & RTIO_SQE_TRANSACTION;
+
+	while (transaction) {
+		sqe = rtio_spsc_next(r->sq, sqe);
+		transaction = sqe->flags & RTIO_SQE_TRANSACTION;
+	}
+
+	rtio_cqe_submit(r, result, sqe->userdata);
 
-	/* Sweep up unused SQEs and tasks, retry suspended tasks */
-	/* TODO Use a try lock here and don't bother doing it if we are already
-	 * doing it elsewhere
-	 */
 	conex_sweep_resume(r, exc);
 
 	k_spin_unlock(&exc->lock, key);
@@ -238,39 +244,41 @@ void rtio_concurrent_ok(struct rtio_iodev_sqe *iodev_sqe, int result)
  */
 void rtio_concurrent_err(struct rtio_iodev_sqe *iodev_sqe, int result)
 {
-	const struct rtio_sqe *nsqe;
 	k_spinlock_key_t key;
 	struct rtio *r = iodev_sqe->r;
 	const struct rtio_sqe *sqe = iodev_sqe->sqe;
 	struct rtio_concurrent_executor *exc = (struct rtio_concurrent_executor *)r->executor;
+	void *userdata = sqe->userdata;
+	bool chained = sqe->flags & RTIO_SQE_CHAINED;
+	bool transaction = sqe->flags & RTIO_SQE_TRANSACTION;
+	uint16_t task_id = conex_task_id(exc, iodev_sqe);
 
 	/* Another interrupt (and sqe complete) may occur in spsc_acquire,
 	 * breaking the contract so spin around it effectively preventing another
 	 * interrupt on this core, and another core trying to concurrently work
 	 * in here.
-	 *
-	 * This can and should be broken up into a few sections with a try
-	 * lock around the sweep and resume.
 	 */
 	key = k_spin_lock(&exc->lock);
 
-	rtio_cqe_submit(r, result, sqe->userdata);
-
-	/* Determine the task id : O(1) */
-	uint16_t task_id = conex_task_id(exc, iodev_sqe);
+	if (!transaction) {
+		rtio_cqe_submit(r, result, userdata);
+	}
 
-	sqe = iodev_sqe->sqe;
+	/* While the last sqe was marked as chained or transactional, do more work */
+	while (chained | transaction) {
+		sqe = rtio_spsc_next(r->sq, sqe);
+		chained = sqe->flags & RTIO_SQE_CHAINED;
+		transaction = sqe->flags & RTIO_SQE_TRANSACTION;
+		userdata = sqe->userdata;
 
-	/* Fail the remaining sqe's in the chain */
-	if (sqe->flags & RTIO_SQE_CHAINED) {
-		nsqe = rtio_spsc_next(r->sq, sqe);
-		while (nsqe != NULL && nsqe->flags & RTIO_SQE_CHAINED) {
-			rtio_cqe_submit(r, -ECANCELED, nsqe->userdata);
-			nsqe = rtio_spsc_next(r->sq, nsqe);
+		if (!transaction) {
+			rtio_cqe_submit(r, result, userdata);
+		} else {
+			rtio_cqe_submit(r, -ECANCELED, userdata);
 		}
 	}
 
-	/* Task is complete (failed) */
+	/* Determine the task id : O(1) */
 	exc->task_status[task_id] |= CONEX_TASK_COMPLETE;
 
 	conex_sweep_resume(r, exc);