userspace: dp: Add option for executing userspace module IPC in DP thread

Add the SOF_USERSPACE_MOD_IPC_BY_DP_THREAD Kconfig option and implement
support for executing userspace module IPC in its DP thread instead of
the shared user worker thread.

Signed-off-by: Adrian Warecki <adrian.warecki@intel.com>

Author: softwarecki

src/audio/module_adapter/library/userspace_proxy.c

@@ -49,6 +49,33 @@ DECLARE_TR_CTX(userspace_proxy_tr, SOF_UUID(userspace_proxy_uuid), LOG_LEVEL_INF
 
 static const struct module_interface userspace_proxy_interface;
 
+#if IS_ENABLED(CONFIG_SOF_USERSPACE_MOD_IPC_BY_DP_THREAD)
+#include <sof/audio/module_adapter/iadk/system_agent.h>
+#include <sof/schedule/dp_schedule.h>
+
+static inline int user_worker_get(void)
+{
+	return 0;
+}
+static inline void user_worker_put(void) { }
+
+struct k_work_user *userspace_proxy_register_ipc_handler(struct processing_module *mod,
+							 struct k_event *event)
+{
+	struct userspace_context * const user_ctx = mod->user_ctx;
+	if (user_ctx) {
+		tr_dbg(&userspace_proxy_tr, "Set DP event %p for module %p",
+		       (void *)event, (void *)mod);
+		user_ctx->dp_event = event;
+		user_ctx->work_item->event = event;
+
+		assert(user_ctx->work_item);
+		return &user_ctx->work_item->work_item;
+	}
+
+	return NULL;
+}
+#else
 /* IPC requests targeting userspace modules are handled through a user work queue.
  * Each userspace module provides its own work item that carries the IPC request parameters.
  * The worker thread is switched into the module's memory domain and receives the work item.
@@ -106,6 +133,7 @@ static void user_worker_put(void)
 		user_stack_free(worker.stack_ptr);
 	}
 }
+#endif
 
 static int user_work_item_init(struct userspace_context *user_ctx, struct k_heap *user_heap)
 {
@@ -128,7 +156,9 @@ static int user_work_item_init(struct userspace_context *user_ctx, struct k_heap
 
 	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;
+#endif
 	work_item->params.context = user_ctx;
 	user_ctx->work_item = work_item;
 
@@ -155,14 +185,19 @@ BUILD_ASSERT(IS_ALIGNED(MAILBOX_HOSTBOX_SIZE, CONFIG_MMU_PAGE_SIZE),
 static int userspace_proxy_invoke(struct userspace_context *user_ctx, uint32_t cmd,
 				  bool ipc_payload_access)
 {
+#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;
+#endif
 	struct module_params *params = user_work_get_params(user_ctx);
 	const uintptr_t ipc_req_buf = (uintptr_t)MAILBOX_HOSTBOX_BASE;
 	struct k_mem_partition ipc_part = {
 		.start = ipc_req_buf,
 		.size = MAILBOX_HOSTBOX_SIZE,
 		.attr = user_get_partition_attr(ipc_req_buf) | K_MEM_PARTITION_P_RO_U_RO,
 	};
-	int ret, ret2;
+	int ret = 0, ret2;
 
 	params->cmd = cmd;
 
@@ -174,6 +209,7 @@ 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);
 	if (ret < 0) {
@@ -193,9 +229,13 @@ static int userspace_proxy_invoke(struct userspace_context *user_ctx, uint32_t c
 		tr_err(&userspace_proxy_tr, "Submit to queue error: %d", ret);
 		goto done;
 	}
+#else
+	assert(event);
+	k_event_post(event, DP_TASK_EVENT_IPC);
+#endif
 
 	/* Timeout value is aligned with the ipc_wait_for_compound_msg function */
-	if (!k_event_wait_safe(&worker.event, DP_TASK_EVENT_IPC_DONE, false,
+	if (!k_event_wait_safe(event, DP_TASK_EVENT_IPC_DONE, false,
 			       Z_TIMEOUT_US(250 * 20))) {
 		tr_err(&userspace_proxy_tr, "IPC processing timedout.");
 		ret = -ETIMEDOUT;
@@ -313,18 +353,27 @@ static int userspace_proxy_start_agent(struct userspace_context *user_ctx,
 {
 	const byte_array_t * const mod_cfg = (byte_array_t *)agent_params->mod_cfg;
 	struct module_params *params = user_work_get_params(user_ctx);
-	int ret;
 
 	params->ext.agent.start_fn = start_fn;
-	params->ext.agent.params = *agent_params;
-	params->ext.agent.mod_cfg = *mod_cfg;
 
-	ret = userspace_proxy_invoke(user_ctx, USER_PROXY_MOD_CMD_AGENT_START, true);
-	if (ret)
-		return ret;
+	/* Start the system agent, if provided. */
+	if (start_fn) {
+		params->ext.agent.params = *agent_params;
+		params->ext.agent.mod_cfg = *mod_cfg;
 
-	*agent_interface = params->ext.agent.out_interface;
-	return params->status;
+		/* In case of processing modules ipc in the DP thread, the agent will be started in the
+		 * init function. At this point the DP thread does not exist yet.
+		 */
+#if !IS_ENABLED(CONFIG_SOF_USERSPACE_MOD_IPC_BY_DP_THREAD)
+		int ret = userspace_proxy_invoke(user_ctx, USER_PROXY_MOD_CMD_AGENT_START, true);
+		if (ret)
+			return ret;
+
+		*agent_interface = params->ext.agent.out_interface;
+		return params->status;
+#endif
+	}
+	return 0;
 }
 
 int userspace_proxy_create(struct userspace_context **user_ctx, const struct comp_driver *drv,
@@ -362,14 +411,10 @@ int userspace_proxy_create(struct userspace_context **user_ctx, const struct com
 	if (ret)
 		goto error_dom;
 
-	/* Start the system agent, if provided. */
-
-	if (start_fn) {
-		ret = userspace_proxy_start_agent(context, start_fn, agent_params, agent_interface);
-		if (ret) {
-			tr_err(&userspace_proxy_tr, "System agent failed with error %d.", ret);
-			goto error_work_item;
-		}
+	ret = userspace_proxy_start_agent(context, start_fn, agent_params, agent_interface);
+	if (ret) {
+		tr_err(&userspace_proxy_tr, "System agent failed with error %d.", ret);
+		goto error_work_item;
 	}
 
 	*user_ctx = context;
@@ -420,6 +465,22 @@ static int userspace_proxy_init(struct processing_module *mod)
 
 	comp_dbg(mod->dev, "start");
 
+#if IS_ENABLED(CONFIG_SOF_USERSPACE_MOD_IPC_BY_DP_THREAD)
+	/* Start the system agent, if provided. Params is already filled by
+	 * the userspace_proxy_start_agent function.
+	 */
+	if (params->ext.agent.start_fn) {
+		ret = userspace_proxy_invoke(mod->user_ctx, USER_PROXY_MOD_CMD_AGENT_START, true);
+		if (ret)
+			return ret;
+
+		if (params->ext.agent.start_fn == system_agent_start)
+			module_set_private_data(mod, (void*)params->ext.agent.out_interface);
+		else
+			mod->user_ctx->interface = params->ext.agent.out_interface;
+	}
+#endif
+
 	params->mod = mod;
 	ret = userspace_proxy_invoke(mod->user_ctx, USER_PROXY_MOD_CMD_INIT, true);
 	if (ret)

src/audio/module_adapter/module/generic.c

@@ -622,7 +622,8 @@ int module_reset(struct processing_module *mod)
 
 	/* cancel task if DP task*/
 	if (mod->dev->ipc_config.proc_domain == COMP_PROCESSING_DOMAIN_DP && mod->dev->task &&
-	    !IS_ENABLED(CONFIG_SOF_USERSPACE_APPLICATION))
+	    !IS_ENABLED(CONFIG_SOF_USERSPACE_APPLICATION) &&
+	    !IS_ENABLED(CONFIG_SOF_USERSPACE_MOD_IPC_BY_DP_THREAD))
 		schedule_task_cancel(mod->dev->task);
 
 	if (ops->reset) {

src/audio/module_adapter/module_adapter.c

@@ -1432,23 +1432,24 @@ void module_adapter_free(struct comp_dev *dev)
 
 	comp_dbg(dev, "start");
 
-	if (dev->task) {
+#if CONFIG_SOF_USERSPACE_APPLICATION
+	if (dev->task)
 		/*
 		 * Run DP module's .free() method in its thread context.
 		 * Unlike with other IPCs we first run module's .free() in
 		 * thread context, then cancel the thread, and then execute
 		 * final clean up
 		 */
-#if CONFIG_SOF_USERSPACE_APPLICATION
 		scheduler_dp_thread_ipc(mod, SOF_IPC4_MOD_DELETE_INSTANCE, NULL);
 #endif
-		schedule_task_free(dev->task);
-	}
 
 	ret = module_free(mod);
 	if (ret)
 		comp_err(dev, "failed with error: %d", ret);
 
+	if (dev->task)
+		schedule_task_free(dev->task);
+
 	list_for_item_safe(blist, _blist, &mod->raw_data_buffers_list) {
 		struct comp_buffer *buffer = container_of(blist, struct comp_buffer,
 							  buffers_list);

src/include/sof/audio/module_adapter/library/userspace_proxy.h

@@ -31,6 +31,7 @@ struct userspace_context {
 	struct k_mem_domain *comp_dom;			/* Module specific memory domain	*/
 	const struct module_interface *interface;	/* Userspace module interface		*/
 	struct user_work_item *work_item;		/* work item for user worker thread	*/
+	struct k_event *dp_event;			/* DP thread event			*/
 };
 #endif /* CONFIG_USERSPACE */
 
@@ -63,6 +64,20 @@ int userspace_proxy_create(struct userspace_context **user_ctx, const struct com
  */
 void userspace_proxy_destroy(const struct comp_driver *drv, struct userspace_context *user_ctx);
 
+#if IS_ENABLED(CONFIG_SOF_USERSPACE_MOD_IPC_BY_DP_THREAD)
+/**
+ * Register a k_event object used to notify the DP thread about a pending IPC
+ * request to process.
+ *
+ * @param mod    Pointer to the processing module.
+ * @param event  Pointer to the event to signal incoming IPC.
+ *
+ * @return Pointer to a k_work_user work item on success, or NULL on failure.
+ */
+struct k_work_user *userspace_proxy_register_ipc_handler(struct processing_module *mod,
+							 struct k_event *event);
+#endif
+
 #endif /* CONFIG_SOF_USERSPACE_PROXY */
 
 #endif /* __SOF_AUDIO_USERSPACE_PROXY_H__ */

src/schedule/zephyr_dp_schedule.h

@@ -32,6 +32,7 @@ enum sof_dp_part_type {
 };
 
 struct ipc4_flat;
+
 struct task_dp_pdata {
 	k_tid_t thread_id;		/* zephyr thread ID */
 	struct k_thread *thread;	/* pointer to the kernels' thread object */
@@ -46,6 +47,9 @@ struct task_dp_pdata {
 #endif
 	struct k_event *event;		/* pointer to event for task scheduling */
 	struct k_event event_struct;	/* event for task scheduling for kernel threads */
+#if IS_ENABLED(CONFIG_SOF_USERSPACE_MOD_IPC_BY_DP_THREAD)
+	struct k_work_user *ipc_work_item;	/* work item for IPC handling */
+#endif
 };
 
 void scheduler_dp_recalculate(struct scheduler_dp_data *dp_sch, bool is_ll_post_run);

src/schedule/zephyr_dp_schedule_thread.c

@@ -3,12 +3,14 @@
  * Copyright(c) 2025 Intel Corporation. All rights reserved.
  *
  * Author: Marcin Szkudlinski
+ *	   Adrian Warecki
  */
 
 #include <rtos/task.h>
 
 #include <sof/audio/module_adapter/module/generic.h>
 #include <sof/audio/module_adapter/library/userspace_proxy.h>
+#include <sof/audio/module_adapter/library/userspace_proxy_user.h>
 #include <sof/common.h>
 #include <sof/list.h>
 #include <sof/schedule/ll_schedule_domain.h>
@@ -115,6 +117,7 @@ void dp_thread_fn(void *p1, void *p2, void *p3)
 	unsigned int lock_key;
 	enum task_state state;
 	bool task_stop;
+	uint32_t event;
 
 	if (!(task->flags & K_USER))
 		dp_sch = scheduler_get_data(SOF_SCHEDULE_DP);
@@ -124,50 +127,62 @@ void dp_thread_fn(void *p1, void *p2, void *p3)
 		 * the thread is started immediately after creation, it will stop on event.
 		 * Event will be signalled once the task is ready to process.
 		 */
-		k_event_wait_safe(task_pdata->event, DP_TASK_EVENT_PROCESS | DP_TASK_EVENT_CANCEL,
-				  false, K_FOREVER);
+		event = k_event_wait_safe(task_pdata->event, DP_TASK_EVENT_PROCESS |
+					  DP_TASK_EVENT_CANCEL | DP_TASK_EVENT_IPC, false,
+					  K_FOREVER);
+
+#if IS_ENABLED(CONFIG_SOF_USERSPACE_MOD_IPC_BY_DP_THREAD)
+		if (event & DP_TASK_EVENT_IPC) {
+			assert(task_pdata->ipc_work_item);
+			userspace_proxy_worker_handler(task_pdata->ipc_work_item);
+		}
+#endif
 
-		if (task->state == SOF_TASK_STATE_RUNNING)
-			state = task_run(task);
-		else
+		if (event & DP_TASK_EVENT_PROCESS) {
 			state = task->state;	/* to avoid undefined variable warning */
+			if (task->state == SOF_TASK_STATE_RUNNING && event & DP_TASK_EVENT_PROCESS)
+				state = task_run(task);
 
-		lock_key = scheduler_dp_lock(task->core);
-		/*
-		 * check if task is still running, may have been canceled by external call
-		 * if not, set the state returned by run procedure
-		 */
-		if (task->state == SOF_TASK_STATE_RUNNING) {
-			task->state = state;
-			switch (state) {
-			case SOF_TASK_STATE_RESCHEDULE:
-				/* mark to reschedule, schedule time is already calculated */
-				task->state = SOF_TASK_STATE_QUEUED;
-				break;
-
-			case SOF_TASK_STATE_CANCEL:
-			case SOF_TASK_STATE_COMPLETED:
-				/* remove from scheduling */
-				list_item_del(&task->list);
-				break;
-
-			default:
-				/* illegal state, serious defect, won't happen */
-				k_panic();
+			lock_key = scheduler_dp_lock(task->core);
+			/*
+			 * check if task is still running, may have been canceled by external call
+			 * if not, set the state returned by run procedure
+			 */
+			if (task->state == SOF_TASK_STATE_RUNNING) {
+				task->state = state;
+				switch (state) {
+				case SOF_TASK_STATE_RESCHEDULE:
+					/* mark to reschedule, schedule time is already calculated */
+					task->state = SOF_TASK_STATE_QUEUED;
+					break;
+
+				case SOF_TASK_STATE_CANCEL:
+				case SOF_TASK_STATE_COMPLETED:
+					/* remove from scheduling */
+					list_item_del(&task->list);
+					break;
+
+				default:
+					/* illegal state, serious defect, won't happen */
+					k_panic();
+				}
 			}
-		}
 
-		/* if true exit the while loop, terminate the thread */
-		task_stop = task->state == SOF_TASK_STATE_COMPLETED ||
-			task->state == SOF_TASK_STATE_CANCEL;
-		/* recalculate all DP tasks readiness and deadlines
-		 * TODO: it should be for all tasks, for all cores
-		 * currently its limited to current core only
-		 */
-		if (dp_sch)
-			scheduler_dp_recalculate(dp_sch, false);
+			/* if true exit the while loop, terminate the thread */
+			task_stop = task->state == SOF_TASK_STATE_COMPLETED ||
+				task->state == SOF_TASK_STATE_CANCEL;
+			/* recalculate all DP tasks readiness and deadlines
+			 * TODO: it should be for all tasks, for all cores
+			 * currently its limited to current core only
+			 */
+			if (dp_sch)
+				scheduler_dp_recalculate(dp_sch, false);
+
+			scheduler_dp_unlock(lock_key);
+		}
 
-		scheduler_dp_unlock(lock_key);
+		if (event & DP_TASK_EVENT_CANCEL)
+			task_stop = true;
 	} while (!task_stop);
 
 	/* call task_complete  */
@@ -294,6 +309,10 @@ int scheduler_dp_task_init(struct task **task,
 	k_event_init(pdata->event);
 	k_thread_start(pdata->thread_id);
 
+#if IS_ENABLED(CONFIG_SOF_USERSPACE_MOD_IPC_BY_DP_THREAD)
+	pdata->ipc_work_item = userspace_proxy_register_ipc_handler(mod, pdata->event);
+#endif
+
 	/* success, fill output parameter */
 	*task = &task_memory->task;
 	return 0;