pipeline: eos: Add support for End Of Stream state

src/audio/component.c

@@ -5,6 +5,7 @@
 // Author: Liam Girdwood <liam.r.girdwood@linux.intel.com>
 
 #include <sof/audio/component_ext.h>
+#include <sof/audio/sink_source_utils.h>
 #include <sof/common.h>
 #include <sof/debug/telemetry/performance_monitor.h>
 #include <rtos/panic.h>
@@ -496,6 +497,62 @@ void audio_stream_copy_to_linear(const struct audio_stream *source, int ioffset,
 	}
 }
 
+static bool comp_check_eos(struct comp_dev *dev)
+{
+	enum sof_audio_buffer_state sink_state = AUDIOBUF_STATE_INITIAL;
+	struct comp_buffer *buffer;
+
+	if (!dev->pipeline->expect_eos)
+		return false;
+
+	comp_dev_for_each_producer(dev, buffer) {
+		struct sof_source *source = audio_buffer_get_source(&buffer->audio_buffer);
+		enum sof_audio_buffer_state state = source_get_state(source);
+
+		if (source_get_pipeline_id(source) != dev->pipeline->pipeline_id)
+			continue;
+
+		if (state == AUDIOBUF_STATE_END_OF_STREAM_FLUSH) {
+			/* Earlier in the pipeline, there is a DP module that has reached
+			 * the EOS state. However, silence is generated to flush its internal
+			 * buffers, so pass this state to the output buffers.
+			 */
+			comp_dbg(dev, "comp_check_eos() - EOS flush detected");
+			sink_state = AUDIOBUF_STATE_END_OF_STREAM_FLUSH;
+			break;
+		} else if (state == AUDIOBUF_STATE_END_OF_STREAM) {
+			/* EOS is detected, so we need to set the sink state to AUDIOBUF_STATE_EOS. */
+			size_t min_avail = source_get_min_available(source);
+
+			if (source_get_data_available(source) < min_avail) {
+				comp_dbg(dev, "comp_check_eos() - EOS detected");
+				if (dev->ipc_config.proc_domain == COMP_PROCESSING_DOMAIN_DP) {
+					/* For DP modules, fill missing input data with silence to
+					 * allow it to process the remaining data.
+					 */
+					struct sof_sink *previous_mod_data_sink =
+						audio_buffer_get_sink(&buffer->audio_buffer);
+					sink_fill_with_silence(previous_mod_data_sink, min_avail);
+					sink_state = AUDIOBUF_STATE_END_OF_STREAM_FLUSH;
+				} else {
+					sink_state = AUDIOBUF_STATE_END_OF_STREAM;
+					break;
+				}
+			}
+		}
+	}
+
+	if (sink_state != AUDIOBUF_STATE_INITIAL) {
+		comp_dev_for_each_consumer(dev, buffer)
+			audio_buffer_set_state(&buffer->audio_buffer, sink_state);
+
+		/* For AUDIOBUF_STATE_END_OF_STREAM_FLUSH process data normally. */
+		return sink_state != AUDIOBUF_STATE_END_OF_STREAM_FLUSH;
+	}
+
+	return false;
+}
+
 /** See comp_ops::copy */
 int comp_copy(struct comp_dev *dev)
 {
@@ -530,6 +587,9 @@ int comp_copy(struct comp_dev *dev)
 		const uint32_t begin_stamp = (uint32_t)telemetry_timestamp();
 #endif
 
+		if (comp_check_eos(dev))
+			return 0;
+
 		ret = dev->drv->ops.copy(dev);
 
 #ifdef CONFIG_SOF_TELEMETRY_PERFORMANCE_MEASUREMENTS

src/audio/host-zephyr.c

@@ -393,6 +393,32 @@ static int host_get_status(struct comp_dev *dev, struct host_data *hd, struct dm
 /* Minimum time between 2 consecutive "no bytes to copy" messages in milliseconds */
 #define SOF_MIN_NO_BYTES_INTERVAL_MS 20
 
+static inline bool host_handle_eos(struct host_data *hd, struct comp_dev *dev,
+				   uint32_t avail_samples)
+{
+	struct sof_audio_buffer *buffer = &hd->local_buffer->audio_buffer;
+	enum sof_audio_buffer_state state = audio_buffer_get_state(buffer);
+
+	if (!dev->pipeline->expect_eos)
+		return false;
+
+	if (!avail_samples) {
+		/* EOS is detected, so we need to set the sink
+		 * state to AUDIOBUF_STATE_END_OF_STREAM.
+		 */
+		if (state != AUDIOBUF_STATE_END_OF_STREAM) {
+			audio_buffer_set_eos(buffer);
+			comp_info(dev, "host_handle_eos() - EOS detected");
+		}
+		return true;
+	}
+
+	if (state == AUDIOBUF_STATE_END_OF_STREAM)
+		comp_warn(dev, "Data available after reporting end of stream!");
+
+	return false;
+}
+
 /**
  * Calculates bytes to be copied in normal mode.
  * @param dev Host component device.
@@ -440,6 +466,9 @@ static uint32_t host_get_copy_bytes_normal(struct host_data *hd, struct comp_dev
 	if (dev->direction == SOF_IPC_STREAM_PLAYBACK) {
 		avail_samples = (dma_stat.pending_length - hd->partial_size) / dma_sample_bytes;
 		free_samples = audio_stream_get_free_samples(&buffer->stream);
+
+		if (host_handle_eos(hd, dev, avail_samples))
+			return 0;
 	} else {
 		avail_samples = audio_stream_get_avail_samples(&buffer->stream);
 		free_samples = (dma_stat.free - hd->partial_size) / dma_sample_bytes;

src/audio/mixin_mixout/mixin_mixout.c

@@ -79,6 +79,8 @@ struct mixin_data {
 #if CONFIG_XRUN_NOTIFICATIONS_ENABLE
 	uint32_t last_reported_underrun;
 	uint32_t underrun_notification_period;
+	uint32_t eos_delay_periods;
+	bool eos_delay_configured;
 #endif
 };
 
@@ -248,23 +250,40 @@ static void silence(struct cir_buf_ptr *stream, uint32_t start_offset,
 
 #if CONFIG_XRUN_NOTIFICATIONS_ENABLE
 static void mixin_check_notify_underrun(struct comp_dev *dev, struct mixin_data *mixin_data,
+					enum sof_audio_buffer_state state,
 					size_t source_avail, size_t sinks_free)
 {
+	const bool eos_detected = state == AUDIOBUF_STATE_END_OF_STREAM_FLUSH ||
+				  state == AUDIOBUF_STATE_END_OF_STREAM;
+
 	struct ipc_msg *notify;
 
 	mixin_data->last_reported_underrun++;
 
-	if (!source_avail && mixin_data->last_reported_underrun >=
-	    mixin_data->underrun_notification_period) {
-		mixin_data->last_reported_underrun = 0;
+	if (!source_avail || eos_detected) {
+		if (eos_detected) {
+			if (mixin_data->eos_delay_configured) {
+				mixin_data->eos_delay_periods--;
+			} else {
+				pipeline_get_dai_comp_latency(dev->pipeline->pipeline_id,
+							      &mixin_data->eos_delay_periods);
+				mixin_data->eos_delay_configured = true;
+			}
+		}
+
+		if ((!eos_detected && mixin_data->last_reported_underrun >=
+		    mixin_data->underrun_notification_period) ||
+		    (eos_detected && mixin_data->eos_delay_periods == 0)) {
+			mixin_data->last_reported_underrun = 0;
 
-		notify = ipc_notification_pool_get(IPC4_RESOURCE_EVENT_SIZE);
-		if (!notify)
-			return;
+			notify = ipc_notification_pool_get(IPC4_RESOURCE_EVENT_SIZE);
+			if (!notify)
+				return;
 
-		mixer_underrun_notif_msg_init(notify, dev->ipc_config.id, false,
-					      source_avail, sinks_free);
-		ipc_msg_send(notify, notify->tx_data, false);
+			mixer_underrun_notif_msg_init(notify, dev->ipc_config.id, eos_detected,
+						      source_avail, sinks_free);
+			ipc_msg_send(notify, notify->tx_data, false);
+		}
 	}
 }
 #endif
@@ -294,9 +313,10 @@ static int mixin_process(struct processing_module *mod,
 	uint32_t source_avail_frames, sinks_free_frames;
 	struct processing_module *active_mixouts[MIXIN_MAX_SINKS];
 	uint16_t sinks_ids[MIXIN_MAX_SINKS];
+	struct pending_frames *pending_frames;
 	uint32_t bytes_to_consume = 0;
 	uint32_t frames_to_copy;
-	struct pending_frames *pending_frames;
+	size_t frame_bytes;
 	int i, ret;
 	struct cir_buf_ptr source_ptr;
 
@@ -389,10 +409,10 @@ static int mixin_process(struct processing_module *mod,
 		return 0;
 
 #if CONFIG_XRUN_NOTIFICATIONS_ENABLE
-	size_t frame_bytes = source_get_frame_bytes(sources[0]);
-	size_t min_frames = MIN(dev->frames, sinks_free_frames);
+	frame_bytes = source_get_frame_bytes(sources[0]);
+	const size_t min_frames = MIN(dev->frames, sinks_free_frames);
 
-	mixin_check_notify_underrun(dev, mixin_data,
+	mixin_check_notify_underrun(dev, mixin_data, source_get_state(sources[0]),
 				    source_avail_frames * frame_bytes,
 				    min_frames * frame_bytes);
 #endif
@@ -461,7 +481,7 @@ static int mixin_process(struct processing_module *mod,
 		 * silence instead of that source data
 		 */
 		if (source_avail_frames == 0) {
-			uint32_t frame_bytes = sink_get_frame_bytes(mixout_mod->sinks[0]);
+			frame_bytes = sink_get_frame_bytes(mixout_mod->sinks[0]);
 
 			/* generate silence */
 			silence(&mixout_data->acquired_buf, start_frame * frame_bytes,
@@ -698,6 +718,7 @@ static int mixin_prepare(struct processing_module *mod,
 	int ret;
 
 	comp_info(dev, "mixin_prepare()");
+	md->eos_delay_configured = false;
 
 	ret = mixin_params(mod);
 	if (ret < 0)

src/audio/pipeline/pipeline-graph.c

@@ -10,6 +10,7 @@
 #include <sof/audio/pipeline.h>
 #include <sof/ipc/msg.h>
 #include <rtos/interrupt.h>
+#include <rtos/symbol.h>
 #include <sof/lib/mm_heap.h>
 #include <sof/lib/uuid.h>
 #include <sof/compiler_attributes.h>
@@ -551,4 +552,6 @@ struct comp_dev *pipeline_get_dai_comp_latency(uint32_t pipeline_id, uint32_t *l
 
 	return NULL;
 }
+EXPORT_SYMBOL(pipeline_get_dai_comp_latency);
+
 #endif

src/include/module/audio/audio_stream.h

@@ -13,6 +13,24 @@
 #include <stdbool.h>
 #include "../ipc/stream.h"
 
+
+/**
+ * @enum sof_audio_buffer_state
+ * @brief Define states of an audio stream buffer connecting two components.
+ *
+ * This enum represents the lifecycle of an audio stream, including its
+ * initialization, readiness, and end-of-stream handling. It is used to
+ * track and manage the state transitions of the stream during audio processing.
+ */
+enum sof_audio_buffer_state {
+	AUDIOBUF_STATE_INITIAL,			/* Initial state, hw params not configured. */
+	AUDIOBUF_STATE_READY,			/* Stream ready, hw params configured */
+	AUDIOBUF_STATE_END_OF_STREAM,		/* Detected End Of Stream */
+	AUDIOBUF_STATE_END_OF_STREAM_FLUSH,	/* Detected End Of Stream, generating silence
+						 * to flush buffers in dp modules.
+						 */
+};
+
 /**
  * set of parameters describing audio stream
  * this structure is shared between audio_stream.h and sink/source interface
@@ -53,7 +71,7 @@ struct sof_audio_stream_params {
 
 	uint16_t chmap[SOF_IPC_MAX_CHANNELS];	/**< channel map - SOF_CHMAP_ */
 
-	bool hw_params_configured; /**< indicates whether hw params were set */
+	enum sof_audio_buffer_state state;	/**< audio stream state */
 };
 
 #endif /* __MODULE_AUDIO_AUDIO_STREAM_H__ */

src/include/module/audio/source_api.h

@@ -323,4 +323,9 @@ static inline struct processing_module *source_get_bound_module(struct sof_sourc
 	return source->bound_module;
 }
 
+static inline enum sof_audio_buffer_state source_get_state(const struct sof_source *source)
+{
+	return source->audio_stream_params->state;
+}
+
 #endif /* __MODULE_AUDIO_SOURCE_API_H__ */

src/include/sof/audio/audio_buffer.h

@@ -214,17 +214,34 @@ static inline bool audio_buffer_is_shared(struct sof_audio_buffer *buffer)
 
 static inline bool audio_buffer_hw_params_configured(struct sof_audio_buffer *buffer)
 {
-	return buffer->audio_stream_params->hw_params_configured;
+	return buffer->audio_stream_params->state != AUDIOBUF_STATE_INITIAL;
 }
 
 static inline void audio_buffer_set_hw_params_configured(struct sof_audio_buffer *buffer)
 {
-	buffer->audio_stream_params->hw_params_configured = true;
+	buffer->audio_stream_params->state = AUDIOBUF_STATE_READY;
 }
 
 static inline void audio_buffer_reset_params(struct sof_audio_buffer *buffer)
 {
-	buffer->audio_stream_params->hw_params_configured = false;
+	buffer->audio_stream_params->state = AUDIOBUF_STATE_INITIAL;
+}
+
+static inline enum sof_audio_buffer_state audio_buffer_get_state(
+	const struct sof_audio_buffer *buffer)
+{
+	return buffer->audio_stream_params->state;
+}
+
+static inline void audio_buffer_set_state(struct sof_audio_buffer *buffer,
+					  enum sof_audio_buffer_state state)
+{
+	buffer->audio_stream_params->state = state;
+}
+
+static inline void audio_buffer_set_eos(struct sof_audio_buffer *buffer)
+{
+	buffer->audio_stream_params->state = AUDIOBUF_STATE_END_OF_STREAM;
 }
 
 static inline uint16_t audio_buffer_get_chmap(struct sof_audio_buffer *buffer, size_t index)

src/include/sof/audio/pipeline.h

@@ -68,6 +68,7 @@ struct pipeline {
 	int32_t xrun_bytes;		/* last xrun length */
 	uint32_t status;		/* pipeline status */
 	struct tr_ctx tctx;		/* trace settings */
+	bool expect_eos;		/* pipeline is expecting end of stream */
 
 	/* scheduling */
 	struct task *pipe_task;		/* pipeline processing task */