Data processing scheduler - SOF tasks as Zephyr preemptive threads

app/boards/intel_adsp_ace15_mtpm.conf

@@ -30,6 +30,7 @@ CONFIG_DAI_INTEL_DMIC_NHLT=y
 CONFIG_DAI_DMIC_HAS_OWNERSHIP=y
 CONFIG_DAI_DMIC_HAS_MULTIPLE_LINE_SYNC=y
 CONFIG_DAI_INTEL_SSP=y
+CONFIG_ZEPHYR_DP_SCHEDULER=y
 CONFIG_DMA=y
 CONFIG_DMA_INTEL_ADSP_GPDMA=y
 CONFIG_DMA_DW_LLI_POOL_SIZE=50

src/audio/component.c

@@ -302,8 +302,9 @@ int comp_copy(struct comp_dev *dev)
 
 	assert(dev->drv->ops.copy);
 
-	/* copy only if we are the owner of the component */
-	if (cpu_is_me(dev->ipc_config.core)) {
+	/* copy only if we are the owner of the LL component */
+	if (dev->ipc_config.proc_domain == COMP_PROCESSING_DOMAIN_LL &&
+	    cpu_is_me(dev->ipc_config.core)) {
 #if CONFIG_PERFORMANCE_COUNTERS
 		perf_cnt_init(&dev->pcd);
 #endif

src/audio/pipeline/pipeline-params.c

@@ -317,7 +317,33 @@ static int pipeline_comp_prepare(struct comp_dev *current,
 		}
 	}
 
-	err = pipeline_comp_task_init(current->pipeline);
+	switch (current->ipc_config.proc_domain) {
+	case COMP_PROCESSING_DOMAIN_LL:
+		/* this is a LL scheduled module */
+		err = pipeline_comp_ll_task_init(current->pipeline);
+		break;
+
+#if CONFIG_ZEPHYR_DP_SCHEDULER
+	case COMP_PROCESSING_DOMAIN_DP:
+		/* this is a DP scheduled module */
+
+		/*
+		 * workaround - because of some issues with cache, currently we can allow DP
+		 * modules to run on the same core as LL pipeline only.
+		 * to be removed once buffering is fixed
+		 */
+		if (current->pipeline->core != current->ipc_config.core)
+			err = -EINVAL;
+		else
+			err = pipeline_comp_dp_task_init(current);
+
+		break;
+#endif /* CONFIG_ZEPHYR_DP_SCHEDULER */
+
+	default:
+		err = -EINVAL;
+	}
+
 	if (err < 0)
 		return err;
 

src/audio/pipeline/pipeline-schedule.c

@@ -12,6 +12,7 @@
 #include <sof/lib/agent.h>
 #include <sof/list.h>
 #include <sof/schedule/ll_schedule.h>
+#include <sof/schedule/dp_schedule.h>
 #include <sof/schedule/schedule.h>
 #include <rtos/task.h>
 #include <rtos/spinlock.h>
@@ -31,6 +32,25 @@ LOG_MODULE_DECLARE(pipe, CONFIG_SOF_LOG_LEVEL);
 DECLARE_SOF_UUID("pipe-task", pipe_task_uuid, 0xf11818eb, 0xe92e, 0x4082,
 		 0x82,  0xa3, 0xdc, 0x54, 0xc6, 0x04, 0xeb, 0xb3);
 
+#if CONFIG_ZEPHYR_DP_SCHEDULER
+
+/* ee755917-96b9-4130-b49e-37b9d0501993 */
+DECLARE_SOF_UUID("dp-task", dp_task_uuid, 0xee755917, 0x96b9, 0x4130,
+		 0xb4,  0x9e, 0x37, 0xb9, 0xd0, 0x50, 0x19, 0x93);
+
+/**
+ * current static stack size for each DP component
+ * TODO: to be taken from module manifest
+ */
+#define TASK_DP_STACK_SIZE 8192
+
+/**
+ * \brief a priority of the DP threads in the system.
+ */
+#define ZEPHYR_DP_THREAD_PRIORITY (CONFIG_NUM_PREEMPT_PRIORITIES - 1)
+
+#endif /* CONFIG_ZEPHYR_DP_SCHEDULER */
+
 static void pipeline_schedule_cancel(struct pipeline *p)
 {
 	schedule_task_cancel(p->pipe_task);
@@ -333,7 +353,7 @@ void pipeline_schedule_triggered(struct pipeline_walk_context *ctx,
 	irq_local_enable(flags);
 }
 
-int pipeline_comp_task_init(struct pipeline *p)
+int pipeline_comp_ll_task_init(struct pipeline *p)
 {
 	uint32_t type;
 
@@ -355,6 +375,40 @@ int pipeline_comp_task_init(struct pipeline *p)
 	return 0;
 }
 
+#if CONFIG_ZEPHYR_DP_SCHEDULER
+static enum task_state dp_task_run(void *data)
+{
+	struct comp_dev *comp = data;
+
+	comp->drv->ops.copy(comp);
+	return SOF_TASK_STATE_RESCHEDULE;
+}
+
+int pipeline_comp_dp_task_init(struct comp_dev *comp)
+{
+	int ret;
+	struct task_ops ops  = {
+		.run		= dp_task_run,
+		.get_deadline	= NULL,
+		.complete	= NULL
+	};
+
+	if (!comp->task) {
+		ret = scheduler_dp_task_init(&comp->task,
+					     SOF_UUID(dp_task_uuid),
+					     &ops,
+					     comp,
+					     comp->ipc_config.core,
+					     TASK_DP_STACK_SIZE,
+					     ZEPHYR_DP_THREAD_PRIORITY);
+		if (ret < 0)
+			return ret;
+	}
+
+	return 0;
+}
+#endif /* CONFIG_ZEPHYR_DP_SCHEDULER */
+
 void pipeline_comp_trigger_sched_comp(struct pipeline *p,
 				      struct comp_dev *comp,
 				      struct pipeline_walk_context *ctx)

src/include/sof/audio/component.h

@@ -535,13 +535,17 @@ struct comp_driver_info {
 	struct list_item list;		/**< list of component drivers */
 };
 
+#define COMP_PROCESSING_DOMAIN_LL 0
+#define COMP_PROCESSING_DOMAIN_DP 1
+
 /**
  * Audio component base configuration from IPC at creation.
  */
 struct comp_ipc_config {
 	uint32_t core;		/**< core we run on */
 	uint32_t id;		/**< component id */
 	uint32_t pipeline_id;	/**< component pipeline id */
+	uint32_t proc_domain;   /**< processing domain - LL or DP */
 	enum sof_comp_type type;	/**< component type */
 	uint32_t periods_sink;	/**< 0 means variable */
 	uint32_t periods_source;/**< 0 means variable */
@@ -569,9 +573,10 @@ struct comp_dev {
 				     *  to run component's processing
 				     */
 
-	struct task *task;	/**< component's processing task used only
-				  *  for components running on different core
-				  *  than the rest of the pipeline
+	struct task *task;	/**< component's processing task used
+				  *  1) for components running on different core
+				  *    than the rest of the pipeline
+				  *  2) for all DP tasks
 				  */
 	uint32_t size;		/**< component's allocated size */
 	uint32_t period;	/**< component's processing period */

src/include/sof/audio/component_ext.h

@@ -50,8 +50,9 @@ static inline void comp_free(struct comp_dev *dev)
 {
 	assert(dev->drv->ops.free);
 
-	/* free task if shared component */
-	if (dev->is_shared && dev->task) {
+	/* free task if shared component or DP task*/
+	if ((dev->is_shared || dev->ipc_config.proc_domain == COMP_PROCESSING_DOMAIN_DP) &&
+	    dev->task) {
 		schedule_task_free(dev->task);
 		rfree(dev->task);
 	}
@@ -147,13 +148,43 @@ static inline int comp_trigger_remote(struct comp_dev *dev, int cmd)
 	return idc_send_msg(&msg, IDC_BLOCKING);
 }
 
+static inline int comp_trigger_local(struct comp_dev *dev, int cmd)
+{
+	int ret;
+
+	ret = dev->drv->ops.trigger(dev, cmd);
+
+	/* start a thread in case of shared component or DP scheduling */
+	if (dev->task) {
+		/* schedule or cancel task */
+		switch (cmd) {
+		case COMP_TRIGGER_START:
+		case COMP_TRIGGER_RELEASE:
+			schedule_task(dev->task, 0, dev->period);
+			break;
+		case COMP_TRIGGER_XRUN:
+		case COMP_TRIGGER_PAUSE:
+		case COMP_TRIGGER_STOP:
+			schedule_task_cancel(dev->task);
+			break;
+		}
+	}
+
+	return ret;
+}
+
 /** See comp_ops::trigger */
 static inline int comp_trigger(struct comp_dev *dev, int cmd)
 {
+	int ret;
 	assert(dev->drv->ops.trigger);
 
-	return (dev->is_shared && !cpu_is_me(dev->ipc_config.core)) ?
-		comp_trigger_remote(dev, cmd) : dev->drv->ops.trigger(dev, cmd);
+	if (dev->is_shared && !cpu_is_me(dev->ipc_config.core))
+		ret = comp_trigger_remote(dev, cmd);
+	else
+		ret = comp_trigger_local(dev, cmd);
+
+	return ret;
 }
 
 /** Runs comp_ops::prepare on the target component's core */

src/include/sof/audio/pipeline.h

@@ -340,11 +340,18 @@ static inline bool pipeline_is_this_cpu(struct pipeline *p)
 }
 
 /**
- * \brief Free's a pipeline.
+ * \brief Init an LL task for a pipeline.
  * \param[in] p pipeline.
  * \return 0 on success.
  */
-int pipeline_comp_task_init(struct pipeline *p);
+int pipeline_comp_ll_task_init(struct pipeline *p);
+
+/**
+ * \brief Init a DP task for a component
+ * \param[in] comp a component the task is created for
+ * \return 0 on success.
+ */
+int pipeline_comp_dp_task_init(struct comp_dev *comp);
 
 /**
  * \brief Free's a pipeline.

src/include/sof/schedule/dp_schedule.h

@@ -0,0 +1,94 @@
+/* SPDX-License-Identifier: BSD-3-Clause */
+/*
+ * Copyright(c) 2023 Intel Corporation. All rights reserved.
+ *
+ * Author: Marcin Szkudlinski
+ */
+
+#ifndef __SOF_SCHEDULE_DP_SCHEDULE_H__
+#define __SOF_SCHEDULE_DP_SCHEDULE_H__
+
+#include <rtos/task.h>
+#include <sof/trace/trace.h>
+#include <user/trace.h>
+#include <stdint.h>
+
+/**
+ *
+ * DP scheduler is a scheduler that creates a separate preemptible Zephyr thread for each SOF task
+ *   There's only one instance of DP in the system, however, threads can be assigned and pinned
+ *   to any core in the system for its execution, there's no SMP processing.
+ *
+ * The task execution may be delayed and task may be re-scheduled periodically
+ * NOTE: delayed start and rescheduling takes place in sync with LL scheduler, meaning the
+ *   DP scheduler is triggered as the last task of LL running on a primary core.
+ *   That implies a limitation: LL scheduler MUST be running on primary core in order to have
+ *   this feature working.
+ *   It is fine, because rescheduling is a feature used for data processing when a pipeline is
+ *   running.
+ *
+ * Other possible usage of DP scheduler is to schedule task with DP_SCHEDULER_RUN_TASK_IMMEDIATELY
+ *   as start parameter. It will force the task to work without any delays and async to LL.
+ *   This kind of scheduling may be used for staring regular zephyr tasks using SOF api
+ *
+ * Task run() may return:
+ *  SOF_TASK_STATE_RESCHEDULE - the task will be rescheduled as specified in scheduler period
+ *				note that task won't ever be rescheduled if LL is not running
+ *  SOF_TASK_STATE_COMPLETED  - the task will be removed from scheduling,
+ *				calling schedule_task will add the task to processing again
+ *				task_complete() will be called
+ *  SOF_TASK_STATE_CANCEL     - the task will be removed from scheduling,
+ *				calling schedule_task will add the task to processing again
+ *				task_complete() won't be called
+ *  other statuses - assert will go off
+ *
+ * NOTE: task - means a SOF task
+ *	 thread - means a Zephyr preemptible thread
+ *
+ * TODO  - EDF:
+ * Threads run on the same priority, lower than thread running LL tasks. Zephyr EDF mechanism
+ * is used for decision which thread/task is to be scheduled next. The DP scheduler calculates
+ * the task deadline and set it in Zephyr thread properties, the final scheduling decision is made
+ * by Zephyr.
+ *
+ * Each time tick the scheduler iterates through the list of all active tasks and calculates
+ * a deadline based on
+ *  - knowledge how the modules are bound
+ *  - declared time required by a task to complete processing
+ *  - the deadline of the last module
+ *
+ */
+
+/** \brief tell the scheduler to run the task immediately, even if LL tick is not yet running */
+#define SCHEDULER_DP_RUN_TASK_IMMEDIATELY ((uint64_t)-1)
+
+/**
+ * \brief Init the Data Processing scheduler
+ */
+int scheduler_dp_init(void);
+
+/**
+ * \brief Set the Data Processing scheduler to be accessible at secondary cores
+ */
+int scheduler_dp_init_secondary_core(void);
+
+/**
+ * \brief initialize a DP task and add it to scheduling
+ *
+ * \param[out] task pointer, pointer to allocated task structure will be return
+ * \param[in] uid pointer to UUID of the task
+ * \param[in] ops pointer to task functions
+ * \param[in] data pointer to the thread private data
+ * \param[in] core CPU the thread should run on
+ * \param[in] stack_size size of stack for a zephyr task
+ * \param[in] task_priority priority of the zephyr task
+ */
+int scheduler_dp_task_init(struct task **task,
+			   const struct sof_uuid_entry *uid,
+			   const struct task_ops *ops,
+			   void *data,
+			   uint16_t core,
+			   size_t stack_size,
+			   uint32_t task_priority);
+
+#endif /* __SOF_SCHEDULE_DP_SCHEDULE_H__ */

src/include/sof/schedule/schedule.h

@@ -31,14 +31,19 @@ enum {
 	SOF_SCHEDULE_LL_DMA,	/**< Low latency DMA, schedules immediately
 				  *  on scheduling component's DMA interrupt
 				  */
+	SOF_SCHEDULE_DP,	/**< DataProcessing scheduler
+				  *  Scheduler based on Zephyr peemptive threads
+				  *  TODO: DP will become the Zephyr EDF scheduler type
+				  *  and will be unified with SOF_SCHEDULE_EDF for Zephyr builds
+				  *  current implementation of Zephyr based EDF is depreciated now
+				  */
 	SOF_SCHEDULE_COUNT	/**< indicates number of scheduler types */
 };
 
 /** \brief Scheduler free available flags */
 #define SOF_SCHEDULER_FREE_IRQ_ONLY	BIT(0) /**< Free function disables only
 						 *  interrupts
 						 */
-
 /**
  * Scheduler operations.
  *

src/init/init.c

@@ -26,6 +26,7 @@
 #include <rtos/idc.h>
 #include <sof/schedule/schedule.h>
 #include <sof/schedule/edf_schedule.h>
+#include <sof/schedule/dp_schedule.h>
 #include <sof/schedule/ll_schedule.h>
 #include <sof/schedule/ll_schedule_domain.h>
 #include <ipc/trace.h>
@@ -183,6 +184,12 @@ int secondary_core_init(struct sof *sof)
 	if (dma_domain)
 		scheduler_init_ll(dma_domain);
 
+#if CONFIG_ZEPHYR_DP_SCHEDULER
+	err = scheduler_dp_init_secondary_core();
+	if (err < 0)
+		return err;
+#endif /* CONFIG_ZEPHYR_DP_SCHEDULER */
+
 	/* initialize IDC mechanism */
 	trace_point(TRACE_BOOT_PLATFORM_IDC);
 	err = idc_init();

src/ipc/ipc3/helper.c

@@ -163,6 +163,7 @@ static void comp_common_builder(struct sof_ipc_comp *comp,
 	config->core = comp->core;
 	config->id = comp->id;
 	config->pipeline_id = comp->pipeline_id;
+	config->proc_domain = COMP_PROCESSING_DOMAIN_LL;
 	config->type = comp->type;
 
 	/* buffers dont have the following data */