rtio: Introduce OP_DELAY as a valid SQE operation

SQE items with this OP will take the specified amount of time
(asynchronously) before completing. This allows to serve as an
asynchronous delay in between SQE items (e.g: A sensor measurement
requested, which requires 50-ms before having the result available).

Signed-off-by: Luis Ubieda <[email protected]>

Author: ubieda

include/zephyr/rtio/rtio.h

@@ -35,6 +35,7 @@
 #include <zephyr/sys/atomic.h>
 #include <zephyr/sys/mem_blocks.h>
 #include <zephyr/sys/util.h>
+#include <zephyr/sys_clock.h>
 #include <zephyr/sys/iterable_sections.h>
 #include <zephyr/sys/mpsc_lockfree.h>
 
@@ -336,6 +337,13 @@ struct rtio_sqe {
 			uint8_t *rx_buf; /**< Buffer to read into */
 		} txrx;
 
+		/** OP_DELAY */
+		struct {
+			k_timeout_t timeout; /**< Delay timeout. */
+			k_timepoint_t deadline; /**< Absolute deadline. Used internally. */
+			sys_snode_t node; /**< Slist node. Used internally */
+		} delay;
+
 		/** OP_I2C_CONFIGURE */
 		uint32_t i2c_config;
 
@@ -540,8 +548,11 @@ struct rtio_iodev {
 /** An operation that transceives (reads and writes simultaneously) */
 #define RTIO_OP_TXRX (RTIO_OP_CALLBACK+1)
 
+/** An operation that takes a specified amount of time (asynchronously) before completing */
+#define RTIO_OP_DELAY (RTIO_OP_TXRX+1)
+
 /** An operation to recover I2C buses */
-#define RTIO_OP_I2C_RECOVER (RTIO_OP_TXRX+1)
+#define RTIO_OP_I2C_RECOVER (RTIO_OP_DELAY+1)
 
 /** An operation to configure I2C buses */
 #define RTIO_OP_I2C_CONFIGURE (RTIO_OP_I2C_RECOVER+1)
@@ -717,6 +728,18 @@ static inline void rtio_sqe_prep_transceive(struct rtio_sqe *sqe,
 	sqe->userdata = userdata;
 }
 
+static inline void rtio_sqe_prep_delay(struct rtio_sqe *sqe,
+				       k_timeout_t timeout,
+				       void *userdata)
+{
+	memset(sqe, 0, sizeof(struct rtio_sqe));
+	sqe->op = RTIO_OP_DELAY;
+	sqe->prio = 0;
+	sqe->iodev = NULL;
+	sqe->delay.timeout = timeout;
+	sqe->userdata = userdata;
+}
+
 static inline struct rtio_iodev_sqe *rtio_sqe_pool_alloc(struct rtio_sqe_pool *pool)
 {
 	struct mpsc_node *node = mpsc_pop(&pool->free_q);

subsys/rtio/CMakeLists.txt

@@ -10,6 +10,7 @@ if(CONFIG_RTIO)
 
 	zephyr_library_sources(rtio_executor.c)
 	zephyr_library_sources(rtio_init.c)
+	zephyr_library_sources(rtio_sched.c)
 	zephyr_library_sources_ifdef(CONFIG_USERSPACE rtio_handlers.c)
 endif()
 

subsys/rtio/rtio_executor.c

@@ -7,6 +7,8 @@
 #include <zephyr/rtio/rtio.h>
 #include <zephyr/kernel.h>
 
+#include "rtio_sched.h"
+
 #include <zephyr/logging/log.h>
 LOG_MODULE_REGISTER(rtio_executor, CONFIG_RTIO_LOG_LEVEL);
 
@@ -22,6 +24,9 @@ static void rtio_executor_op(struct rtio_iodev_sqe *iodev_sqe)
 		sqe->callback.callback(iodev_sqe->r, sqe, sqe->callback.arg0);
 		rtio_iodev_sqe_ok(iodev_sqe, 0);
 		break;
+	case RTIO_OP_DELAY:
+		rtio_sched_alarm(iodev_sqe, sqe->delay.timeout);
+		break;
 	default:
 		rtio_iodev_sqe_err(iodev_sqe, -EINVAL);
 	}

subsys/rtio/rtio_sched.c

@@ -0,0 +1,94 @@
+/*
+ * Copyright (c) 2025 Croxel Inc.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <zephyr/kernel.h>
+#include <zephyr/rtio/rtio.h>
+#include <zephyr/spinlock.h>
+#include "rtio_sched.h"
+
+static struct k_spinlock rio_sched_lock;
+static sys_slist_t rtio_sched_list = SYS_SLIST_STATIC_INIT(&rtio_sched_list);
+
+static inline void add_alarm_to_list(struct rtio_sqe *sqe)
+{
+	struct rtio_sqe *curr_sqe, *prev_sqe = NULL;
+
+	/** We assume we have a sorted list: earlier alarms go first */
+	SYS_SLIST_FOR_EACH_CONTAINER(&rtio_sched_list, curr_sqe, delay.node) {
+		if (sys_timepoint_cmp(sqe->delay.deadline, curr_sqe->delay.deadline) <= 0) {
+			sys_slist_insert(&rtio_sched_list,
+					 (prev_sqe ? &prev_sqe->delay.node : NULL),
+					 &sqe->delay.node);
+			return;
+		}
+		prev_sqe = curr_sqe;
+	}
+
+	sys_slist_append(&rtio_sched_list, &sqe->delay.node);
+}
+
+static inline void remove_alarm_from_list(struct rtio_sqe *sqe)
+{
+	sys_slist_find_and_remove(&rtio_sched_list, &sqe->delay.node);
+}
+
+static inline struct rtio_sqe *get_next_alarm_from_list(void)
+{
+	sys_snode_t *node = sys_slist_peek_head(&rtio_sched_list);
+
+	if (node == NULL) {
+		return NULL;
+	}
+
+	struct rtio_sqe *sqe = CONTAINER_OF(node, struct rtio_sqe, delay.node);
+
+	return sqe;
+}
+
+static void rtio_sched_alarm_expired(struct k_timer *timer)
+{
+	struct rtio_iodev_sqe *iodev_sqe;
+
+	K_SPINLOCK(&rio_sched_lock) {
+		struct rtio_sqe *sqe = get_next_alarm_from_list();
+
+		iodev_sqe = CONTAINER_OF(sqe, struct rtio_iodev_sqe, sqe);
+
+		remove_alarm_from_list(sqe);
+
+		sqe = get_next_alarm_from_list();
+		if (sqe != NULL) {
+			k_timer_start(timer,
+				      sys_timepoint_timeout(sqe->delay.deadline),
+				      K_NO_WAIT);
+		}
+	}
+
+	rtio_iodev_sqe_ok(iodev_sqe, 0);
+}
+
+static K_TIMER_DEFINE(rtio_delay_timer, rtio_sched_alarm_expired, NULL);
+
+void rtio_sched_alarm(struct rtio_iodev_sqe *iodev_sqe, k_timeout_t timeout)
+{
+	struct rtio_sqe *sqe = &iodev_sqe->sqe;
+	k_timepoint_t *sqe_deadline = &sqe->delay.deadline;
+
+	*sqe_deadline = sys_timepoint_calc(timeout);
+
+	K_SPINLOCK(&rio_sched_lock) {
+		struct rtio_sqe *next_sqe;
+
+		add_alarm_to_list(sqe);
+
+		next_sqe = get_next_alarm_from_list();
+
+		/** We assume we have an alarm to schedule, since we just added a new one */
+		k_timer_start(&rtio_delay_timer,
+			      sys_timepoint_timeout(next_sqe->delay.deadline),
+			      K_NO_WAIT);
+	}
+}

subsys/rtio/rtio_sched.h

@@ -0,0 +1,14 @@
+/*
+ * Copyright (c) 2025 Croxel Inc.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <zephyr/kernel.h>
+
+#ifndef ZEPHYR_SUBSYS_RTIO_SCHED_H_
+#define ZEPHYR_SUBSYS_RTIO_SCHED_H_
+
+void rtio_sched_alarm(struct rtio_iodev_sqe *iodev_sqe, k_timeout_t timeout);
+
+#endif /* ZEPHYR_SUBSYS_RTIO_SCHED_H_ */