kernel: work: add k_work_queue_run()

Add the ability to block and process a work queue by invoking
`k_work_queue_run` from an existing thread. This can be particularly useful
for using the main thread to process work items, in the same vein as the
system work queue, but from a lower priority/preemptible thread.

Signed-off-by: Bjarki Arge Andreasen <bjarki.andreasen@nordicsemi.no>
Co-authored-by: Peter Johanson <peter@peterjohanson.com>

Author: bjarki-andreasen

include/zephyr/kernel.h

@@ -3625,6 +3625,18 @@ void k_work_queue_start(struct k_work_q *queue,
 			k_thread_stack_t *stack, size_t stack_size,
 			int prio, const struct k_work_queue_config *cfg);
 
+/** @brief Run work queue using calling thread
+ *
+ * This will run the work queue forever, the function never returns
+ *
+ * @param queue the queue to run
+ *
+ * @param cfg optional additional configuration parameters.  Pass @c
+ * NULL if not required, to use the defaults documented in
+ * k_work_queue_config.
+ */
+void k_work_queue_run(struct k_work_q *queue, const struct k_work_queue_config *cfg);
+
 /** @brief Access the thread that animates a work queue.
  *
  * This is necessary to grant a work queue thread access to things the work

kernel/work.c

@@ -262,7 +262,7 @@ static inline int queue_submit_locked(struct k_work_q *queue,
 	}
 
 	int ret;
-	bool chained = (_current == &queue->thread) && !k_is_in_isr();
+	bool chained = (_current == queue->thread_id) && !k_is_in_isr();
 	bool draining = flag_test(&queue->flags, K_WORK_QUEUE_DRAIN_BIT);
 	bool plugged = flag_test(&queue->flags, K_WORK_QUEUE_PLUGGED_BIT);
 
@@ -722,6 +722,28 @@ void k_work_queue_init(struct k_work_q *queue)
 	SYS_PORT_TRACING_OBJ_INIT(k_work_queue, queue);
 }
 
+void k_work_queue_run(struct k_work_q *queue, const struct k_work_queue_config *cfg)
+{
+	__ASSERT_NO_MSG(!flag_test(&queue->flags, K_WORK_QUEUE_STARTED_BIT));
+
+	uint32_t flags = K_WORK_QUEUE_STARTED;
+
+	if ((cfg != NULL) && cfg->no_yield) {
+		flags |= K_WORK_QUEUE_NO_YIELD;
+	}
+
+	if ((cfg != NULL) && (cfg->name != NULL)) {
+		k_thread_name_set(_current, cfg->name);
+	}
+
+	sys_slist_init(&queue->pending);
+	z_waitq_init(&queue->notifyq);
+	z_waitq_init(&queue->drainq);
+	queue->thread_id = _current;
+	flags_set(&queue->flags, flags);
+	work_queue_main(queue, NULL, NULL);
+}
+
 void k_work_queue_start(struct k_work_q *queue,
 			k_thread_stack_t *stack,
 			size_t stack_size,
@@ -731,6 +753,7 @@ void k_work_queue_start(struct k_work_q *queue,
 	__ASSERT_NO_MSG(queue);
 	__ASSERT_NO_MSG(stack);
 	__ASSERT_NO_MSG(!flag_test(&queue->flags, K_WORK_QUEUE_STARTED_BIT));
+
 	uint32_t flags = K_WORK_QUEUE_STARTED;
 
 	SYS_PORT_TRACING_OBJ_FUNC_ENTER(k_work_queue, start, queue);
@@ -762,6 +785,7 @@ void k_work_queue_start(struct k_work_q *queue,
 	}
 
 	k_thread_start(&queue->thread);
+	queue->thread_id = &queue->thread;
 
 	SYS_PORT_TRACING_OBJ_FUNC_EXIT(k_work_queue, start, queue);
 }

tests/kernel/workq/work/src/main.c

@@ -95,6 +95,7 @@ static inline int coophi_counter(void)
 }
 
 static K_THREAD_STACK_DEFINE(cooplo_stack, STACK_SIZE);
+static struct k_thread cooplo_thread;
 static struct k_work_q cooplo_queue;
 static atomic_t cooplo_ctr;
 static inline int cooplo_counter(void)
@@ -141,13 +142,13 @@ static inline void reset_counters(void)
 static void counter_handler(struct k_work *work)
 {
 	last_handle_ms = k_uptime_get_32();
-	if (k_current_get() == &coophi_queue.thread) {
+	if (k_current_get() == coophi_queue.thread_id) {
 		atomic_inc(&coophi_ctr);
-	} else if (k_current_get() == &k_sys_work_q.thread) {
+	} else if (k_current_get() == k_sys_work_q.thread_id) {
 		atomic_inc(&system_ctr);
-	} else if (k_current_get() == &cooplo_queue.thread) {
+	} else if (k_current_get() == cooplo_queue.thread_id) {
 		atomic_inc(&cooplo_ctr);
-	} else if (k_current_get() == &preempt_queue.thread) {
+	} else if (k_current_get() == preempt_queue.thread_id) {
 		atomic_inc(&preempt_ctr);
 	}
 	if (atomic_dec(&resubmits_left) > 0) {
@@ -221,6 +222,21 @@ ZTEST(work, test_unstarted)
 	zassert_equal(rc, -ENODEV);
 }
 
+static void cooplo_main(void *workq_ptr, void *p2, void *p3)
+{
+	ARG_UNUSED(p2);
+	ARG_UNUSED(p3);
+
+	struct k_work_q *queue = (struct k_work_q *)workq_ptr;
+
+	struct k_work_queue_config cfg = {
+		.name = "wq.cooplo",
+		.no_yield = true,
+	};
+
+	k_work_queue_run(queue, &cfg);
+}
+
 static void test_queue_start(void)
 {
 	struct k_work_queue_config cfg = {
@@ -261,10 +277,11 @@ static void test_queue_start(void)
 	zassert_equal(coophi_queue.flags,
 		      K_WORK_QUEUE_STARTED | K_WORK_QUEUE_NO_YIELD, NULL);
 
-	cfg.name = "wq.cooplo";
-	cfg.no_yield = true;
-	k_work_queue_start(&cooplo_queue, cooplo_stack, STACK_SIZE,
-			    COOPLO_PRIORITY, &cfg);
+	(void)k_thread_create(&cooplo_thread, cooplo_stack, STACK_SIZE, cooplo_main, &cooplo_queue,
+			      NULL, NULL, COOPLO_PRIORITY, 0, K_FOREVER);
+
+	k_thread_start(&cooplo_thread);
+
 	zassert_equal(cooplo_queue.flags,
 		      K_WORK_QUEUE_STARTED | K_WORK_QUEUE_NO_YIELD, NULL);
 }