module: mux: rework module to use sink/source api

Rework the mux module to only use the sink/source api to
prepare the SOF for the full transition to pipeline 2.0.

Signed-off-by: Piotr Hoppe <piotr.hoppe@intel.com>

Author: piotrhoppeintel

src/audio/buffers/comp_buffer.c

@@ -100,6 +100,20 @@ static int comp_buffer_commit_buffer(struct sof_sink *sink, size_t commit_size)
 	return 0;
 }
 
+static int comp_buffer_sink_get_state(struct sof_sink *sink)
+{
+	struct comp_buffer *buffer = comp_buffer_get_from_sink(sink);
+
+	return comp_get_state(comp_buffer_get_sink_component(buffer));
+}
+
+static int comp_buffer_source_get_state(struct sof_source *source)
+{
+	struct comp_buffer *buffer = comp_buffer_get_from_source(source);
+
+	return comp_get_state(comp_buffer_get_source_component(buffer));
+}
+
 static int comp_buffer_set_ipc_params(struct sof_audio_buffer *audio_buffer,
 				      struct sof_ipc_stream_params *params,
 				      bool force_update)
@@ -179,6 +193,7 @@ APP_TASK_DATA static const struct source_ops comp_buffer_source_ops = {
 	.audio_set_ipc_params = audio_buffer_source_set_ipc_params,
 	.on_audio_format_set = audio_buffer_source_on_audio_format_set,
 	.set_alignment_constants = audio_buffer_source_set_alignment_constants,
+	.get_state = comp_buffer_source_get_state,
 };
 
 APP_TASK_DATA static const struct sink_ops comp_buffer_sink_ops = {
@@ -189,6 +204,7 @@ APP_TASK_DATA static const struct sink_ops comp_buffer_sink_ops = {
 	.on_audio_format_set = audio_buffer_sink_on_audio_format_set,
 	.set_alignment_constants = audio_buffer_sink_set_alignment_constants,
 	.get_lft = audio_buffer_sink_get_lft,
+	.get_state = comp_buffer_sink_get_state,
 };
 
 static const struct audio_buffer_ops audio_buffer_ops = {

src/audio/mux/mux.c

@@ -156,10 +156,10 @@ static int get_stream_index(struct comp_dev *dev, struct comp_data *cd, uint32_t
 }
 
 static void mux_prepare_active_look_up(struct comp_data *cd,
-				       struct audio_stream *sink,
-				       const struct audio_stream **sources)
+				       struct sof_sink *sink,
+				       struct sof_source **sources)
 {
-	const struct audio_stream *source;
+	struct sof_source *source;
 	int elem;
 	int active_elem = 0;
 
@@ -169,9 +169,10 @@ static void mux_prepare_active_look_up(struct comp_data *cd,
 		if (!source)
 			continue;
 
-		if (cd->lookup[0].copy_elem[elem].in_ch >= audio_stream_get_channels(source) ||
-		    cd->lookup[0].copy_elem[elem].out_ch >= audio_stream_get_channels(sink))
+		if (cd->lookup[0].copy_elem[elem].in_ch >= source_get_channels(source) ||
+		    cd->lookup[0].copy_elem[elem].out_ch >= sink_get_channels(sink)) {
 			continue;
+		}
 
 		cd->active_lookup.copy_elem[active_elem] = cd->lookup[0].copy_elem[elem];
 		active_elem++;
@@ -203,18 +204,19 @@ static struct mux_look_up *get_lookup_table(struct comp_dev *dev, struct comp_da
 }
 
 static void demux_prepare_active_look_up(struct comp_data *cd,
-					 struct audio_stream *sink,
-					 const struct audio_stream *source,
+					 struct sof_sink *sink,
+					 struct sof_source *source,
 					 struct mux_look_up *look_up)
 {
 	int elem;
 	int active_elem = 0;
 
 	/* init pointers */
 	for (elem = 0; elem < look_up->num_elems; elem++) {
-		if (look_up->copy_elem[elem].in_ch >= audio_stream_get_channels(source) ||
-		    look_up->copy_elem[elem].out_ch >= audio_stream_get_channels(sink))
+		if (look_up->copy_elem[elem].in_ch >= source_get_channels(source) ||
+		    look_up->copy_elem[elem].out_ch >= sink_get_channels(sink)) {
 			continue;
+		}
 
 		cd->active_lookup.copy_elem[active_elem] = look_up->copy_elem[elem];
 		active_elem++;
@@ -225,56 +227,89 @@ static void demux_prepare_active_look_up(struct comp_data *cd,
 
 /* process and copy stream data from source to sink buffers */
 static int demux_process(struct processing_module *mod,
-			 struct input_stream_buffer *input_buffers, int num_input_buffers,
-			 struct output_stream_buffer *output_buffers, int num_output_buffers)
+			 struct sof_source **sources, int num_of_sources,
+			 struct sof_sink **sinks, int num_of_sinks)
 {
 	struct comp_data *cd = module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
-	struct comp_buffer *sink;
-	struct audio_stream *sinks_stream[MUX_MAX_STREAMS] = { NULL };
-	struct mux_look_up *look_ups[MUX_MAX_STREAMS] = { NULL };
-	int frames;
-	int sink_bytes;
-	int source_bytes;
+	struct sof_source *source;
+	const void *source_data;
+	const void *source_start;
+	size_t source_size;
+	size_t source_bytes;
+	uint32_t frames;
+	int ret;
 	int i;
 
 	comp_dbg(dev, "entry");
 
-	/* align sink streams with their respective configurations */
-	comp_dev_for_each_consumer(dev, sink) {
-		if (comp_buffer_get_sink_state(sink) == dev->state) {
-			i = get_stream_index(dev, cd, buffer_pipeline_id(sink));
-			/* return if index wrong */
-			if (i < 0) {
-				return i;
-			}
+	if (sources == NULL && sources[0] == NULL) {
+		return 0;
+	}
+	source = sources[0];
 
-			look_ups[i] = get_lookup_table(dev, cd, buffer_pipeline_id(sink));
-			sinks_stream[i] = &sink->stream;
+	/* if there are no sinks active, then there is nothing to do */
+	if (num_of_sinks == 0) {
+		return 0;
+	}
+
+	/* the same number of frames is distributed to every sink, so it is
+	 * limited by both the source availability and every active sink's free
+	 * space
+	 */
+	frames = source_get_data_frames_available(source);
+	for (i = 0; i < num_of_sinks; i++) {
+		if (sink_get_state(sinks[i]) != dev->state) {
+			comp_warn(dev, "the sink %d has diffrent state, culd not process", i);
+			continue;
 		}
+
+		frames = MIN(frames, sink_get_free_frames(sinks[i]));
 	}
 
-	/* if there are no sinks active, then sinks[] is also empty */
-	if (num_output_buffers == 0)
+	if (!frames) {
 		return 0;
+	}
 
-	frames = input_buffers[0].size;
-	source_bytes = frames * audio_stream_frame_bytes(mod->input_buffers[0].data);
-	sink_bytes = frames * audio_stream_frame_bytes(mod->output_buffers[0].data);
+	/* the source is read-only and shared by all sinks, so it is obtained
+	 * once here and released once after all sinks have been served
+	 */
+	source_bytes = frames * source_get_frame_bytes(source);
+	ret = source_get_data(source, source_bytes, &source_data, &source_start,
+			      &source_size);
+	if (ret) {
+		return ret;
+	}
 
 	/* produce output, one sink at a time */
-	for (i = 0; i < num_output_buffers; i++) {
-		if (sinks_stream[i]) {
-			demux_prepare_active_look_up(cd, sinks_stream[i],
-						     input_buffers[0].data, look_ups[i]);
-			cd->demux(dev, sinks_stream[i], input_buffers[0].data,
-				  frames, &cd->active_lookup);
+	for (i = 0; i < num_of_sinks; i++) {
+		struct sof_sink *sink = sinks[i];
+		uint32_t pipeline_id = sink_get_pipeline_id(sink);
+		struct mux_look_up *look_up;
+
+		/* skip sinks that are not in the same state as the component */
+		if (sink_get_state(sink) != dev->state) {
+			continue;
+		}
+
+		/* return if configuration for this pipeline is missing */
+		if (get_stream_index(dev, cd, pipeline_id) < 0) {
+			source_release_data(source, 0);
+			return -EINVAL;
+		}
+
+		look_up = get_lookup_table(dev, cd, pipeline_id);
+		demux_prepare_active_look_up(cd, sink, source, look_up);
+		ret = cd->demux(dev, sink, source, source_data, source_start,
+				source_size, frames, &cd->active_lookup);
+		if (ret) {
+			source_release_data(source, 0);
+			return ret;
 		}
-		mod->output_buffers[i].size = sink_bytes;
 	}
 
-	/* Update consumed */
-	mod->input_buffers[0].consumed = source_bytes;
+	/* consume the processed data from the source */
+	source_release_data(source, source_bytes);
 	return 0;
 }
 
@@ -305,58 +340,96 @@ static int demux_trigger(struct processing_module *mod, int cmd)
 
 /* process and copy stream data from source to sink buffers */
 static int mux_process(struct processing_module *mod,
-		       struct input_stream_buffer *input_buffers, int num_input_buffers,
-		       struct output_stream_buffer *output_buffers, int num_output_buffers)
+		       struct sof_source **sources, int num_of_sources,
+		       struct sof_sink **sinks, int num_of_sinks)
 {
 	struct comp_data *cd = module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
-	struct comp_buffer *source;
-	const struct audio_stream *sources_stream[MUX_MAX_STREAMS] = { NULL };
-	int frames = 0;
-	int sink_bytes;
-	int i, j;
+	struct sof_sink *sink = sinks[0];
+	struct sof_source *active_sources[MUX_MAX_STREAMS] = { NULL };
+	struct cir_buf_source source_bufs[MUX_MAX_STREAMS] = { 0 };
+	size_t source_bytes[MUX_MAX_STREAMS] = { 0 };
+	struct cir_buf_sink sink_buf;
+	size_t sink_bytes, size;
+	uint32_t frames;
+	int i, idx, ret;
 
 	comp_dbg(dev, "entry");
 
-	/* align source streams with their respective configurations */
-	j = 0;
-	comp_dev_for_each_producer(dev, source) {
-		if (comp_buffer_get_source_state(source) == dev->state) {
-			if (frames)
-				frames = MIN(frames, input_buffers[j].size);
-			else
-				frames = input_buffers[j].size;
-
-			i = get_stream_index(dev, cd, buffer_pipeline_id(source));
-			/* return if index wrong */
-			if (i < 0) {
-				return i;
-			}
+	/* if there are no sources or sinks active, then there is nothing to do */
+	if (num_of_sinks == 0 || num_of_sources == 0) {
+		return 0;
+	}
+
+	/* the same number of frames is taken from every active source and
+	 * written to the single sink, so it is limited by both the sink free
+	 * space and every active source's availability
+	 */
+	frames = sink_get_free_frames(sink);
+	for (i = 0; i < num_of_sources; i++) {
+		struct sof_source *source = sources[i];
 
-			sources_stream[i] = &source->stream;
+		/* skip sources that are not in the same state as the component */
+		if (source_get_comp_state(source) != dev->state) {
+			continue;
 		}
-		j++;
+
+		/* map the source to its configured stream index */
+		idx = get_stream_index(dev, cd, source_get_pipeline_id(source));
+		if (idx < 0) {
+			return idx;
+		}
+
+		active_sources[idx] = source;
+		frames = MIN(frames, source_get_data_frames_available(source));
 	}
 
-	/* check if there are any sources active */
-	if (num_input_buffers == 0)
+	if (!frames) {
 		return 0;
+	}
 
-	sink_bytes = frames * audio_stream_frame_bytes(mod->output_buffers[0].data);
-	mux_prepare_active_look_up(cd, output_buffers[0].data, &sources_stream[0]);
+	/* build the active routing table for the current connections */
+	mux_prepare_active_look_up(cd, sink, active_sources);
 
-	/* produce output */
-	cd->mux(dev, output_buffers[0].data, &sources_stream[0], frames, &cd->active_lookup);
-
-	/* Update consumed per source using each source's own frame size */
-	j = 0;
-	comp_dev_for_each_producer(dev, source) {
-		if (comp_buffer_get_source_state(source) == dev->state)
-			mod->input_buffers[j].consumed =
-				frames * audio_stream_frame_bytes(mod->input_buffers[j].data);
-		j++;
+	/* acquire the single sink circular buffer for the whole period */
+	sink_bytes = frames * sink_get_frame_bytes(sink);
+	ret = sink_get_buffer(sink, sink_bytes, &sink_buf.ptr, &sink_buf.buf_start, &size);
+	if (ret) {
+		return ret;
 	}
-	mod->output_buffers[0].size = sink_bytes;
+	sink_buf.buf_end = (char *)sink_buf.buf_start + size;
+
+	/* acquire every active source circular buffer */
+	for (i = 0; i < MUX_MAX_STREAMS; i++) {
+		if (!active_sources[i]) {
+			continue;
+		}
+
+		source_bytes[i] = frames * source_get_frame_bytes(active_sources[i]);
+		ret = source_get_data(active_sources[i], source_bytes[i], &source_bufs[i].ptr,
+				      &source_bufs[i].buf_start, &size);
+		if (ret) {
+			/* release the sources acquired so far and the sink */
+			for (--i; i >= 0; i--)
+				if (active_sources[i]) {
+					source_release_data(active_sources[i], 0);
+				}
+			sink_commit_buffer(sink, 0);
+			return ret;
+		}
+		source_bufs[i].buf_end = (const char *)source_bufs[i].buf_start + size;
+	}
+
+	/* produce output */
+	cd->mux(dev, sink, &sink_buf, active_sources, source_bufs, frames, &cd->active_lookup);
+
+	/* commit the produced data and consume from every active source */
+	sink_commit_buffer(sink, sink_bytes);
+	for (i = 0; i < MUX_MAX_STREAMS; i++)
+		if (active_sources[i]) {
+			source_release_data(active_sources[i], source_bytes[i]);
+		}
+
 	return 0;
 }
 
@@ -443,7 +516,7 @@ static const struct module_interface mux_interface = {
 	.set_configuration = mux_set_config,
 	.get_configuration = mux_get_config,
 	.prepare = mux_prepare,
-	.process_audio_stream = mux_process,
+	.process = mux_process,
 	.reset = mux_reset,
 	.free = mux_free,
 };
@@ -454,7 +527,7 @@ static const struct module_interface demux_interface = {
 	.set_configuration = mux_set_config,
 	.get_configuration = mux_get_config,
 	.prepare = mux_prepare,
-	.process_audio_stream = demux_process,
+	.process = demux_process,
 	.trigger = demux_trigger,
 	.reset = mux_reset,
 	.free = mux_free,

src/audio/mux/mux.h

@@ -28,6 +28,10 @@
 #endif
 struct comp_buffer;
 struct comp_dev;
+struct sof_source;
+struct sof_sink;
+struct cir_buf_source;
+struct cir_buf_sink;
 
 /** \brief Supported streams count. */
 #if CONFIG_IPC_MAJOR_3
@@ -66,11 +70,13 @@ struct mux_stream_data {
 	uint8_t reserved2[3]; // padding to ensure proper alignment of following instances
 } __attribute__((packed, aligned(4)));
 
-typedef void(*demux_func)(struct comp_dev *dev, struct audio_stream *sink,
-			  const struct audio_stream *source, uint32_t frames,
-			  struct mux_look_up *look_up);
-typedef void(*mux_func)(struct comp_dev *dev, struct audio_stream *sink,
-			const struct audio_stream **sources, uint32_t frames,
+typedef int(*demux_func)(struct comp_dev *dev, struct sof_sink *sink,
+			  struct sof_source *source, const void *source_data,
+			  const void *source_start, size_t source_size,
+			  uint32_t frames, struct mux_look_up *look_up);
+typedef void(*mux_func)(struct comp_dev *dev, struct sof_sink *sink,
+			struct cir_buf_sink *sink_buf, struct sof_source **sources,
+			struct cir_buf_source *source_bufs, uint32_t frames,
 			struct mux_look_up *look_up);
 
 /**