ipc4: pipeline: split trigger and copy processing

posix/include/rtos/task.h

@@ -58,7 +58,7 @@ struct task {
 	uint64_t start;		/**< start time in [ms] since now (LL only) */
 	const struct sof_uuid_entry *uid; /**< Uuid */
 	uint16_t type;		/**< type of the task (LL or EDF) */
-	uint16_t priority;	/**< priority of the task (used by LL) */
+	int16_t priority;	/**< priority of the task (used by LL); lower runs first */
 	uint16_t core;		/**< execution core */
 	uint16_t flags;		/**< custom flags */
 	struct schedule_data *sch;	/**< scheduler bound to task */

src/audio/pipeline/pipeline-graph.c

@@ -260,12 +260,7 @@ int pipeline_free(struct pipeline *p)
 	 */
 
 	/* remove from any scheduling */
-	if (p->pipe_task) {
-#if !CONFIG_LIBRARY || UNIT_TEST
-		schedule_task_free(p->pipe_task);
-#endif
-		sof_heap_free(p->heap, p->pipe_task);
-	}
+	pipeline_comp_ll_task_free(p);
 
 	ipc_msg_free(p->msg);
 

src/audio/pipeline/pipeline-schedule.c

@@ -20,6 +20,7 @@
 #include <rtos/task.h>
 #include <rtos/spinlock.h>
 #include <rtos/string.h>
+#include <rtos/userspace_helper.h>
 #include <ipc/header.h>
 #include <ipc/stream.h>
 #include <ipc/topology.h>
@@ -34,6 +35,16 @@ LOG_MODULE_DECLARE(pipe, CONFIG_SOF_LOG_LEVEL);
 
 SOF_DEFINE_REG_UUID(pipe_task);
 
+#ifdef CONFIG_IPC_MAJOR_4
+SOF_DEFINE_REG_UUID(pipe_trigger_task);
+
+/*
+ * Track a delayed trigger so later trigger tasks cannot overtake
+ * it and change the pipeline order requested by the host.
+ */
+static APP_SYSUSER_BSS struct pipeline *delayed_trigger_owner[CONFIG_CORE_COUNT];
+#endif
+
 #if CONFIG_ZEPHYR_DP_SCHEDULER
 
 SOF_DEFINE_REG_UUID(dp_task);
@@ -160,7 +171,9 @@ static enum task_state pipeline_task_cmd(struct pipeline *p,
 	return err;
 }
 
-static enum task_state pipeline_task(void *arg)
+#ifndef CONFIG_IPC_MAJOR_4
+/** Run IPC3 pipeline trigger and copy processing. */
+static enum task_state ipc3_pipeline_task(void *arg)
 {
 	struct sof_ipc_reply reply = {
 		.hdr.cmd = SOF_IPC_GLB_REPLY,
@@ -239,6 +252,120 @@ static enum task_state pipeline_task(void *arg)
 	return SOF_TASK_STATE_RESCHEDULE;
 }
 
+#else /* CONFIG_IPC_MAJOR_4 */
+
+/** Run an IPC4 pipeline trigger task. */
+static enum task_state ipc4_pipeline_trigger_task(void *arg)
+{
+	struct sof_ipc_reply reply = {
+		.hdr.cmd = SOF_IPC_GLB_REPLY,
+		.hdr.size = sizeof(reply),
+	};
+	struct pipeline *p = arg;
+	enum task_state state;
+	struct pipeline *owner = delayed_trigger_owner[p->core];
+
+	/* some other trigger task is still waiting for its delayed trigger */
+	if (owner && owner != p)
+		return SOF_TASK_STATE_RESCHEDULE;
+
+	if (p->xrun_bytes) {
+		delayed_trigger_owner[p->core] = NULL;
+
+		if (p->trigger.cmd != COMP_TRIGGER_NO_ACTION) {
+			p->trigger.cmd = COMP_TRIGGER_NO_ACTION;
+			reply.error = -EPIPE;
+			ipc_msg_reply(&reply);
+		}
+
+		return SOF_TASK_STATE_COMPLETED;
+	}
+
+	if (p->trigger.delay) {
+		delayed_trigger_owner[p->core] = p;
+		p->trigger.delay--;
+		return SOF_TASK_STATE_RESCHEDULE;
+	}
+
+	if (p->trigger.cmd == COMP_TRIGGER_NO_ACTION) {
+		delayed_trigger_owner[p->core] = NULL;
+		return SOF_TASK_STATE_COMPLETED;
+	}
+
+	state = pipeline_task_cmd(p, &reply);
+	if (state == SOF_TASK_STATE_RESCHEDULE && p->trigger.delay) {
+		delayed_trigger_owner[p->core] = p;
+		return state;
+	}
+
+	delayed_trigger_owner[p->core] = NULL;
+
+	/* RUNNING means that the independent copy task should keep running. */
+	return state == SOF_TASK_STATE_RUNNING ? SOF_TASK_STATE_COMPLETED : state;
+}
+
+/** Run IPC4 pipeline copy processing. */
+static enum task_state ipc4_pipeline_copy_task(void *arg)
+{
+	struct pipeline *p = arg;
+	int err;
+
+	pipe_dbg(p, "entry");
+
+	/* are we in xrun ? */
+	if (p->xrun_bytes) {
+		/* try to recover */
+		err = pipeline_xrun_recover(p);
+		if (err < 0)
+			/* skip copy if still in xrun */
+			return SOF_TASK_STATE_COMPLETED;
+	}
+
+	/* Do not copy until the pipeline trigger sequence has completed. */
+	if (task_is_active(p->trigger_task))
+		return SOF_TASK_STATE_RESCHEDULE;
+
+	if (p->status == COMP_STATE_PAUSED)
+		return SOF_TASK_STATE_COMPLETED;
+
+	err = pipeline_copy(p);
+	if (err < 0) {
+		/* try to recover */
+		err = pipeline_xrun_recover(p);
+		if (err < 0) {
+			pipe_err(p, "xrun recovery failed! pipeline is stopped.");
+			/* failed - host will stop this pipeline */
+			return SOF_TASK_STATE_COMPLETED;
+		}
+	}
+
+	pipe_dbg(p, "sched");
+
+	return SOF_TASK_STATE_RESCHEDULE;
+}
+
+/** Allocate and initialize an IPC4 pipeline trigger task. */
+static struct task *ipc4_pipeline_trigger_task_init(struct pipeline *p, uint32_t type)
+{
+	struct task *task;
+
+	task = sof_heap_alloc(p->heap, SOF_MEM_FLAG_USER, sizeof(*task), 0);
+	if (!task)
+		return NULL;
+
+	memset(task, 0, sizeof(*task));
+
+	/* All trigger tasks use the highest priority, regardless of pipeline priority. */
+	if (schedule_task_init_ll(task, SOF_UUID(pipe_trigger_task_uuid), type, -1,
+				  ipc4_pipeline_trigger_task, p, p->core, 0) < 0) {
+		sof_heap_free(p->heap, task);
+		return NULL;
+	}
+
+	return task;
+}
+#endif /* CONFIG_IPC_MAJOR_4 */
+
 static struct task *pipeline_task_init(struct pipeline *p, uint32_t type)
 {
 	struct pipeline_task *task = NULL;
@@ -251,7 +378,12 @@ static struct task *pipeline_task_init(struct pipeline *p, uint32_t type)
 	memset(task, 0, sizeof(*task));
 
 	if (schedule_task_init_ll(&task->task, SOF_UUID(pipe_task_uuid), type,
-				  p->priority, pipeline_task,
+				  p->priority,
+#ifdef CONFIG_IPC_MAJOR_4
+				  ipc4_pipeline_copy_task,
+#else
+				  ipc3_pipeline_task,
+#endif
 				  p, p->core, 0) < 0) {
 		sof_heap_free(p->heap, task);
 		return NULL;
@@ -325,6 +457,10 @@ void pipeline_schedule_triggered(struct pipeline_walk_context *ctx,
 				p->trigger.pending = true;
 				p->trigger.host = ppl_data->start;
 				ppl_data->start = NULL;
+#ifdef CONFIG_IPC_MAJOR_4
+				if (schedule_task(p->trigger_task, 0, 0) < 0)
+					pipe_err(p, "failed to schedule trigger task");
+#endif
 			} else {
 				pipeline_schedule_cancel(p);
 				p->status = COMP_STATE_PAUSED;
@@ -346,6 +482,10 @@ void pipeline_schedule_triggered(struct pipeline_walk_context *ctx,
 				p->trigger.pending = true;
 				p->trigger.host = ppl_data->start;
 				ppl_data->start = NULL;
+#ifdef CONFIG_IPC_MAJOR_4
+				if (schedule_task(p->trigger_task, 0, 0) < 0)
+					pipe_err(p, "failed to schedule trigger task");
+#endif
 			} else {
 				p->status = COMP_STATE_ACTIVE;
 			}
@@ -374,14 +514,24 @@ int pipeline_comp_ll_task_init(struct pipeline *p)
 {
 	uint32_t type;
 
+	/* right now we always consider pipeline as a low latency
+	 * component, but it may change in the future
+	 */
+	type = pipeline_is_timer_driven(p) ? SOF_SCHEDULE_LL_TIMER :
+		SOF_SCHEDULE_LL_DMA;
+
+#ifdef CONFIG_IPC_MAJOR_4
+	if (!p->trigger_task) {
+		p->trigger_task = ipc4_pipeline_trigger_task_init(p, type);
+		if (!p->trigger_task) {
+			pipe_err(p, "trigger task init failed");
+			return -ENOMEM;
+		}
+	}
+#endif
+
 	/* initialize task if necessary */
 	if (!p->pipe_task) {
-		/* right now we always consider pipeline as a low latency
-		 * component, but it may change in the future
-		 */
-		type = pipeline_is_timer_driven(p) ? SOF_SCHEDULE_LL_TIMER :
-			SOF_SCHEDULE_LL_DMA;
-
 		p->pipe_task = pipeline_task_init(p, type);
 		if (!p->pipe_task) {
 			pipe_err(p, "task init failed");
@@ -392,6 +542,37 @@ int pipeline_comp_ll_task_init(struct pipeline *p)
 	return 0;
 }
 
+/** Free pipeline LL tasks and release delayed-trigger ownership. */
+void pipeline_comp_ll_task_free(struct pipeline *p)
+{
+#ifdef CONFIG_IPC_MAJOR_4
+	if (p->trigger_task) {
+#if !CONFIG_LIBRARY || UNIT_TEST
+		schedule_task_free(p->trigger_task);
+#endif
+	}
+
+	/*
+	 * Since the task has now been removed from the scheduler, a possible race
+	 * with LL between the check and store below cannot cause any harm: other
+	 * tasks do nothing when they see another owner. Therefore, LL locking can
+	 * be skipped here.
+	 */
+	if (delayed_trigger_owner[p->core] == p)
+		delayed_trigger_owner[p->core] = NULL;
+
+	if (p->trigger_task)
+		sof_heap_free(p->heap, p->trigger_task);
+#endif
+
+	if (p->pipe_task) {
+#if !CONFIG_LIBRARY || UNIT_TEST
+		schedule_task_free(p->pipe_task);
+#endif
+		sof_heap_free(p->heap, p->pipe_task);
+	}
+}
+
 #if CONFIG_ZEPHYR_DP_SCHEDULER
 static enum task_state dp_task_run(void *data)
 {

src/include/sof/audio/pipeline.h

@@ -73,7 +73,10 @@ struct pipeline {
 	bool expect_eos;		/* pipeline is expecting end of stream */
 
 	/* scheduling */
-	struct task *pipe_task;		/* pipeline processing task */
+#ifdef CONFIG_IPC_MAJOR_4
+	struct task *trigger_task;	/* IPC4 pipeline trigger task */
+#endif
+	struct task *pipe_task;		/* IPC3 pipeline processing task or IPC4 pipeline copy task */
 	struct pipeline *sched_next;	/* pipeline scheduled after this */
 	struct pipeline *sched_prev;	/* pipeline scheduled before this */
 
@@ -361,6 +364,12 @@ static inline bool pipeline_is_this_cpu(struct pipeline *p)
  */
 int pipeline_comp_ll_task_init(struct pipeline *p);
 
+/**
+ * \brief Free the LL tasks owned by a pipeline.
+ * \param[in] p pipeline.
+ */
+void pipeline_comp_ll_task_free(struct pipeline *p);
+
 /**
  * \brief Init a DP task for a component
  * \param[in] comp a component the task is created for

src/include/sof/schedule/ll_schedule.h

@@ -40,14 +40,14 @@ int scheduler_init_ll(struct ll_schedule_domain *domain);
 
 int schedule_task_init_ll(struct task *task,
 			  const struct sof_uuid_entry *uid, uint16_t type,
-			  uint16_t priority, enum task_state (*run)(void *data),
+			  int16_t priority, enum task_state (*run)(void *data),
 			  void *data, uint16_t core, uint32_t flags);
 #else
 int zephyr_ll_scheduler_init(struct ll_schedule_domain *domain);
 
 int zephyr_ll_task_init(struct task *task,
 			const struct sof_uuid_entry *uid, uint16_t type,
-			uint16_t priority, enum task_state (*run)(void *data),
+			int16_t priority, enum task_state (*run)(void *data),
 			void *data, uint16_t core, uint32_t flags);
 
 #define scheduler_init_ll zephyr_ll_scheduler_init

src/include/sof/schedule/schedule.h

@@ -359,7 +359,7 @@ static inline struct k_thread *scheduler_init_context(struct task *task)
  */
 int schedule_task_init(struct task *task,
 		       const struct sof_uuid_entry *uid, uint16_t type,
-		       uint16_t priority, enum task_state (*run)(void *data),
+		       int16_t priority, enum task_state (*run)(void *data),
 		       void *data, uint16_t core, uint32_t flags);
 
 /**

src/ipc/ipc4/handler-user.c

@@ -556,17 +556,6 @@ int ipc4_set_pipeline_state(struct ipc4_message_request *ipc4)
 			ipc_compound_pre_start(state.primary.r.type);
 			ret = ipc4_pipeline_trigger(ppl_icd, cmd, &delayed);
 			ipc_compound_post_start(state.primary.r.type, ret, delayed);
-			if (delayed) {
-				/* To maintain pipeline order for triggers, we must
-				 * do a blocking wait until trigger is processed.
-				 * This will add a max delay of 'ppl_count' LL ticks
-				 * to process the full trigger list.
-				 */
-				if (ipc_wait_for_compound_msg() != 0) {
-					ipc_cmd_err(&ipc_tr, "ipc4: fail with delayed trigger");
-					return IPC4_FAILURE;
-				}
-			}
 		}
 
 		if (ret != 0)

src/ipc/ipc4/helper.c

@@ -529,6 +529,12 @@ __cold int ipc_pipeline_new(struct ipc *ipc, ipc_pipe_new *_pipe_desc)
 
 	tr_dbg(&ipc_tr, "ipc: pipeline id = %u", (uint32_t)pipe_desc->primary.r.instance_id);
 
+	if (pipe_desc->extension.r.core_id >= CONFIG_CORE_COUNT) {
+		tr_err(&ipc_tr, "invalid pipeline core ID: %u",
+		       (uint32_t)pipe_desc->extension.r.core_id);
+		return IPC4_INVALID_CORE_ID;
+	}
+
 	/* pass IPC to target core */
 	if (!cpu_is_me(pipe_desc->extension.r.core_id))
 		return ipc4_process_on_core(pipe_desc->extension.r.core_id, false);

src/platform/library/include/platform/lib/ll_schedule.h

@@ -19,7 +19,7 @@ int scheduler_init_ll(struct ll_schedule_domain *domain);
 
 int schedule_task_init_ll(struct task *task,
 			  const struct sof_uuid_entry *uid, uint16_t type,
-			  uint16_t priority, enum task_state (*run)(void *data),
+			  int16_t priority, enum task_state (*run)(void *data),
 			  void *data, uint16_t core, uint32_t flags);
 
 #endif /* __LIBRARY_INCLUDE_LIB_SCHEDULE_H__ */

src/platform/library/schedule/ll_schedule.c

@@ -105,7 +105,7 @@ static struct scheduler_ops schedule_ll_ops = {
 
 int schedule_task_init_ll(struct task *task,
 			  const struct sof_uuid_entry *uid, uint16_t type,
-			  uint16_t priority, enum task_state (*run)(void *data),
+			  int16_t priority, enum task_state (*run)(void *data),
 			  void *data, uint16_t core, uint32_t flags)
 {
 	return schedule_task_init(task, uid, SOF_SCHEDULE_LL_TIMER, 0, run,

src/platform/library/schedule/schedule.c

@@ -22,7 +22,7 @@ struct schedulers **arch_schedulers_get(void)
 
 int schedule_task_init(struct task *task,
 		       const struct sof_uuid_entry *uid, uint16_t type,
-		       uint16_t priority, enum task_state (*run)(void *data),
+		       int16_t priority, enum task_state (*run)(void *data),
 		       void *data, uint16_t core, uint32_t flags)
 {
 	struct schedulers *schedulers = *arch_schedulers_get();