userspace: proxy: fix multicore work queue handling

src/audio/module_adapter/library/userspace_proxy.c

@@ -30,6 +30,7 @@
 #include <sof/audio/component.h>
 #include <sof/schedule/dp_schedule.h>
 #include <rtos/userspace_helper.h>
+#include <rtos/user_work.h>
 #include <utilities/array.h>
 #include <zephyr/sys/sem.h>
 #include <sof/audio/module_adapter/module/generic.h>
@@ -54,12 +55,16 @@ static const struct module_interface userspace_proxy_interface;
 #include <sof/audio/module_adapter/iadk/system_agent.h>
 #include <sof/schedule/dp_schedule.h>
 
-static inline int user_worker_get(void)
+static inline int user_worker_get(int core)
 {
+	ARG_UNUSED(core);
 	return 0;
 }
 
-static inline void user_worker_put(void) { }
+static inline void user_worker_put(int core)
+{
+	ARG_UNUSED(core);
+}
 
 struct k_work_user *userspace_proxy_register_ipc_handler(struct processing_module *mod,
 							 struct k_event *event)
@@ -86,9 +91,9 @@ struct k_work_user *userspace_proxy_register_ipc_handler(struct processing_modul
  * It invokes the appropriate module function in userspace context and writes the operation
  * result back into the work item.
  *
- * There is only a single work queue, which is shared by all userspace modules. It is created
- * dynamically when needed. Because SOF uses a single dedicated thread for handling IPC, there
- * is no need to perform any additional serialization when accessing the worker.
+ * Each core owns one work queue, shared by the userspace modules assigned to that core.
+ * Each work queue has its own worker thread, which is pinned to its core before it is first
+ * started and never migrates.
  */
 struct user_worker {
 	k_tid_t thread_id;			/* ipc worker thread ID			*/
@@ -98,70 +103,94 @@ struct user_worker {
 	struct k_event event;
 };
 
-static struct user_worker worker;
+static struct user_worker workers[CONFIG_CORE_COUNT];
 
-static int user_worker_get(void)
+static int user_worker_get(int core)
 {
-	if (worker.reference_count) {
-		worker.reference_count++;
+	struct user_worker *worker;
+	int ret;
+
+	if (core < 0 || core >= CONFIG_CORE_COUNT)
+		return -EINVAL;
+
+	worker = &workers[core];
+	if (worker->reference_count) {
+		worker->reference_count++;
 		return 0;
 	}
 
-	worker.stack_ptr = user_stack_allocate(CONFIG_SOF_USERSPACE_PROXY_WORKER_STACK_SIZE,
-					       K_USER);
-	if (!worker.stack_ptr) {
+	worker->stack_ptr = user_stack_allocate(CONFIG_SOF_USERSPACE_PROXY_WORKER_STACK_SIZE,
+						K_USER);
+	if (!worker->stack_ptr) {
 		tr_err(&userspace_proxy_tr, "Userspace worker stack allocation failed.");
 		return -ENOMEM;
 	}
 
-	k_event_init(&worker.event);
-	k_work_user_queue_start(&worker.work_queue, worker.stack_ptr,
-				CONFIG_SOF_USERSPACE_PROXY_WORKER_STACK_SIZE, 0, NULL);
+	k_event_init(&worker->event);
+	sof_work_user_queue_create(&worker->work_queue, worker->stack_ptr,
+				   CONFIG_SOF_USERSPACE_PROXY_WORKER_STACK_SIZE, 0, NULL);
+	worker->thread_id = k_work_user_queue_thread_get(&worker->work_queue);
+	k_thread_access_grant(worker->thread_id, &worker->event);
 
-	worker.thread_id = k_work_user_queue_thread_get(&worker.work_queue);
+#ifdef CONFIG_SCHED_CPU_MASK
+	ret = k_thread_cpu_pin(worker->thread_id, core);
+	if (ret < 0) {
+		tr_err(&userspace_proxy_tr, "Failed to pin userspace worker to core %d: %d",
+		       core, ret);
+		k_thread_abort(worker->thread_id);
+		user_stack_free(worker->stack_ptr);
+		worker->stack_ptr = NULL;
+		worker->thread_id = NULL;
+		return ret;
+	}
+#elif CONFIG_CORE_COUNT > 1
+#error "CONFIG_SCHED_CPU_MASK is not enabled"
+#endif
 
-	k_thread_access_grant(worker.thread_id, &worker.event);
+	k_thread_start(worker->thread_id);
 
-	worker.reference_count++;
+	worker->reference_count++;
 	return 0;
 }
 
-static void user_worker_put(void)
+static void user_worker_put(int core)
 {
+	struct user_worker *worker = &workers[core];
+
 	/* Module removed so decrement counter */
-	worker.reference_count--;
+	worker->reference_count--;
 
 	/* Free worker resources if no more active user space modules */
-	if (worker.reference_count == 0) {
-		k_thread_abort(worker.thread_id);
-		user_stack_free(worker.stack_ptr);
+	if (worker->reference_count == 0) {
+		k_thread_abort(worker->thread_id);
+		user_stack_free(worker->stack_ptr);
+		worker->stack_ptr = NULL;
+		worker->thread_id = NULL;
 	}
 }
 #endif
 
 static int user_work_item_init(struct userspace_context *user_ctx, struct k_heap *user_heap)
 {
 	struct user_work_item *work_item = NULL;
+	const int core = cpu_get_id();
 	int ret;
 
-	ret = user_worker_get();
+	ret = user_worker_get(core);
 	if (ret)
 		return ret;
 
-	/* We have only a single userspace IPC worker. It handles requests for all userspace
-	 * modules, which may run on different cores. Because the worker processes work items
-	 * coming from any core, the work item must be allocated in coherent memory.
-	 */
-	work_item = sof_heap_alloc(user_heap, SOF_MEM_FLAG_COHERENT, sizeof(*work_item), 0);
+	/* Work item is allocated, submitted and processed on the same core -- can be cached. */
+	work_item = sof_heap_alloc(user_heap, 0, sizeof(*work_item), 0);
 	if (!work_item) {
-		user_worker_put();
+		user_worker_put(core);
 		return -ENOMEM;
 	}
 
 	k_work_user_init(&work_item->work_item, userspace_proxy_worker_handler);
 
 #if !IS_ENABLED(CONFIG_SOF_USERSPACE_MOD_IPC_BY_DP_THREAD)
-	work_item->event = &worker.event;
+	work_item->event = &workers[core].event;
 #endif
 	work_item->params.context = user_ctx;
 	work_item->params.mod = NULL;
@@ -173,7 +202,7 @@ static int user_work_item_init(struct userspace_context *user_ctx, struct k_heap
 static void user_work_item_free(struct userspace_context *user_ctx, struct k_heap *user_heap)
 {
 	sof_heap_free(user_heap, user_ctx->work_item);
-	user_worker_put();
+	user_worker_put(cpu_get_id());
 }
 
 static inline struct module_params *user_work_get_params(struct userspace_context *user_ctx)
@@ -193,7 +222,8 @@ static int userspace_proxy_invoke(struct userspace_context *user_ctx, uint32_t c
 #if IS_ENABLED(CONFIG_SOF_USERSPACE_MOD_IPC_BY_DP_THREAD)
 	struct k_event * const event = user_ctx->dp_event;
 #else
-	struct k_event * const event = &worker.event;
+	struct user_worker *worker = &workers[cpu_get_id()];
+	struct k_event * const event = &worker->event;
 #endif
 	struct module_params *params = user_work_get_params(user_ctx);
 	const uintptr_t ipc_req_buf = (uintptr_t)MAILBOX_HOSTBOX_BASE;
@@ -216,22 +246,13 @@ static int userspace_proxy_invoke(struct userspace_context *user_ctx, uint32_t c
 
 #if !IS_ENABLED(CONFIG_SOF_USERSPACE_MOD_IPC_BY_DP_THREAD)
 	/* Switch worker thread to module memory domain */
-	ret = k_mem_domain_add_thread(user_ctx->comp_dom, worker.thread_id);
+	ret = k_mem_domain_add_thread(user_ctx->comp_dom, worker->thread_id);
 	if (ret < 0) {
 		tr_err(&userspace_proxy_tr, "Failed to switch memory domain, error: %d", ret);
 		goto done;
 	}
 
-#ifdef CONFIG_SCHED_CPU_MASK
-	/* Pin worker thread to the same core as the module */
-	ret = k_thread_cpu_pin(worker.thread_id, cpu_get_id());
-	if (ret < 0) {
-		tr_err(&userspace_proxy_tr, "Failed to pin cpu, error: %d", ret);
-		goto done;
-	}
-#endif
-
-	ret = k_work_user_submit_to_queue(&worker.work_queue, &user_ctx->work_item->work_item);
+	ret = k_work_user_submit_to_queue(&worker->work_queue, &user_ctx->work_item->work_item);
 	if (ret < 0) {
 		tr_err(&userspace_proxy_tr, "Submit to queue error: %d", ret);
 		goto done;

zephyr/CMakeLists.txt

@@ -599,6 +599,10 @@ zephyr_library_sources(
 	lib.c
 )
 
+zephyr_library_sources_ifndef(CONFIG_SOF_USERSPACE_MOD_IPC_BY_DP_THREAD
+	lib/user_work.c
+)
+
 # SOF module interface functions
 add_subdirectory(../src/module module_unused_install/)
 

zephyr/include/rtos/user_work.h

@@ -0,0 +1,28 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ *
+ * Copyright 2026 Intel Corporation. All rights reserved.
+ */
+
+#ifndef __ZEPHYR_RTOS_USER_WORK_H__
+#define __ZEPHYR_RTOS_USER_WORK_H__
+
+#include <zephyr/kernel.h>
+
+/**
+ * Create a userspace work queue without starting its worker thread.
+ *
+ * Based on Zephyr's k_work_user_queue_create(), but does not start its
+ * worker thread. The caller must configure the thread, for example, pin it
+ * to the appropriate core, and call k_thread_start() when it is ready to
+ * process work.
+ *
+ * @param work_q Work queue to initialize.
+ * @param stack Worker thread stack.
+ * @param stack_size Size of @p stack in bytes.
+ * @param prio Worker thread priority.
+ * @param name Optional worker thread name.
+ */
+void sof_work_user_queue_create(struct k_work_user_q *work_q, k_thread_stack_t *stack,
+				size_t stack_size, int prio, const char *name);
+
+#endif /* __ZEPHYR_RTOS_USER_WORK_H__ */

zephyr/lib/user_work.c

@@ -0,0 +1,86 @@
+/*
+ * Copyright (c) 2018 Intel Corporation
+ * Copyright (c) 2016 Wind River Systems, Inc.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+/*
+ * This file is mostly copied from Zephyr's user_work.c. Retain the original
+ * license header above.
+ */
+
+/*
+ * On Xtensa, with CONFIG_SCHED_CPU_MASK_PIN_ONLY=y, k_thread_cpu_pin()
+ * cannot safely change a thread's core after it has been active on another
+ * core. Enabling CONFIG_SCHED_CPU_MASK_PIN_ONLY also enables an optimization
+ * in the thread-switch code: because Zephyr assumes that a thread will resume
+ * on the same core, it does not write back the thread's cached stack.
+ *
+ * If k_thread_cpu_pin() subsequently moves the thread to another core, the
+ * cached stack on the new core contains garbage. The cached stack was not
+ * written back on the previous core and is not invalidated on the new core.
+ * With CONFIG_SCHED_CPU_MASK_PIN_ONLY disabled, Zephyr writes back the stack
+ * when the thread becomes inactive and invalidates it when the thread becomes
+ * active on another core.
+ *
+ * k_work_user_queue_create() starts its worker thread on core 0, leaving no
+ * safe way to re-pin it when CONFIG_SCHED_CPU_MASK_PIN_ONLY=y. Because we do
+ * not want to disable CONFIG_SCHED_CPU_MASK_PIN_ONLY or use an uncached stack,
+ * this file implements a copy of k_work_user_queue_create() that does not
+ * start the worker thread. The caller configures the thread, for example by
+ * pinning it to the required core, and then calls k_thread_start().
+ */
+
+#include <zephyr/kernel.h>
+
+/* This is an intact copy of Zephyr's z_work_user_q_main(). */
+static void z_work_user_q_main(void *work_q_ptr, void *p2, void *p3)
+{
+	struct k_work_user_q *work_q = work_q_ptr;
+
+	ARG_UNUSED(p2);
+	ARG_UNUSED(p3);
+
+	while (true) {
+		struct k_work_user *work;
+		k_work_user_handler_t handler;
+
+		work = k_queue_get(&work_q->queue, K_FOREVER);
+		if (work == NULL) {
+			continue;
+		}
+
+		handler = work->handler;
+		__ASSERT(handler != NULL, "handler must be provided");
+
+		/* Reset pending state so it can be resubmitted by handler */
+		if (atomic_test_and_clear_bit(&work->flags,
+					      K_WORK_USER_STATE_PENDING)) {
+			handler(work);
+		}
+
+		/* Make sure we don't hog up the CPU if the FIFO never (or
+		 * very rarely) gets empty.
+		 */
+		k_yield();
+	}
+}
+
+/* This is Zephyr's k_work_user_queue_create() with k_thread_start() removed. */
+void sof_work_user_queue_create(struct k_work_user_q *work_q, k_thread_stack_t *stack,
+			    size_t stack_size, int prio, const char *name)
+{
+	k_queue_init(&work_q->queue);
+
+	/* Created worker thread will inherit object permissions and memory
+	 * domain configuration of the caller
+	 */
+	k_thread_create(&work_q->thread, stack, stack_size, z_work_user_q_main,
+			work_q, NULL, NULL, prio, K_USER | K_INHERIT_PERMS,
+			K_FOREVER);
+	k_object_access_grant(&work_q->queue, &work_q->thread);
+	if (name != NULL) {
+		k_thread_name_set(&work_q->thread, name);
+	}
+}