audio: pipeline: change locking strategy for user LL builds

src/audio/pipeline/pipeline-graph.c

@@ -29,6 +29,7 @@
 #include <stdbool.h>
 #include <stddef.h>
 #include <stdint.h>
+#include <sof/schedule/ll_schedule_domain.h>
 
 LOG_MODULE_REGISTER(pipe, CONFIG_SOF_LOG_LEVEL);
 
@@ -181,19 +182,22 @@ static void buffer_set_comp(struct comp_buffer *buffer, struct comp_dev *comp,
 }
 
 #ifdef CONFIG_SOF_USERSPACE_LL
+/*
+ * User-space LL: IPC-level user_ll_lock_sched() provides mutual
+ * exclusion with the LL thread. No per-pipeline lock needed.
+ */
 #define PPL_LOCK_DECLARE
-#define PPL_LOCK() do { \
-		int ret = sys_mutex_lock(&comp->list_mutex, K_FOREVER); \
-		assert(ret == 0); \
-	} while (0)
-#define PPL_UNLOCK() do { \
-		int ret = sys_mutex_unlock(&comp->list_mutex); \
-		assert(ret == 0); \
-	} while (0)
+#define PPL_LOCK(x) user_ll_lock_sched(x)
+#define PPL_UNLOCK(x) user_ll_unlock_sched(x)
 #else
+/*
+ * Kernel-space LL. When modifying pipeline connections, block IRQs
+ * and prevent LL from running. No locking needed when iterating
+ * the pipeline in the LL thread.
+ */
 #define PPL_LOCK_DECLARE uint32_t flags
-#define PPL_LOCK() irq_local_disable(flags)
-#define PPL_UNLOCK()  irq_local_enable(flags)
+#define PPL_LOCK(x) irq_local_disable(flags)
+#define PPL_UNLOCK(x)  irq_local_enable(flags)
 #endif
 
 int pipeline_connect(struct comp_dev *comp, struct comp_buffer *buffer,
@@ -208,19 +212,19 @@ int pipeline_connect(struct comp_dev *comp, struct comp_buffer *buffer,
 	else
 		comp_info(comp, "connect buffer %d as source", buf_get_id(buffer));
 
-	PPL_LOCK();
+	PPL_LOCK(buffer->core);
 
 	comp_list = comp_buffer_list(comp, dir);
 	ret = buffer_attach(buffer, comp_list, dir);
 	if (ret < 0) {
 		comp_err(comp, "buffer %d already connected dir %d",
 			 buf_get_id(buffer), dir);
-		PPL_UNLOCK();
+		PPL_UNLOCK(buffer->core);
 		return ret;
 	}
 	buffer_set_comp(buffer, comp, dir);
 
-	PPL_UNLOCK();
+	PPL_UNLOCK(buffer->core);
 
 	return 0;
 }
@@ -235,13 +239,13 @@ void pipeline_disconnect(struct comp_dev *comp, struct comp_buffer *buffer, int
 	else
 		comp_dbg(comp, "disconnect buffer %d as source", buf_get_id(buffer));
 
-	PPL_LOCK();
+	PPL_LOCK(buffer->core);
 
 	comp_list = comp_buffer_list(comp, dir);
 	buffer_detach(buffer, comp_list, dir);
 	buffer_set_comp(buffer, NULL, dir);
 
-	PPL_UNLOCK();
+	PPL_UNLOCK(buffer->core);
 }
 
 /* pipelines must be inactive */

src/audio/pipeline/pipeline-stream.c

@@ -21,6 +21,7 @@
 #include <rtos/kernel.h>
 #include <sof/audio/module_adapter/module/generic.h>
 #include <sof/lib/cpu-clk-manager.h>
+#include <sof/schedule/ll_schedule_domain.h>
 
 #ifdef CONFIG_IPC_MAJOR_4
 #include <ipc4/notification.h>
@@ -145,6 +146,19 @@ static int pipeline_comp_copy(struct comp_dev *current,
 	return err;
 }
 
+#ifdef CONFIG_SOF_USERSPACE_LL
+/*
+ * User-space LL: The LL thread runs at cooperative priority (-16)
+ * and cannot be preempted by IPC (priority 1). No per-pipeline
+ * lock needed in pipeline_copy() hot path.
+ */
+#define PPL_LOCK(x) user_ll_lock_sched(x)
+#define PPL_UNLOCK(x) user_ll_unlock_sched(x)
+#else
+#define PPL_LOCK(x)
+#define PPL_UNLOCK(x)
+#endif
+
 /* Copy data across all pipeline components.
  * For capture pipelines it always starts from source component
  * and continues downstream and for playback pipelines it first
@@ -162,6 +176,8 @@ int pipeline_copy(struct pipeline *p)
 	uint32_t dir;
 	int ret;
 
+	PPL_LOCK(p->core);
+
 	if (p->source_comp->direction == SOF_IPC_STREAM_PLAYBACK) {
 		dir = PPL_DIR_UPSTREAM;
 		start = p->sink_comp;
@@ -178,6 +194,8 @@ int pipeline_copy(struct pipeline *p)
 		pipe_err(p, "ret = %d, start->comp.id = %u, dir = %u",
 			 ret, dev_comp_id(start), dir);
 
+	PPL_UNLOCK(p->core);
+
 	return ret;
 }
 

src/include/sof/audio/component.h

@@ -680,10 +680,6 @@ struct comp_dev {
 	struct list_item bsource_list;	/**< list of source buffers */
 	struct list_item bsink_list;	/**< list of sink buffers */
 
-#ifdef CONFIG_SOF_USERSPACE_LL
-	struct sys_mutex list_mutex;     /**< protect lists of source/sinks */
-#endif
-
 	/* performance data*/
 	struct comp_perf_data perf_data;
 	/* Input Buffer Size for pin 0, add array for other pins if needed */
@@ -868,9 +864,6 @@ static inline void comp_init(const struct comp_driver *drv,
 	dev->state = COMP_STATE_INIT;
 	list_init(&dev->bsink_list);
 	list_init(&dev->bsource_list);
-#ifdef CONFIG_SOF_USERSPACE_LL
-	sys_mutex_init(&dev->list_mutex);
-#endif
 #ifndef __ZEPHYR__
 	memcpy_s(&dev->tctx, sizeof(dev->tctx),
 		 trace_comp_drv_get_tr_ctx(dev->drv), sizeof(struct tr_ctx));

src/include/sof/schedule/ll_schedule_domain.h

@@ -120,6 +120,8 @@ struct task *zephyr_ll_task_alloc(void);
 struct k_heap *zephyr_ll_user_heap(void);
 bool zephyr_ll_user_heap_verify(struct k_heap *heap);
 void zephyr_ll_user_resources_init(void);
+void user_ll_lock_sched(int core);
+void user_ll_unlock_sched(int core);
 #endif /* CONFIG_SOF_USERSPACE_LL */
 
 static inline struct ll_schedule_domain *domain_init

src/ipc/ipc-helper.c

@@ -17,6 +17,7 @@
 #include <sof/ipc/msg.h>
 #include <sof/ipc/driver.h>
 #include <sof/ipc/schedule.h>
+#include <sof/schedule/ll_schedule_domain.h>
 #include <rtos/alloc.h>
 #include <rtos/cache.h>
 #include <sof/lib/cpu.h>
@@ -336,7 +337,15 @@ __cold int ipc_comp_free(struct ipc *ipc, uint32_t comp_id)
 		return -EINVAL;
 	}
 
+	/* Lock buffer lists to prevent racing with the LL scheduler.
+	 * In user-space builds, use the LL scheduler's sys_mutex
+	 * (re-entrant, so safe if caller already holds it).
+	 */
+#ifdef CONFIG_SOF_USERSPACE_LL
+	user_ll_lock_sched(icd->core);
+#else
 	irq_local_disable(flags);
+#endif
 	comp_dev_for_each_producer_safe(icd->cd, buffer, safe) {
 		comp_buffer_set_sink_component(buffer, NULL);
 		/* This breaks the list, but we anyway delete all buffers */
@@ -349,7 +358,11 @@ __cold int ipc_comp_free(struct ipc *ipc, uint32_t comp_id)
 		comp_buffer_reset_source_list(buffer);
 	}
 
+#ifdef CONFIG_SOF_USERSPACE_LL
+	user_ll_unlock_sched(icd->core);
+#else
 	irq_local_enable(flags);
+#endif
 
 	/*
 	 * A completed pipeline stores raw comp_dev pointers in its

src/ipc/ipc4/helper.c

@@ -448,10 +448,21 @@ __cold static int ipc_pipeline_module_free(uint32_t pipeline_id)
 	struct ipc *ipc = ipc_get();
 	struct ipc_comp_dev *icd;
 	int ret;
+#ifdef CONFIG_SOF_USERSPACE_LL
+	int ppl_core;
+#endif
 
 	assert_can_be_cold();
 
 	icd = ipc_get_comp_by_ppl_id(ipc, COMP_TYPE_COMPONENT, pipeline_id, IPC_COMP_ALL);
+	if (!icd)
+		return IPC4_SUCCESS;
+
+#ifdef CONFIG_SOF_USERSPACE_LL
+	ppl_core = icd->core;
+	user_ll_lock_sched(ppl_core);
+#endif
+
 	while (icd) {
 		struct comp_buffer *buffer;
 		struct comp_buffer *safe;
@@ -481,12 +492,20 @@ __cold static int ipc_pipeline_module_free(uint32_t pipeline_id)
 		else
 			ret = ipc_comp_free(ipc, icd->id);
 
-		if (ret)
+		if (ret) {
+#ifdef CONFIG_SOF_USERSPACE_LL
+			user_ll_unlock_sched(ppl_core);
+#endif
 			return IPC4_INVALID_RESOURCE_STATE;
+		}
 
 		icd = ipc_get_comp_by_ppl_id(ipc, COMP_TYPE_COMPONENT, pipeline_id, IPC_COMP_ALL);
 	}
 
+#ifdef CONFIG_SOF_USERSPACE_LL
+	user_ll_unlock_sched(ppl_core);
+#endif
+
 	return IPC4_SUCCESS;
 }
 
@@ -560,21 +579,25 @@ __cold static struct comp_buffer *ipc4_create_buffer(struct comp_dev *src, bool
  * disable any interrupts.
  */
 
-#define ll_block(cross_core_bind, flags) \
+#if CONFIG_SOF_USERSPACE_LL
+#error "CONFIG_SOF_USERSPACE_LL not compatible with cross-core streams"
+#else
+#define ll_block(src_core, dst_core, flags) \
 	do { \
-		if (cross_core_bind) \
+		if (src_core != dst_core) \
 			domain_block(sof_get()->platform_timer_domain); \
 		else \
 			irq_local_disable(flags); \
 	} while (0)
 
-#define ll_unblock(cross_core_bind, flags) \
+#define ll_unblock(src_core, dst_core, flags) \
 	do { \
-		if (cross_core_bind) \
+		if (src_core != dst_core) \
 			domain_unblock(sof_get()->platform_timer_domain); \
 		else \
 			irq_local_enable(flags); \
 	} while (0)
+#endif
 
 /* Calling both ll_block() and ll_wait_finished_on_core() makes sure LL will not start its
  * next cycle and its current cycle on specified core has finished.
@@ -606,8 +629,15 @@ static int ll_wait_finished_on_core(struct comp_dev *dev)
 
 #else
 
-#define ll_block(cross_core_bind, flags)	irq_local_disable(flags)
-#define ll_unblock(cross_core_bind, flags)	irq_local_enable(flags)
+#if CONFIG_SOF_USERSPACE_LL
+/* note: cross-core streams are disabled in the build in this branch,
+ * so we can safely use 'src_core' only */
+#define ll_block(src_core, dst_core, flags)	do { user_ll_lock_sched(src_core); (void)flags; } while(0)
+#define ll_unblock(src_core, dst_core, flags)	do { user_ll_unlock_sched(src_core) ; (void)flags; } while(0)
+#else
+#define ll_block(src_core, dst_core, flags)	irq_local_disable(flags)
+#define ll_unblock(src_core, dst_core, flags)	irq_local_enable(flags)
+#endif
 
 #endif
 
@@ -794,7 +824,7 @@ __cold int ipc_comp_connect(struct ipc *ipc, ipc_pipe_comp_connect *_connect)
 	 * blocked on corresponding core(s) to prevent IPC or IDC task getting preempted which
 	 * could result in buffers being only half connected when a pipeline task gets executed.
 	 */
-	ll_block(cross_core_bind, flags);
+	ll_block(source->ipc_config.core, sink->ipc_config.core, flags);
 
 	if (cross_core_bind) {
 #if CONFIG_CROSS_CORE_STREAM
@@ -851,7 +881,7 @@ __cold int ipc_comp_connect(struct ipc *ipc, ipc_pipe_comp_connect *_connect)
 		source->direction_set = true;
 	}
 
-	ll_unblock(cross_core_bind, flags);
+	ll_unblock(source->ipc_config.core, sink->ipc_config.core, flags);
 
 	return IPC4_SUCCESS;
 
@@ -865,7 +895,7 @@ __cold int ipc_comp_connect(struct ipc *ipc, ipc_pipe_comp_connect *_connect)
 e_sink_connect:
 	pipeline_disconnect(source, buffer, PPL_CONN_DIR_COMP_TO_BUFFER);
 free:
-	ll_unblock(cross_core_bind, flags);
+	ll_unblock(source->ipc_config.core, sink->ipc_config.core, flags);
 	buffer_free(buffer);
 	return IPC4_INVALID_RESOURCE_STATE;
 }
@@ -938,24 +968,24 @@ __cold int ipc_comp_disconnect(struct ipc *ipc, ipc_pipe_comp_connect *_connect)
 	 * IPC or IDC task getting preempted which could result in buffers being only half connected
 	 * when a pipeline task gets executed.
 	 */
-	ll_block(cross_core_unbind, flags);
+	ll_block(src->ipc_config.core, sink->ipc_config.core, flags);
 
 	if (cross_core_unbind) {
 #if CONFIG_CROSS_CORE_STREAM
 		/* Make sure LL has finished on both cores */
 		if (!cpu_is_me(src->ipc_config.core))
 			if (ll_wait_finished_on_core(src) < 0) {
-				ll_unblock(cross_core_unbind, flags);
+				ll_unblock(src->ipc_config.core, sink->ipc_config.core, flags);
 				return IPC4_FAILURE;
 			}
 		if (!cpu_is_me(sink->ipc_config.core))
 			if (ll_wait_finished_on_core(sink) < 0) {
-				ll_unblock(cross_core_unbind, flags);
+				ll_unblock(src->ipc_config.core, sink->ipc_config.core, flags);
 				return IPC4_FAILURE;
 			}
 #else
 		tr_err(&ipc_tr, "Cross-core binding is disabled");
-		ll_unblock(cross_core_unbind, flags);
+		ll_unblock(src->ipc_config.core, sink->ipc_config.core, flags);
 		return IPC4_FAILURE;
 #endif
 	}
@@ -972,7 +1002,7 @@ __cold int ipc_comp_disconnect(struct ipc *ipc, ipc_pipe_comp_connect *_connect)
 	unbind_data.source = audio_buffer_get_source(&buffer->audio_buffer);
 	ret1 = comp_unbind(sink, &unbind_data);
 
-	ll_unblock(cross_core_unbind, flags);
+	ll_unblock(src->ipc_config.core, sink->ipc_config.core, flags);
 
 	buffer_free(buffer);