tests: rtio: Fix potential race in iodev test

Race was possible though very unlikely between the atomic cas
and queue push/pop operations. The outcome of the race had it shown up
would have been a submission not worked on due to the timer never being
started. A small critical section fixes this and clarifies where the single
conumer part of the mpsc queue comes in despite there being multiple
contexts which may enter that section.

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

Author: teburd

tests/subsys/rtio/rtio_api/src/rtio_iodev_test.h

@@ -13,61 +13,67 @@
 #define RTIO_IODEV_TEST_H_
 
 struct rtio_iodev_test_data {
-	/**
-	 * k_timer for an asynchronous task
-	 */
+	/* k_timer for an asynchronous task */
 	struct k_timer timer;
 
-	/**
-	 * Currently executing sqe
-	 */
-	atomic_ptr_t iodev_sqe;
+	/* Currently executing sqe */
+	struct rtio_iodev_sqe *iodev_sqe;
+
+	/* Count of submit calls */
+	atomic_t submit_count;
+
+	/* Lock around kicking off next timer */
+	struct k_spinlock lock;
 };
 
-static void rtio_iodev_timer_fn(struct k_timer *tm)
+static void rtio_iodev_test_next(struct rtio_iodev *iodev)
 {
-	struct rtio_iodev_test_data *data = CONTAINER_OF(tm, struct rtio_iodev_test_data, timer);
-	struct rtio_iodev_sqe *iodev_sqe = atomic_ptr_get(&data->iodev_sqe);
-	struct rtio_mpsc_node *next =
-		rtio_mpsc_pop((struct rtio_mpsc *)&iodev_sqe->sqe->iodev->iodev_sq);
+	struct rtio_iodev_test_data *data = iodev->data;
+
+	/* The next section must be serialized to ensure single consumer semantics */
+	k_spinlock_key_t key = k_spin_lock(&data->lock);
+
+	if (data->iodev_sqe != NULL) {
+		goto out;
+	}
+
+	struct rtio_mpsc_node *next = rtio_mpsc_pop(&iodev->iodev_sq);
 
 	if (next != NULL) {
 		struct rtio_iodev_sqe *next_sqe = CONTAINER_OF(next, struct rtio_iodev_sqe, q);
 
-		atomic_ptr_set(&data->iodev_sqe, next_sqe);
-		TC_PRINT("starting timer again from queued iodev_sqe %p!\n", next);
+		data->iodev_sqe = next_sqe;
 		k_timer_start(&data->timer, K_MSEC(10), K_NO_WAIT);
-	} else {
-		atomic_ptr_set(&data->iodev_sqe, NULL);
 	}
 
-	/* Complete the request with Ok and a result */
-	TC_PRINT("sqe ok callback\n");
+out:
+	k_spin_unlock(&data->lock, key);
+}
 
+static void rtio_iodev_timer_fn(struct k_timer *tm)
+{
+	struct rtio_iodev_test_data *data = CONTAINER_OF(tm, struct rtio_iodev_test_data, timer);
+	struct rtio_iodev_sqe *iodev_sqe = data->iodev_sqe;
+	struct rtio_iodev *iodev = (struct rtio_iodev *)iodev_sqe->sqe->iodev;
+
+	/* Complete the request with Ok and a result, clear the current task */
 	rtio_iodev_sqe_ok(iodev_sqe, 0);
+	data->iodev_sqe = NULL;
+
+	rtio_iodev_test_next(iodev);
 }
 
 static void rtio_iodev_test_submit(struct rtio_iodev_sqe *iodev_sqe)
 {
 	struct rtio_iodev *iodev = (struct rtio_iodev *)iodev_sqe->sqe->iodev;
 	struct rtio_iodev_test_data *data = iodev->data;
 
-	/*
-	 * If a task is already going queue up the next request in the mpsc.
-	 */
-	if (!atomic_ptr_cas(&data->iodev_sqe, NULL, iodev_sqe)) {
-		TC_PRINT("adding queued sqe\n");
-		rtio_mpsc_push(&iodev->iodev_sq, &iodev_sqe->q);
-	}
+	atomic_inc(&data->submit_count);
+
+	/* The only safe operation is enqueuing */
+	rtio_mpsc_push(&iodev->iodev_sq, &iodev_sqe->q);
 
-	/*
-	 * Simulate an async hardware request with a one shot timer
-	 *
-	 * In reality the time to complete might have some significant variance
-	 * but this is proof enough of a working API flow.
-	 */
-	TC_PRINT("starting one shot\n");
-	k_timer_start(&data->timer, K_MSEC(10), K_NO_WAIT);
+	rtio_iodev_test_next(iodev);
 }
 
 const struct rtio_iodev_api rtio_iodev_test_api = {
@@ -79,7 +85,7 @@ void rtio_iodev_test_init(struct rtio_iodev *test)
 	struct rtio_iodev_test_data *data = test->data;
 
 	rtio_mpsc_init(&test->iodev_sq);
-	atomic_ptr_set(&data->iodev_sqe, NULL);
+	data->iodev_sqe = NULL;
 	k_timer_init(&data->timer, rtio_iodev_timer_fn, NULL);
 }