diff --git a/src/audio/aria/aria.c b/src/audio/aria/aria.c
index 7f3d3e5e10f8..2c703fc277bc 100644
--- a/src/audio/aria/aria.c
+++ b/src/audio/aria/aria.c
@@ -178,10 +178,10 @@ static int aria_prepare(struct processing_module *mod,
 
 	comp_info(dev, "aria_prepare()");
 
-	source = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	source = comp_dev_get_first_data_producer(dev);
 	aria_set_stream_params(source, mod);
 
-	sink = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sink = comp_dev_get_first_data_consumer(dev);
 	aria_set_stream_params(sink, mod);
 
 	if (audio_stream_get_valid_fmt(&source->stream) != SOF_IPC_FRAME_S24_4LE ||
diff --git a/src/audio/asrc/asrc.c b/src/audio/asrc/asrc.c
index b67ac3e8518d..8300a52250c8 100644
--- a/src/audio/asrc/asrc.c
+++ b/src/audio/asrc/asrc.c
@@ -396,10 +396,8 @@ static int asrc_params(struct processing_module *mod)
 		return -EINVAL;
 	}
 
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer,
-				  sink_list);
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer,
-				source_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* update the source/sink buffer formats. Sink rate will be modified below */
 	asrc_update_buffer_format(sourceb, cd);
@@ -452,7 +450,7 @@ static int asrc_dai_find(struct comp_dev *dev, struct comp_data *cd)
 	if (cd->mode == ASRC_OM_PUSH) {
 		/* In push mode check if sink component is DAI */
 		do {
-			sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+			sinkb = comp_dev_get_first_data_consumer(dev);
 
 			dev = sinkb->sink;
 
@@ -470,7 +468,7 @@ static int asrc_dai_find(struct comp_dev *dev, struct comp_data *cd)
 	} else {
 		/* In pull mode check if source component is DAI */
 		do {
-			sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+			sourceb = comp_dev_get_first_data_producer(dev);
 
 			dev = sourceb->source;
 
@@ -545,10 +543,8 @@ static int asrc_prepare(struct processing_module *mod,
 		return ret;
 
 	/* SRC component will only ever have 1 source and 1 sink buffer */
-	sourceb = list_first_item(&dev->bsource_list,
-				  struct comp_buffer, sink_list);
-	sinkb = list_first_item(&dev->bsink_list,
-				struct comp_buffer, source_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* get source data format and period bytes */
 	cd->source_format = audio_stream_get_frm_fmt(&sourceb->stream);
@@ -796,10 +792,8 @@ static int asrc_process(struct processing_module *mod,
 		return ret;
 
 	/* asrc component needs 1 source and 1 sink buffer */
-	source = list_first_item(&dev->bsource_list, struct comp_buffer,
-				 sink_list);
-	sink = list_first_item(&dev->bsink_list, struct comp_buffer,
-			       source_list);
+	source = comp_dev_get_first_data_producer(dev);
+	sink = comp_dev_get_first_data_consumer(dev);
 
 	frames_src = audio_stream_get_avail_frames(source_s);
 	frames_snk = audio_stream_get_free_frames(sink_s);
diff --git a/src/audio/codec/dts/dts.c b/src/audio/codec/dts/dts.c
index c6ef230d9be5..39fad1409e62 100644
--- a/src/audio/codec/dts/dts.c
+++ b/src/audio/codec/dts/dts.c
@@ -76,8 +76,7 @@ static int dts_effect_convert_sof_interface_result(struct comp_dev *dev,
 static int dts_effect_populate_buffer_configuration(struct comp_dev *dev,
 	DtsSofInterfaceBufferConfiguration *buffer_config)
 {
-	struct comp_buffer *source = list_first_item(&dev->bsource_list, struct comp_buffer,
-						     sink_list);
+	struct comp_buffer *source = comp_dev_get_first_data_producer(dev);
 	const struct audio_stream *stream;
 	DtsSofInterfaceBufferLayout buffer_layout;
 	DtsSofInterfaceBufferFormat buffer_format;
diff --git a/src/audio/copier/copier.c b/src/audio/copier/copier.c
index 7c8d53726d20..a82e54990311 100644
--- a/src/audio/copier/copier.c
+++ b/src/audio/copier/copier.c
@@ -388,7 +388,7 @@ static int copier_comp_trigger(struct comp_dev *dev, int cmd)
 			return -EINVAL;
 		}
 
-		buffer = list_first_item(&dai_copier->bsource_list, struct comp_buffer, sink_list);
+		buffer = comp_dev_get_first_data_producer(dai_copier);
 		pipe_reg.stream_start_offset = posn.dai_posn +
 			latency * audio_stream_period_bytes(&buffer->stream, dev->frames);
 		pipe_reg.stream_end_offset = 0;
@@ -412,7 +412,7 @@ static int copier_comp_trigger(struct comp_dev *dev, int cmd)
 			return -EINVAL;
 		}
 
-		buffer = list_first_item(&dai_copier->bsource_list, struct comp_buffer, sink_list);
+		buffer = comp_dev_get_first_data_producer(dai_copier);
 		pipe_reg.stream_start_offset += latency *
 			audio_stream_period_bytes(&buffer->stream, dev->frames);
 		mailbox_sw_regs_write(cd->pipeline_reg_offset, &pipe_reg.stream_start_offset,
@@ -455,15 +455,13 @@ static int copier_copy_to_sinks(struct copier_data *cd, struct comp_dev *dev,
 				struct comp_buffer *src_c,
 				struct comp_copy_limits *processed_data)
 {
-	struct list_item *sink_list;
 	struct comp_buffer *sink;
 	int ret = 0;
 
 	/* module copy, one source to multiple sink buffers */
-	list_for_item(sink_list, &dev->bsink_list) {
+	comp_dev_for_each_consumer(dev, sink) {
 		struct comp_dev *sink_dev;
 
-		sink = container_of(sink_list, struct comp_buffer, source_list);
 		sink_dev = sink->sink;
 		processed_data->sink_bytes = 0;
 		if (sink_dev->state == COMP_STATE_ACTIVE) {
@@ -563,7 +561,7 @@ static int copier_multi_endpoint_dai_copy(struct copier_data *cd, struct comp_de
 		return -EINVAL;
 	}
 
-	src = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	src = comp_dev_get_first_data_producer(dev);
 
 	/* gateway(s) on output */
 	ret = do_conversion_copy(dev, cd, src, cd->multi_endpoint_buffer, &processed_data);
@@ -1066,14 +1064,14 @@ static int copier_bind(struct processing_module *mod, void *data)
 	const uint32_t src_queue_id = bu->extension.r.src_queue;
 	struct copier_data *cd = module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
-	struct list_item *list;
 
 	if (dev->ipc_config.id != src_id)
 		return 0; /* Another component is a data producer */
 
 	/* update sink format */
-	list_for_item(list, &dev->bsink_list) {
-		struct comp_buffer *buffer = container_of(list, struct comp_buffer, source_list);
+	struct comp_buffer *buffer;
+
+	comp_dev_for_each_consumer(dev, buffer) {
 		uint32_t id = IPC4_SRC_QUEUE_ID(buf_get_id(buffer));
 
 		if (src_queue_id == id) {
diff --git a/src/audio/copier/copier_generic.c b/src/audio/copier/copier_generic.c
index 4cb480ebe3f9..48107d238cae 100644
--- a/src/audio/copier/copier_generic.c
+++ b/src/audio/copier/copier_generic.c
@@ -62,7 +62,6 @@ void copier_update_params(struct copier_data *cd, struct comp_dev *dev,
 			  struct sof_ipc_stream_params *params)
 {
 	struct comp_buffer *sink;
-	struct list_item *sink_list;
 
 	memset(params, 0, sizeof(*params));
 	params->direction = cd->direction;
@@ -80,11 +79,8 @@ void copier_update_params(struct copier_data *cd, struct comp_dev *dev,
 	params->no_stream_position = 1;
 
 	/* update each sink format */
-	list_for_item(sink_list, &dev->bsink_list) {
+	comp_dev_for_each_consumer(dev, sink) {
 		int j;
-
-		sink = container_of(sink_list, struct comp_buffer, source_list);
-
 		j = IPC4_SINK_QUEUE_ID(buf_get_id(sink));
 
 		ipc4_update_buffer_format(sink, &cd->out_fmt[j]);
diff --git a/src/audio/copier/copier_ipcgtw.c b/src/audio/copier/copier_ipcgtw.c
index b5dccbd12986..604d9fecced4 100644
--- a/src/audio/copier/copier_ipcgtw.c
+++ b/src/audio/copier/copier_ipcgtw.c
@@ -75,14 +75,14 @@ static inline struct comp_buffer *get_buffer(struct comp_dev *dev)
 	if (dev->direction == SOF_IPC_STREAM_PLAYBACK) {
 		if (list_is_empty(&dev->bsink_list))
 			return NULL;
-		return list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+		return comp_dev_get_first_data_consumer(dev);
 	}
 
 	assert(dev->direction == SOF_IPC_STREAM_CAPTURE);
 
 	if (list_is_empty(&dev->bsource_list))
 		return NULL;
-	return list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	return comp_dev_get_first_data_producer(dev);
 }
 
 int copier_ipcgtw_process(const struct ipc4_ipcgtw_cmd *cmd,
diff --git a/src/audio/crossover/crossover.c b/src/audio/crossover/crossover.c
index 9fffda1858c5..ffeef13aae1f 100644
--- a/src/audio/crossover/crossover.c
+++ b/src/audio/crossover/crossover.c
@@ -100,15 +100,13 @@ static int crossover_assign_sinks(struct processing_module *mod,
 	struct sof_crossover_config *config = cd->config;
 	struct comp_dev *dev = mod->dev;
 	struct comp_buffer *sink;
-	struct list_item *sink_list;
 	int num_sinks = 0;
 	int i;
 	int j = 0;
 
-	list_for_item(sink_list, &dev->bsink_list) {
+	comp_dev_for_each_consumer(dev, sink) {
 		unsigned int sink_id, state;
 
-		sink = container_of(sink_list, struct comp_buffer, source_list);
 		sink_id = crossover_get_sink_id(cd, buffer_pipeline_id(sink), j);
 		state = sink->sink->state;
 		if (state != dev->state) {
@@ -529,7 +527,6 @@ static int crossover_prepare(struct processing_module *mod,
 	struct comp_data *cd =  module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
 	struct comp_buffer *source, *sink;
-	struct list_item *sink_list;
 	int channels;
 	int ret = 0;
 
@@ -539,15 +536,14 @@ static int crossover_prepare(struct processing_module *mod,
 
 	/* Crossover has a variable number of sinks */
 	mod->max_sinks = SOF_CROSSOVER_MAX_STREAMS;
-	source = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	source = comp_dev_get_first_data_producer(dev);
 
 	/* Get source data format */
 	cd->source_format = audio_stream_get_frm_fmt(&source->stream);
 	channels = audio_stream_get_channels(&source->stream);
 
 	/* Validate frame format and buffer size of sinks */
-	list_for_item(sink_list, &dev->bsink_list) {
-		sink = container_of(sink_list, struct comp_buffer, source_list);
+	comp_dev_for_each_consumer(dev, sink) {
 		if (cd->source_format != audio_stream_get_frm_fmt(&sink->stream)) {
 			comp_err(dev, "crossover_prepare(): Source fmt %d and sink fmt %d are different.",
 				 cd->source_format, audio_stream_get_frm_fmt(&sink->stream));
diff --git a/src/audio/crossover/crossover_ipc3.c b/src/audio/crossover/crossover_ipc3.c
index c715f1710460..07eb39ebca95 100644
--- a/src/audio/crossover/crossover_ipc3.c
+++ b/src/audio/crossover/crossover_ipc3.c
@@ -37,15 +37,13 @@ int crossover_check_sink_assign(struct processing_module *mod,
 {
 	struct comp_dev *dev = mod->dev;
 	struct comp_buffer *sink;
-	struct list_item *sink_list;
 	int num_assigned_sinks = 0;
 	uint8_t assigned_sinks[SOF_CROSSOVER_MAX_STREAMS] = {0};
 	int i;
 
-	list_for_item(sink_list, &dev->bsink_list) {
+	comp_dev_for_each_consumer(dev, sink) {
 		unsigned int pipeline_id;
 
-		sink = container_of(sink_list, struct comp_buffer, source_list);
 		pipeline_id = buffer_pipeline_id(sink);
 
 		i = crossover_get_stream_index(mod, config, pipeline_id);
diff --git a/src/audio/crossover/crossover_ipc4.c b/src/audio/crossover/crossover_ipc4.c
index d0a8fe4adab2..c928a5945034 100644
--- a/src/audio/crossover/crossover_ipc4.c
+++ b/src/audio/crossover/crossover_ipc4.c
@@ -111,7 +111,6 @@ void crossover_params(struct processing_module *mod)
 {
 	struct sof_ipc_stream_params *params = mod->stream_params;
 	struct comp_buffer *sinkb, *sourceb;
-	struct list_item *sink_list;
 	struct comp_dev *dev = mod->dev;
 
 	comp_dbg(dev, "crossover_params()");
@@ -119,11 +118,10 @@ void crossover_params(struct processing_module *mod)
 	ipc4_base_module_cfg_to_stream_params(&mod->priv.cfg.base_cfg, params);
 	component_set_nearest_period_frames(dev, params->rate);
 
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	ipc4_update_buffer_format(sourceb, &mod->priv.cfg.base_cfg.audio_fmt);
 
-	list_for_item(sink_list, &dev->bsink_list) {
-		sinkb = container_of(sink_list, struct comp_buffer, source_list);
+	comp_dev_for_each_consumer(dev, sinkb) {
 		ipc4_update_buffer_format(sinkb, &mod->priv.cfg.base_cfg.audio_fmt);
 	}
 }
diff --git a/src/audio/dai-legacy.c b/src/audio/dai-legacy.c
index af28b8616d9c..189926f6f3d9 100644
--- a/src/audio/dai-legacy.c
+++ b/src/audio/dai-legacy.c
@@ -493,13 +493,9 @@ int dai_common_params(struct dai_data *dd, struct comp_dev *dev,
 	}
 
 	if (dev->direction == SOF_IPC_STREAM_PLAYBACK)
-		dd->local_buffer = list_first_item(&dev->bsource_list,
-						   struct comp_buffer,
-						   sink_list);
+		dd->local_buffer = comp_dev_get_first_data_producer(dev);
 	else
-		dd->local_buffer = list_first_item(&dev->bsink_list,
-						   struct comp_buffer,
-						   source_list);
+		dd->local_buffer = comp_dev_get_first_data_consumer(dev);
 
 	/* check if already configured */
 	if (dev->state == COMP_STATE_PREPARE) {
diff --git a/src/audio/dai-zephyr.c b/src/audio/dai-zephyr.c
index aea7a64d1031..132cc43f1453 100644
--- a/src/audio/dai-zephyr.c
+++ b/src/audio/dai-zephyr.c
@@ -269,18 +269,16 @@ dai_dma_cb(struct dai_data *dd, struct comp_dev *dev, uint32_t bytes,
 
 	if (dev->direction == SOF_IPC_STREAM_PLAYBACK) {
 #if CONFIG_IPC_MAJOR_4
-		struct list_item *sink_list;
 		/*
 		 * copy from local buffer to all sinks that are not gateway buffers
 		 * using the right PCM converter function.
 		 */
-		list_for_item(sink_list, &dev->bsink_list) {
+		struct comp_buffer *sink;
+
+		comp_dev_for_each_consumer(dev, sink) {
 			struct comp_dev *sink_dev;
-			struct comp_buffer *sink;
 			int j;
 
-			sink = container_of(sink_list, struct comp_buffer, source_list);
-
 			if (sink == dd->dma_buffer)
 				continue;
 
@@ -319,20 +317,18 @@ dai_dma_cb(struct dai_data *dd, struct comp_dev *dev, uint32_t bytes,
 		ret = stream_copy_from_no_consume(dd->dma_buffer, dd->local_buffer,
 						  dd->process, bytes, dd->chmap);
 #if CONFIG_IPC_MAJOR_4
-		struct list_item *sink_list;
 		/* Skip in case of endpoint DAI devices created by the copier */
 		if (converter) {
 			/*
 			 * copy from DMA buffer to all sink buffers using the right PCM converter
 			 * function
 			 */
-			list_for_item(sink_list, &dev->bsink_list) {
+			struct comp_buffer *sink;
+
+			comp_dev_for_each_consumer(dev, sink) {
 				struct comp_dev *sink_dev;
-				struct comp_buffer *sink;
 				int j;
 
-				sink = container_of(sink_list, struct comp_buffer, source_list);
-
 				/* this has been handled above already */
 				if (sink == dd->local_buffer)
 					continue;
@@ -852,11 +848,9 @@ static int dai_set_dma_buffer(struct dai_data *dd, struct comp_dev *dev,
 	comp_dbg(dev, "dai_set_dma_buffer()");
 
 	if (dev->direction == SOF_IPC_STREAM_PLAYBACK)
-		dd->local_buffer = list_first_item(&dev->bsource_list, struct comp_buffer,
-						   sink_list);
+		dd->local_buffer = comp_dev_get_first_data_producer(dev);
 	else
-		dd->local_buffer = list_first_item(&dev->bsink_list, struct comp_buffer,
-						   source_list);
+		dd->local_buffer = comp_dev_get_first_data_consumer(dev);
 
 	/* check if already configured */
 	if (dev->state == COMP_STATE_PREPARE) {
@@ -1524,13 +1518,9 @@ static void set_new_local_buffer(struct dai_data *dd, struct comp_dev *dev)
 	uint32_t local_fmt;
 
 	if (dev->direction == SOF_IPC_STREAM_PLAYBACK)
-		dd->local_buffer = list_first_item(&dev->bsource_list,
-						   struct comp_buffer,
-						   sink_list);
+		dd->local_buffer = comp_dev_get_first_data_producer(dev);
 	else
-		dd->local_buffer = list_first_item(&dev->bsink_list,
-						   struct comp_buffer,
-						   source_list);
+		dd->local_buffer = comp_dev_get_first_data_consumer(dev);
 
 	local_fmt = audio_stream_get_frm_fmt(&dd->local_buffer->stream);
 
@@ -1595,17 +1585,16 @@ int dai_common_copy(struct dai_data *dd, struct comp_dev *dev, pcm_converter_fun
 		sink_frames = free_bytes / audio_stream_frame_bytes(&dd->dma_buffer->stream);
 		frames = MIN(src_frames, sink_frames);
 
-		struct list_item *sink_list;
 		/*
 		 * In the case of playback DAI's with multiple sink buffers, compute the
 		 * minimum number of frames based on the DMA avail_bytes and the free
 		 * samples in all active sink buffers.
 		 */
-		list_for_item(sink_list, &dev->bsink_list) {
+		struct comp_buffer *sink;
+
+		comp_dev_for_each_consumer(dev, sink) {
 			struct comp_dev *sink_dev;
-			struct comp_buffer *sink;
 
-			sink = container_of(sink_list, struct comp_buffer, source_list);
 			sink_dev = sink->sink;
 
 			if (sink_dev && sink_dev->state == COMP_STATE_ACTIVE &&
@@ -1616,8 +1605,6 @@ int dai_common_copy(struct dai_data *dd, struct comp_dev *dev, pcm_converter_fun
 			}
 		}
 	} else {
-		struct list_item *sink_list;
-
 		src_frames = avail_bytes / audio_stream_frame_bytes(&dd->dma_buffer->stream);
 
 		/*
@@ -1633,11 +1620,11 @@ int dai_common_copy(struct dai_data *dd, struct comp_dev *dev, pcm_converter_fun
 			 * minimum number of samples based on the DMA avail_bytes and the free
 			 * samples in all active sink buffers.
 			 */
-			list_for_item(sink_list, &dev->bsink_list) {
+			struct comp_buffer *sink;
+
+			comp_dev_for_each_consumer(dev, sink) {
 				struct comp_dev *sink_dev;
-				struct comp_buffer *sink;
 
-				sink = container_of(sink_list, struct comp_buffer, source_list);
 				sink_dev = sink->sink;
 
 				if (sink_dev && sink_dev->state == COMP_STATE_ACTIVE &&
diff --git a/src/audio/dcblock/dcblock.c b/src/audio/dcblock/dcblock.c
index 6b7c23edd906..de8a958733e5 100644
--- a/src/audio/dcblock/dcblock.c
+++ b/src/audio/dcblock/dcblock.c
@@ -199,8 +199,8 @@ static int dcblock_prepare(struct processing_module *mod,
 	dcblock_params(mod);
 
 	/* DC Filter component will only ever have one source and sink buffer */
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* get source data format */
 	cd->source_format = audio_stream_get_frm_fmt(&sourceb->stream);
diff --git a/src/audio/dcblock/dcblock_ipc4.c b/src/audio/dcblock/dcblock_ipc4.c
index e89dc0a22415..534510841b85 100644
--- a/src/audio/dcblock/dcblock_ipc4.c
+++ b/src/audio/dcblock/dcblock_ipc4.c
@@ -61,10 +61,10 @@ void dcblock_params(struct processing_module *mod)
 	ipc4_base_module_cfg_to_stream_params(&mod->priv.cfg.base_cfg, params);
 	component_set_nearest_period_frames(dev, params->rate);
 
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 	ipc4_update_buffer_format(sinkb, &mod->priv.cfg.base_cfg.audio_fmt);
 
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	ipc4_update_buffer_format(sourceb, &mod->priv.cfg.base_cfg.audio_fmt);
 }
 
diff --git a/src/audio/drc/drc.c b/src/audio/drc/drc.c
index 13b33415953b..f02c8ff2a4fe 100644
--- a/src/audio/drc/drc.c
+++ b/src/audio/drc/drc.c
@@ -303,10 +303,10 @@ static void drc_params(struct processing_module *mod)
 	ipc4_base_module_cfg_to_stream_params(&mod->priv.cfg.base_cfg, params);
 	component_set_nearest_period_frames(dev, params->rate);
 
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 	ipc4_update_buffer_format(sinkb, &mod->priv.cfg.base_cfg.audio_fmt);
 
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	ipc4_update_buffer_format(sourceb, &mod->priv.cfg.base_cfg.audio_fmt);
 }
 #endif /* CONFIG_IPC_MAJOR_4 */
@@ -329,8 +329,8 @@ static int drc_prepare(struct processing_module *mod,
 #endif
 
 	/* DRC component will only ever have 1 source and 1 sink buffer */
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* get source data format */
 	cd->source_format = audio_stream_get_frm_fmt(&sourceb->stream);
diff --git a/src/audio/eq_fir/eq_fir.c b/src/audio/eq_fir/eq_fir.c
index 13871815ffdb..0fb40145aa97 100644
--- a/src/audio/eq_fir/eq_fir.c
+++ b/src/audio/eq_fir/eq_fir.c
@@ -424,8 +424,8 @@ static int eq_fir_prepare(struct processing_module *mod,
 	}
 
 	/* EQ component will only ever have 1 source and 1 sink buffer. */
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 	eq_fir_set_alignment(&sourceb->stream, &sinkb->stream);
 	channels = audio_stream_get_channels(&sinkb->stream);
 	frame_fmt = audio_stream_get_frm_fmt(&sourceb->stream);
diff --git a/src/audio/eq_fir/eq_fir_ipc4.c b/src/audio/eq_fir/eq_fir_ipc4.c
index 047d74e0e3b9..eec4165b748b 100644
--- a/src/audio/eq_fir/eq_fir_ipc4.c
+++ b/src/audio/eq_fir/eq_fir_ipc4.c
@@ -59,10 +59,10 @@ int eq_fir_params(struct processing_module *mod)
 	ipc4_base_module_cfg_to_stream_params(&mod->priv.cfg.base_cfg, params);
 	component_set_nearest_period_frames(dev, params->rate);
 
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	ipc4_update_buffer_format(sourceb, &mod->priv.cfg.base_cfg.audio_fmt);
 
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 	ipc4_update_buffer_format(sinkb, &mod->priv.cfg.base_cfg.audio_fmt);
 
 	return 0;
diff --git a/src/audio/eq_iir/eq_iir.c b/src/audio/eq_iir/eq_iir.c
index a5b04a418691..7e2e54df7df6 100644
--- a/src/audio/eq_iir/eq_iir.c
+++ b/src/audio/eq_iir/eq_iir.c
@@ -193,8 +193,8 @@ static int eq_iir_prepare(struct processing_module *mod,
 		return ret;
 
 	/* EQ component will only ever have 1 source and 1 sink buffer */
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 	eq_iir_set_alignment(&sourceb->stream, &sinkb->stream);
 
 	/* get source and sink data format */
diff --git a/src/audio/eq_iir/eq_iir_ipc3.c b/src/audio/eq_iir/eq_iir_ipc3.c
index 6958dd6b536b..7ccc375021fe 100644
--- a/src/audio/eq_iir/eq_iir_ipc3.c
+++ b/src/audio/eq_iir/eq_iir_ipc3.c
@@ -274,10 +274,8 @@ static int eq_iir_verify_params(struct comp_dev *dev,
 	comp_dbg(dev, "eq_iir_verify_params()");
 
 	/* EQ component will only ever have 1 source and 1 sink buffer */
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer,
-				  sink_list);
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer,
-				source_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* we check whether we can support frame_fmt conversion (whether we have
 	 * such conversion function) due to source and sink buffer frame_fmt's.
diff --git a/src/audio/eq_iir/eq_iir_ipc4.c b/src/audio/eq_iir/eq_iir_ipc4.c
index 474b820b14de..7106ec3d7343 100644
--- a/src/audio/eq_iir/eq_iir_ipc4.c
+++ b/src/audio/eq_iir/eq_iir_ipc4.c
@@ -125,7 +125,7 @@ static int eq_iir_params(struct processing_module *mod)
 		comp_params.chmap[i] = (mod->priv.cfg.base_cfg.audio_fmt.ch_map >> i * 4) & 0xf;
 
 	component_set_nearest_period_frames(dev, comp_params.rate);
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 	ret = buffer_set_params(sinkb, &comp_params, true);
 	return ret;
 }
diff --git a/src/audio/google/google_ctc_audio_processing.c b/src/audio/google/google_ctc_audio_processing.c
index 68d384789ab5..859d2c7caba9 100644
--- a/src/audio/google/google_ctc_audio_processing.c
+++ b/src/audio/google/google_ctc_audio_processing.c
@@ -355,7 +355,7 @@ static int ctc_prepare(struct processing_module *mod,
 
 	comp_info(mod->dev, "ctc_prepare()");
 
-	source = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	source = comp_dev_get_first_data_producer(dev);
 	switch (audio_stream_get_frm_fmt(&source->stream)) {
 #if CONFIG_FORMAT_S16LE
 	case SOF_IPC_FRAME_S16_LE:
diff --git a/src/audio/google/google_hotword_detect.c b/src/audio/google/google_hotword_detect.c
index c79d5917c432..9921deae06a7 100644
--- a/src/audio/google/google_hotword_detect.c
+++ b/src/audio/google/google_hotword_detect.c
@@ -170,8 +170,7 @@ static int ghd_params(struct comp_dev *dev,
 	}
 
 	/* This detector component will only ever have 1 source */
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer,
-				  sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 
 	if (audio_stream_get_channels(sourceb->stream) != 1) {
 		comp_err(dev, "ghd_params(): Only single-channel supported");
@@ -391,8 +390,7 @@ static int ghd_copy(struct comp_dev *dev)
 	}
 
 	/* keyword components will only ever have 1 source */
-	source = list_first_item(&dev->bsource_list,
-				 struct comp_buffer, sink_list);
+	source = comp_dev_get_first_data_producer(dev);
 	stream = &sourceb->stream;
 
 	bytes = audio_stream_get_avail_bytes(stream);
diff --git a/src/audio/google/google_rtc_audio_processing.c b/src/audio/google/google_rtc_audio_processing.c
index cc5dc1bb6983..0fbe20b47976 100644
--- a/src/audio/google/google_rtc_audio_processing.c
+++ b/src/audio/google/google_rtc_audio_processing.c
@@ -633,10 +633,9 @@ static int google_rtc_audio_processing_prepare(struct processing_module *mod,
 	/* Don't need the ref buffer on IPC4 as pipelines are always
 	 * activated in tandem; also the API is deprecated
 	 */
-	cd->ref_comp_buffer = list_first_item(&dev->bsource_list,
-					      struct comp_buffer, sink_list);
+	cd->ref_comp_buffer = comp_dev_get_first_data_producer(dev);
 	if (cd->aec_reference_source == 1)
-		cd->ref_comp_buffer = list_next_item(cd->ref_comp_buffer, sink_list);
+		cd->ref_comp_buffer = comp_dev_get_next_data_producer(dev, cd->ref_comp_buffer);
 #endif
 
 #ifdef CONFIG_IPC_MAJOR_4
diff --git a/src/audio/host-legacy.c b/src/audio/host-legacy.c
index 5326845fe021..73fa30e1b61c 100644
--- a/src/audio/host-legacy.c
+++ b/src/audio/host-legacy.c
@@ -712,13 +712,9 @@ int host_common_params(struct host_data *hd, struct comp_dev *dev,
 	}
 
 	if (dev->direction == SOF_IPC_STREAM_PLAYBACK)
-		hd->local_buffer = list_first_item(&dev->bsink_list,
-						   struct comp_buffer,
-						   source_list);
+		hd->local_buffer = comp_dev_get_first_data_consumer(dev);
 	else
-		hd->local_buffer = list_first_item(&dev->bsource_list,
-						   struct comp_buffer,
-						   sink_list);
+		hd->local_buffer = comp_dev_get_first_data_producer(dev);
 
 	period_bytes = dev->frames *
 		audio_stream_frame_bytes(&hd->local_buffer->stream);
diff --git a/src/audio/host-zephyr.c b/src/audio/host-zephyr.c
index ac1751059eb5..d3bc506d2fbf 100644
--- a/src/audio/host-zephyr.c
+++ b/src/audio/host-zephyr.c
@@ -793,13 +793,9 @@ int host_common_params(struct host_data *hd, struct comp_dev *dev,
 	}
 
 	if (params->direction == SOF_IPC_STREAM_PLAYBACK)
-		hd->local_buffer = list_first_item(&dev->bsink_list,
-						   struct comp_buffer,
-						   source_list);
+		hd->local_buffer = comp_dev_get_first_data_consumer(dev);
 	else
-		hd->local_buffer = list_first_item(&dev->bsource_list,
-						   struct comp_buffer,
-						   sink_list);
+		hd->local_buffer = comp_dev_get_first_data_producer(dev);
 
 	period_bytes = dev->frames * get_frame_bytes(params->frame_fmt, params->channels);
 
diff --git a/src/audio/kpb.c b/src/audio/kpb.c
index 542984e88867..6e83747831eb 100644
--- a/src/audio/kpb.c
+++ b/src/audio/kpb.c
@@ -328,7 +328,6 @@ static int kpb_bind(struct comp_dev *dev, void *data)
 {
 	struct comp_data *kpb = comp_get_drvdata(dev);
 	struct ipc4_module_bind_unbind *bu;
-	struct list_item *blist;
 	int buf_id;
 	int ret = 0;
 
@@ -343,9 +342,9 @@ static int kpb_bind(struct comp_dev *dev, void *data)
 	 * (Detector/MicSel has one input pin). To properly connect KPB sink
 	 * with Detector source we're looking for buffer with id=0.
 	 */
+	struct comp_buffer *sink;
 
-	list_for_item(blist, &dev->bsink_list) {
-		struct comp_buffer *sink = container_of(blist, struct comp_buffer, source_list);
+	comp_dev_for_each_consumer(dev, sink) {
 		int sink_buf_id;
 
 		if (!sink->sink) {
@@ -858,10 +857,9 @@ static int kpb_prepare(struct comp_dev *dev)
 	 * NOTE! We assume here that channel selector component device
 	 * is connected to the KPB sinks as well as host device.
 	 */
-	struct list_item *blist;
+	struct comp_buffer *sink;
 
-	list_for_item(blist, &dev->bsink_list) {
-		struct comp_buffer *sink = container_of(blist, struct comp_buffer, source_list);
+	comp_dev_for_each_consumer(dev, sink) {
 		enum sof_comp_type type;
 
 		if (!sink->sink) {
@@ -888,13 +886,11 @@ static int kpb_prepare(struct comp_dev *dev)
 	 * If OBS is not equal to IBS it means that KPB will work in micselector mode.
 	 */
 	if (kpb->ipc4_cfg.base_cfg.ibs != kpb->ipc4_cfg.base_cfg.obs) {
-		struct list_item *sink_list;
 		uint32_t sink_id;
 
-		list_for_item(sink_list, &dev->bsink_list) {
-			struct comp_buffer *sink =
-				container_of(sink_list, struct comp_buffer, source_list);
+		struct comp_buffer *sink;
 
+		comp_dev_for_each_consumer(dev, sink) {
 			sink_id = buf_get_id(sink);
 
 			if (sink_id == 0)
@@ -1186,8 +1182,7 @@ static int kpb_copy(struct comp_dev *dev)
 	}
 
 	/* Get source and sink buffers */
-	source = list_first_item(&dev->bsource_list, struct comp_buffer,
-				 sink_list);
+	source = comp_dev_get_first_data_producer(dev);
 
 	/* Validate source */
 	if (!audio_stream_get_rptr(&source->stream)) {
diff --git a/src/audio/mfcc/mfcc.c b/src/audio/mfcc/mfcc.c
index bc07a5058496..3a56f2f95b46 100644
--- a/src/audio/mfcc/mfcc.c
+++ b/src/audio/mfcc/mfcc.c
@@ -192,8 +192,8 @@ static int mfcc_prepare(struct processing_module *mod,
 	comp_info(dev, "mfcc_prepare()");
 
 	/* MFCC component will only ever have 1 source and 1 sink buffer */
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* get source data format */
 	source_format = audio_stream_get_frm_fmt(&sourceb->stream);
diff --git a/src/audio/mixer/mixer.c b/src/audio/mixer/mixer.c
index 6e9d6acf4493..c04bc7dff395 100644
--- a/src/audio/mixer/mixer.c
+++ b/src/audio/mixer/mixer.c
@@ -163,16 +163,15 @@ static int mixer_reset(struct processing_module *mod)
 {
 	struct mixer_data *md = module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
-	struct list_item *blist;
 	int dir = dev->pipeline->source_comp->direction;
 
 	comp_dbg(dev, "mixer_reset()");
 
 	if (dir == SOF_IPC_STREAM_PLAYBACK) {
-		list_for_item(blist, &dev->bsource_list) {
+		struct comp_buffer *source;
+
+		comp_dev_for_each_producer(dev, source) {
 			/* FIXME: this is racy and implicitly protected by serialised IPCs */
-			struct comp_buffer *source = container_of(blist, struct comp_buffer,
-								  sink_list);
 			bool stop = false;
 
 			if (source->source && source->source->state > COMP_STATE_READY)
@@ -214,16 +213,15 @@ static int mixer_prepare(struct processing_module *mod,
 	struct mixer_data *md = module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
 	struct comp_buffer *sink;
-	struct list_item *blist;
 
-	sink = list_first_item(&dev->bsink_list, struct comp_buffer,
-			       source_list);
+	sink = comp_dev_get_first_data_consumer(dev);
 	md->mix_func = mixer_get_processing_function(dev, sink);
 	mixer_set_frame_alignment(&sink->stream);
 
 	/* check each mixer source state */
-	list_for_item(blist, &dev->bsource_list) {
-		struct comp_buffer *source;
+	struct comp_buffer *source;
+
+	comp_dev_for_each_producer(dev, source) {
 		bool stop;
 
 		/*
@@ -234,7 +232,6 @@ static int mixer_prepare(struct processing_module *mod,
 		 * preparing the mixer, so they shouldn't touch it until we're
 		 * done.
 		 */
-		source = container_of(blist, struct comp_buffer, sink_list);
 		mixer_set_frame_alignment(&source->stream);
 		stop = source->source && (source->source->state == COMP_STATE_PAUSED ||
 					    source->source->state == COMP_STATE_ACTIVE);
diff --git a/src/audio/module_adapter/module/waves/waves.c b/src/audio/module_adapter/module/waves/waves.c
index efa1780fd482..8ea54c8e21ed 100644
--- a/src/audio/module_adapter/module/waves/waves.c
+++ b/src/audio/module_adapter/module/waves/waves.c
@@ -215,10 +215,8 @@ static int waves_effect_allocate(struct processing_module *mod)
 /* checks if sink/source parameters fit MaxxEffect */
 static int waves_effect_check(struct comp_dev *dev)
 {
-	struct comp_buffer *sink = list_first_item(&dev->bsink_list, struct comp_buffer,
-						    source_list);
-	struct comp_buffer *source = list_first_item(&dev->bsource_list, struct comp_buffer,
-						     sink_list);
+	struct comp_buffer *sink = comp_dev_get_first_data_consumer(dev);
+	struct comp_buffer *source = comp_dev_get_first_data_producer(dev);
 	const struct audio_stream *src_fmt = &source->stream;
 	const struct audio_stream *snk_fmt = &sink->stream;
 
@@ -284,8 +282,7 @@ static int waves_effect_check(struct comp_dev *dev)
 static int waves_effect_init(struct processing_module *mod)
 {
 	struct comp_dev *dev = mod->dev;
-	struct comp_buffer *source = list_first_item(&dev->bsource_list, struct comp_buffer,
-						     sink_list);
+	struct comp_buffer *source = comp_dev_get_first_data_producer(dev);
 	struct module_data *codec = &mod->priv;
 	struct waves_codec_data *waves_codec = codec->private;
 	const struct audio_stream *src_fmt = &source->stream;
diff --git a/src/audio/multiband_drc/multiband_drc.c b/src/audio/multiband_drc/multiband_drc.c
index 47c2343912c6..8a4c9530cdfa 100644
--- a/src/audio/multiband_drc/multiband_drc.c
+++ b/src/audio/multiband_drc/multiband_drc.c
@@ -364,7 +364,7 @@ static int multiband_drc_prepare(struct processing_module *mod,
 		return ret;
 
 	/* DRC component will only ever have 1 source and 1 sink buffer */
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 
 	/* get source data format */
 	cd->source_format = audio_stream_get_frm_fmt(&sourceb->stream);
diff --git a/src/audio/multiband_drc/multiband_drc_ipc4.c b/src/audio/multiband_drc/multiband_drc_ipc4.c
index e2471d6f2a8e..a573d1101c5b 100644
--- a/src/audio/multiband_drc/multiband_drc_ipc4.c
+++ b/src/audio/multiband_drc/multiband_drc_ipc4.c
@@ -98,7 +98,7 @@ int multiband_drc_params(struct processing_module *mod)
 		comp_params.chmap[i] = (mod->priv.cfg.base_cfg.audio_fmt.ch_map >> i * 4) & 0xf;
 
 	component_set_nearest_period_frames(dev, comp_params.rate);
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 	ret = buffer_set_params(sinkb, &comp_params, true);
 
 	return ret;
diff --git a/src/audio/mux/mux.c b/src/audio/mux/mux.c
index 9ac9365297e6..91bf8f45e82d 100644
--- a/src/audio/mux/mux.c
+++ b/src/audio/mux/mux.c
@@ -233,7 +233,6 @@ static int demux_process(struct processing_module *mod,
 {
 	struct comp_data *cd = module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
-	struct list_item *clist;
 	struct comp_buffer *sink;
 	struct audio_stream *sinks_stream[MUX_MAX_STREAMS] = { NULL };
 	struct mux_look_up *look_ups[MUX_MAX_STREAMS] = { NULL };
@@ -245,8 +244,7 @@ static int demux_process(struct processing_module *mod,
 	comp_dbg(dev, "demux_process()");
 
 	/* align sink streams with their respective configurations */
-	list_for_item(clist, &dev->bsink_list) {
-		sink = container_of(clist, struct comp_buffer, source_list);
+	comp_dev_for_each_consumer(dev, sink) {
 		if (sink->sink->state == dev->state) {
 			i = get_stream_index(dev, cd, buffer_pipeline_id(sink));
 			/* return if index wrong */
@@ -291,7 +289,6 @@ static int mux_process(struct processing_module *mod,
 	struct comp_data *cd = module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
 	struct comp_buffer *source;
-	struct list_item *clist;
 	const struct audio_stream *sources_stream[MUX_MAX_STREAMS] = { NULL };
 	int frames = 0;
 	int sink_bytes;
@@ -302,8 +299,7 @@ static int mux_process(struct processing_module *mod,
 
 	/* align source streams with their respective configurations */
 	j = 0;
-	list_for_item(clist, &dev->bsource_list) {
-		source = container_of(clist, struct comp_buffer, sink_list);
+	comp_dev_for_each_producer(dev, source) {
 		if (source->source->state == dev->state) {
 			if (frames)
 				frames = MIN(frames, input_buffers[j].size);
@@ -334,8 +330,7 @@ static int mux_process(struct processing_module *mod,
 
 	/* Update consumed and produced */
 	j = 0;
-	list_for_item(clist, &dev->bsource_list) {
-		source = container_of(clist, struct comp_buffer, sink_list);
+	comp_dev_for_each_producer(dev, source) {
 		if (source->source->state == dev->state)
 			mod->input_buffers[j].consumed = source_bytes;
 		j++;
@@ -346,7 +341,7 @@ static int mux_process(struct processing_module *mod,
 
 static int mux_reset(struct processing_module *mod)
 {
-	struct list_item *blist;
+	struct comp_buffer *source;
 	struct comp_data *cd = module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
 	int dir = dev->pipeline->source_comp->direction;
@@ -354,9 +349,7 @@ static int mux_reset(struct processing_module *mod)
 	comp_dbg(dev, "mux_reset()");
 
 	if (dir == SOF_IPC_STREAM_PLAYBACK) {
-		list_for_item(blist, &dev->bsource_list) {
-			struct comp_buffer *source = container_of(blist, struct comp_buffer,
-								  sink_list);
+		comp_dev_for_each_producer(dev, source) {
 			int state = source->source->state;
 
 			/* only mux the sources with the same state with mux */
@@ -436,9 +429,6 @@ static int mux_set_config(struct processing_module *mod, uint32_t config_id,
 
 static int demux_trigger(struct processing_module *mod, int cmd)
 {
-	struct list_item *li;
-	struct comp_buffer *b;
-
 	/* Check for cross-pipeline sinks: in general foreign
 	 * pipelines won't be started synchronously with ours (it's
 	 * under control of host software), so output can't be
@@ -448,8 +438,9 @@ static int demux_trigger(struct processing_module *mod, int cmd)
 	 * themselves.
 	 */
 	if (cmd == COMP_TRIGGER_PRE_START) {
-		list_for_item(li, &mod->dev->bsink_list) {
-			b = container_of(li, struct comp_buffer, source_list);
+		struct comp_buffer *b;
+
+		comp_dev_for_each_producer(mod->dev, b) {
 			if (b->sink->pipeline != mod->dev->pipeline)
 				audio_stream_set_overrun(&b->stream, true);
 		}
diff --git a/src/audio/mux/mux_generic.c b/src/audio/mux/mux_generic.c
index ee4b1eddc91c..991b79097fec 100644
--- a/src/audio/mux/mux_generic.c
+++ b/src/audio/mux/mux_generic.c
@@ -527,8 +527,7 @@ mux_func mux_get_processing_function(struct processing_module *mod)
 	if (list_is_empty(&dev->bsink_list))
 		return NULL;
 
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer,
-				source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	for (i = 0; i < ARRAY_SIZE(mux_func_map); i++) {
 		enum sof_ipc_frame fmt = audio_stream_get_frm_fmt(&sinkb->stream);
@@ -550,8 +549,7 @@ demux_func demux_get_processing_function(struct processing_module *mod)
 	if (list_is_empty(&dev->bsource_list))
 		return NULL;
 
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer,
-				sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 
 	for (i = 0; i < ARRAY_SIZE(mux_func_map); i++) {
 		enum sof_ipc_frame fmt = audio_stream_get_frm_fmt(&sourceb->stream);
diff --git a/src/audio/mux/mux_ipc4.c b/src/audio/mux/mux_ipc4.c
index 72c9e9e29c7d..6d183d6973f6 100644
--- a/src/audio/mux/mux_ipc4.c
+++ b/src/audio/mux/mux_ipc4.c
@@ -74,7 +74,6 @@ static void set_mux_params(struct processing_module *mod)
 	struct comp_data *cd = module_get_private_data(mod);
 	struct comp_dev *dev = mod->dev;
 	struct comp_buffer *sink, *source;
-	struct list_item *source_list;
 	int j;
 
 	params->direction = dev->direction;
@@ -94,7 +93,7 @@ static void set_mux_params(struct processing_module *mod)
 
 	/* update sink format */
 	if (!list_is_empty(&dev->bsink_list)) {
-		sink = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+		sink = comp_dev_get_first_data_consumer(dev);
 
 		if (!sink->hw_params_configured) {
 			ipc4_update_buffer_format(sink, &cd->md.output_format);
@@ -106,9 +105,7 @@ static void set_mux_params(struct processing_module *mod)
 	if (!list_is_empty(&dev->bsource_list)) {
 		struct ipc4_audio_format *audio_fmt;
 
-		list_for_item(source_list, &dev->bsource_list)
-		{
-			source = container_of(source_list, struct comp_buffer, sink_list);
+		comp_dev_for_each_producer(dev, source) {
 			j = buf_get_id(source);
 			cd->config.streams[j].pipeline_id = buffer_pipeline_id(source);
 			if (j == BASE_CFG_QUEUED_ID)
diff --git a/src/audio/rtnr/rtnr.c b/src/audio/rtnr/rtnr.c
index c12034b6e793..4ac99d1b35f9 100644
--- a/src/audio/rtnr/rtnr.c
+++ b/src/audio/rtnr/rtnr.c
@@ -781,10 +781,10 @@ static void rtnr_params(struct processing_module *mod)
 	ipc4_base_module_cfg_to_stream_params(&mod->priv.cfg.base_cfg, params);
 	component_set_nearest_period_frames(dev, params->rate);
 
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	ipc4_update_buffer_format(sourceb, &mod->priv.cfg.base_cfg.audio_fmt);
 
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 	ipc4_update_buffer_format(sinkb, &mod->priv.cfg.base_cfg.audio_fmt);
 }
 #endif
@@ -814,10 +814,10 @@ static int rtnr_prepare(struct processing_module *mod,
 	/* Initialize RTNR */
 
 	/* Get sink data format */
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 	cd->sink_format = audio_stream_get_frm_fmt(&sinkb->stream);
 	cd->sink_stream.frame_fmt = audio_stream_get_frm_fmt(&sinkb->stream);
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	ret = rtnr_check_params(mod, &sourceb->stream, &sinkb->stream);
 	if (ret)
 		goto err;
diff --git a/src/audio/selector/selector.c b/src/audio/selector/selector.c
index 79140baec302..a14e584007bb 100644
--- a/src/audio/selector/selector.c
+++ b/src/audio/selector/selector.c
@@ -62,8 +62,7 @@ static int selector_verify_params(struct comp_dev *dev,
 
 	comp_dbg(dev, "selector_verify_params()");
 
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer,
-				source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* check whether params->channels (received from driver) are equal to
 	 * cd->config.in_channels_count (PLAYBACK) or
@@ -73,8 +72,7 @@ static int selector_verify_params(struct comp_dev *dev,
 	 */
 	if (dev->direction == SOF_IPC_STREAM_PLAYBACK) {
 		/* fetch sink buffer for playback */
-		buffer = list_first_item(&dev->bsink_list, struct comp_buffer,
-					 source_list);
+		buffer = comp_dev_get_first_data_consumer(dev);
 		if (cd->config.in_channels_count &&
 		    cd->config.in_channels_count != params->channels) {
 			comp_err(dev, "selector_verify_params(): src in_channels_count does not match pcm channels");
@@ -92,8 +90,7 @@ static int selector_verify_params(struct comp_dev *dev,
 		params->channels = out_channels;
 	} else {
 		/* fetch source buffer for capture */
-		buffer = list_first_item(&dev->bsource_list, struct comp_buffer,
-					 sink_list);
+		buffer = comp_dev_get_first_data_producer(dev);
 		if (cd->config.out_channels_count &&
 		    cd->config.out_channels_count != params->channels) {
 			comp_err(dev, "selector_verify_params(): src in_channels_count does not match pcm channels");
@@ -355,8 +352,7 @@ static int selector_trigger(struct comp_dev *dev, int cmd)
 
 	comp_dbg(dev, "selector_trigger()");
 
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer,
-				  sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 
 	ret = comp_set_state(dev, cmd);
 
@@ -385,10 +381,8 @@ static int selector_copy(struct comp_dev *dev)
 	comp_dbg(dev, "selector_copy()");
 
 	/* selector component will have 1 source and 1 sink buffer */
-	source = list_first_item(&dev->bsource_list, struct comp_buffer,
-				 sink_list);
-	sink = list_first_item(&dev->bsink_list, struct comp_buffer,
-			       source_list);
+	source = comp_dev_get_first_data_producer(dev);
+	sink = comp_dev_get_first_data_consumer(dev);
 
 	if (!audio_stream_get_avail(&source->stream))
 		return PPL_STATUS_PATH_STOP;
@@ -434,10 +428,8 @@ static int selector_prepare(struct comp_dev *dev)
 		return PPL_STATUS_PATH_STOP;
 
 	/* selector component will have 1 source and 1 sink buffer */
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer,
-				  sink_list);
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer,
-				source_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* get source data format and period bytes */
 	cd->source_format = audio_stream_get_frm_fmt(&sourceb->stream);
@@ -648,7 +640,6 @@ static void set_selector_params(struct processing_module *mod,
 	const struct sof_selector_ipc4_config *sel_cfg = &cd->sel_ipc4_cfg;
 	const struct ipc4_audio_format *out_fmt = NULL;
 	struct comp_buffer *src_buf;
-	struct list_item *sink_list;
 	int i;
 
 	if (cd->sel_ipc4_cfg.init_payload_fmt == IPC4_SEL_INIT_PAYLOAD_BASE_WITH_EXT)
@@ -668,10 +659,9 @@ static void set_selector_params(struct processing_module *mod,
 		params->chmap[i] = (out_fmt->ch_map >> i * 4) & 0xf;
 
 	/* update each sink format */
-	list_for_item(sink_list, &dev->bsink_list) {
-		struct comp_buffer *sink_buf =
-			container_of(sink_list, struct comp_buffer, source_list);
+	struct comp_buffer *sink_buf;
 
+	comp_dev_for_each_consumer(dev, sink_buf) {
 		ipc4_update_buffer_format(sink_buf, out_fmt);
 		audio_stream_set_channels(&sink_buf->stream, params->channels);
 		audio_stream_set_rate(&sink_buf->stream, params->rate);
@@ -683,7 +673,7 @@ static void set_selector_params(struct processing_module *mod,
 	 * for a short time when the second pipeline already started
 	 * and the first one is not ready yet along with sink buffers params
 	 */
-	src_buf = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	src_buf = comp_dev_get_first_data_producer(dev);
 
 	if (!src_buf->hw_params_configured)
 		ipc4_update_buffer_format(src_buf, &mod->priv.cfg.base_cfg.audio_fmt);
@@ -715,15 +705,15 @@ static int selector_verify_params(struct processing_module *mod,
 	/* apply input/output channels count according to stream direction */
 	if (dev->direction == SOF_IPC_STREAM_PLAYBACK) {
 		params->channels = out_channels;
-		buffer = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+		buffer = comp_dev_get_first_data_consumer(dev);
 	} else {
 		params->channels = in_channels;
-		buffer = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+		buffer = comp_dev_get_first_data_producer(dev);
 	}
 	buffer_set_params(buffer, params, BUFFER_UPDATE_FORCE);
 
 	/* set component period frames */
-	buffer = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	buffer = comp_dev_get_first_data_consumer(dev);
 	component_set_nearest_period_frames(dev, audio_stream_get_rate(&buffer->stream));
 
 	return 0;
@@ -843,8 +833,8 @@ static int selector_prepare(struct processing_module *mod,
 		return PPL_STATUS_PATH_STOP;
 
 	/* selector component will have 1 source and 1 sink buffer */
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	audio_stream_set_align(4, 1, &sourceb->stream);
 	audio_stream_set_align(4, 1, &sinkb->stream);
diff --git a/src/audio/smart_amp/smart_amp.c b/src/audio/smart_amp/smart_amp.c
index 154761fd0937..e55d9cdb26b6 100644
--- a/src/audio/smart_amp/smart_amp.c
+++ b/src/audio/smart_amp/smart_amp.c
@@ -731,7 +731,6 @@ static int smart_amp_resolve_mod_fmt(struct comp_dev *dev, uint32_t least_req_de
 static int smart_amp_prepare(struct comp_dev *dev)
 {
 	struct smart_amp_data *sad = comp_get_drvdata(dev);
-	struct list_item *blist;
 	uint16_t ff_src_fmt, fb_src_fmt, resolved_mod_fmt;
 	uint32_t least_req_depth;
 	uint32_t rate;
@@ -744,10 +743,9 @@ static int smart_amp_prepare(struct comp_dev *dev)
 		return ret;
 
 	/* searching for stream and feedback source buffers */
-	list_for_item(blist, &dev->bsource_list) {
-		struct comp_buffer *source_buffer = container_of(blist, struct comp_buffer,
-								 sink_list);
+	struct comp_buffer *source_buffer;
 
+	comp_dev_for_each_producer(dev, source_buffer) {
 		if (source_buffer->source->ipc_config.type == SOF_COMP_DEMUX)
 			sad->feedback_buf = source_buffer;
 		else
@@ -755,8 +753,7 @@ static int smart_amp_prepare(struct comp_dev *dev)
 	}
 
 	/* sink buffer */
-	sad->sink_buf = list_first_item(&dev->bsink_list, struct comp_buffer,
-					source_list);
+	sad->sink_buf = comp_dev_get_first_data_consumer(dev);
 
 	/* get frame format and channels param of stream and feedback source */
 	ff_src_fmt = audio_stream_get_frm_fmt(&sad->source_buf->stream);
diff --git a/src/audio/tdfb/tdfb.c b/src/audio/tdfb/tdfb.c
index 128a39b19e1c..f4821623966a 100644
--- a/src/audio/tdfb/tdfb.c
+++ b/src/audio/tdfb/tdfb.c
@@ -739,8 +739,8 @@ static int tdfb_prepare(struct processing_module *mod,
 	}
 
 	/* Find source and sink buffers */
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 	tdfb_set_alignment(&sourceb->stream, &sinkb->stream);
 
 	frame_fmt = audio_stream_get_frm_fmt(&sourceb->stream);
diff --git a/src/audio/tdfb/tdfb_ipc4.c b/src/audio/tdfb/tdfb_ipc4.c
index 5360b1013e97..7a16328d952d 100644
--- a/src/audio/tdfb/tdfb_ipc4.c
+++ b/src/audio/tdfb/tdfb_ipc4.c
@@ -198,10 +198,10 @@ int tdfb_params(struct processing_module *mod)
 
 	ipc4_base_module_cfg_to_stream_params(&mod->priv.cfg.base_cfg, params);
 	component_set_nearest_period_frames(dev, params->rate);
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	ipc4_update_buffer_format(sourceb, &mod->priv.cfg.input_pins[0].audio_fmt);
 
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 	ipc4_update_buffer_format(sinkb, &mod->priv.cfg.output_pins[0].audio_fmt);
 	return 0;
 }
diff --git a/src/audio/tone.c b/src/audio/tone.c
index d1975029fa04..d289c12d5f57 100644
--- a/src/audio/tone.c
+++ b/src/audio/tone.c
@@ -428,11 +428,9 @@ static int tone_params(struct comp_dev *dev,
 	struct comp_data *cd = comp_get_drvdata(dev);
 	struct comp_buffer *sourceb, *sinkb;
 
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer,
-				  sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer,
-				source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	comp_info(dev, "tone_params(), config->frame_fmt = %u",
 		  dev->ipc_config.frame_fmt);
@@ -674,8 +672,7 @@ static int tone_prepare(struct comp_dev *dev)
 	if (ret == COMP_STATUS_STATE_ALREADY_SET)
 		return PPL_STATUS_PATH_STOP;
 
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer,
-				  sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 
 	cd->channels = audio_stream_get_channels(&sourceb->stream);
 	comp_info(dev, "tone_prepare(), cd->channels = %u, cd->rate = %u",
diff --git a/src/audio/volume/volume.c b/src/audio/volume/volume.c
index 000a2f27d869..9d5ad6d43e4c 100644
--- a/src/audio/volume/volume.c
+++ b/src/audio/volume/volume.c
@@ -688,10 +688,8 @@ static int volume_prepare(struct processing_module *mod,
 	ret = volume_peak_prepare(cd, mod);
 
 	/* volume component will only ever have 1 sink and source buffer */
-	sinkb = list_first_item(&dev->bsink_list,
-				struct comp_buffer, source_list);
-	sourceb = list_first_item(&dev->bsource_list,
-				  struct comp_buffer, sink_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
+	sourceb = comp_dev_get_first_data_producer(dev);
 
 	volume_set_alignment(&sourceb->stream, &sinkb->stream);
 
diff --git a/src/audio/volume/volume_ipc3.c b/src/audio/volume/volume_ipc3.c
index 2c65f6bba2dc..2ae1d362a787 100644
--- a/src/audio/volume/volume_ipc3.c
+++ b/src/audio/volume/volume_ipc3.c
@@ -39,11 +39,9 @@ void set_volume_process(struct vol_data *cd, struct comp_dev *dev, bool source_o
 	struct comp_buffer *bufferb;
 
 	if (source_or_sink)
-		bufferb = list_first_item(&dev->bsource_list,
-					  struct comp_buffer, sink_list);
+		bufferb = comp_dev_get_first_data_producer(dev);
 	else
-		bufferb = list_first_item(&dev->bsink_list,
-					  struct comp_buffer, source_list);
+		bufferb = comp_dev_get_first_data_consumer(dev);
 
 	cd->scale_vol = vol_get_processing_function(dev, bufferb, cd);
 }
diff --git a/src/audio/volume/volume_ipc4.c b/src/audio/volume/volume_ipc4.c
index 39c4dd061e8c..1e3f8fe3705f 100644
--- a/src/audio/volume/volume_ipc4.c
+++ b/src/audio/volume/volume_ipc4.c
@@ -413,10 +413,10 @@ static int volume_params(struct processing_module *mod)
 	component_set_nearest_period_frames(dev, params->rate);
 
 	/* volume component will only ever have 1 sink buffer */
-	sinkb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sinkb = comp_dev_get_first_data_consumer(dev);
 	ipc4_update_buffer_format(sinkb, &mod->priv.cfg.base_cfg.audio_fmt);
 
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	ipc4_update_buffer_format(sourceb, &mod->priv.cfg.base_cfg.audio_fmt);
 
 	return 0;
diff --git a/src/include/sof/audio/component.h b/src/include/sof/audio/component.h
index c10b3bf67858..1463da2e2c61 100644
--- a/src/include/sof/audio/component.h
+++ b/src/include/sof/audio/component.h
@@ -634,6 +634,64 @@ struct comp_dev {
 #endif
 };
 
+/**
+ * Get a pointer to a first comp_buffer object providing data to the component
+ * The procedure will return NULL if there's no data provider
+ */
+static inline struct comp_buffer *comp_dev_get_first_data_producer(struct comp_dev *component)
+{
+	return list_is_empty(&component->bsource_list) ? NULL :
+	       list_first_item(&component->bsource_list, struct comp_buffer, sink_list);
+}
+
+/**
+ * Get a pointer to a next comp_buffer object providing data to the component
+ * The procedure will return NULL if there're no more data providers
+ */
+static inline struct comp_buffer *comp_dev_get_next_data_producer(struct comp_dev *component,
+								  struct comp_buffer *producer)
+{
+	return producer->sink_list.next == &component->bsource_list ? NULL :
+	       list_item(producer->sink_list.next, struct comp_buffer, sink_list);
+}
+
+/**
+ * Get a pointer to a first comp_buffer object receiving data from the component
+ * The procedure will return NULL if there's no data consumers
+ */
+static inline struct comp_buffer *comp_dev_get_first_data_consumer(struct comp_dev *component)
+{
+	return list_is_empty(&component->bsink_list) ? NULL :
+	       list_first_item(&component->bsink_list, struct comp_buffer, source_list);
+}
+
+/**
+ * Get a pointer to a next comp_buffer object receiving data from the component
+ * The procedure will return NULL if there're no more data consumers
+ */
+static inline struct comp_buffer *comp_dev_get_next_data_consumer(struct comp_dev *component,
+								  struct comp_buffer *consumer)
+{
+	return consumer->source_list.next == &component->bsink_list ? NULL :
+			list_item(consumer->source_list.next, struct comp_buffer, source_list);
+}
+
+/*
+ * a macro for easy iteration through component's list of producers
+ */
+#define comp_dev_for_each_producer(_dev, _producer)			\
+	for (_producer = comp_dev_get_first_data_producer(_dev);	\
+	     _producer != NULL;						\
+	     _producer = comp_dev_get_next_data_producer(_dev, _producer))
+
+/*
+ * a macro for easy iteration through component's list of consumers
+ */
+#define comp_dev_for_each_consumer(_dev, _consumer)			\
+	for (_consumer = comp_dev_get_first_data_consumer(_dev);	\
+	     _consumer != NULL;						\
+	     _consumer = comp_dev_get_next_data_consumer(_dev, _consumer))
+
 /** @}*/
 
 /* Common helper function used internally by the component implementations
diff --git a/src/probe/probe.c b/src/probe/probe.c
index 721004555636..97f460edf26a 100644
--- a/src/probe/probe.c
+++ b/src/probe/probe.c
@@ -1061,13 +1061,11 @@ static bool probe_purpose_needs_ext_dma(uint32_t purpose)
 static struct comp_buffer *ipc4_get_buffer(struct ipc_comp_dev *dev, probe_point_id_t probe_point)
 {
 	struct comp_buffer *buf;
-	struct list_item *sink_list, *source_list;
 	unsigned int queue_id;
 
 	switch (probe_point.fields.type) {
 	case PROBE_TYPE_INPUT:
-		list_for_item(source_list, &dev->cd->bsource_list) {
-			buf = container_of(source_list, struct comp_buffer, sink_list);
+		comp_dev_for_each_producer(dev->cd, buf) {
 			queue_id = IPC4_SRC_QUEUE_ID(buf_get_id(buf));
 
 			if (queue_id == probe_point.fields.index)
@@ -1075,8 +1073,7 @@ static struct comp_buffer *ipc4_get_buffer(struct ipc_comp_dev *dev, probe_point
 		}
 		break;
 	case PROBE_TYPE_OUTPUT:
-		list_for_item(sink_list, &dev->cd->bsink_list) {
-			buf = container_of(sink_list, struct comp_buffer, source_list);
+		comp_dev_for_each_consumer(dev->cd, buf) {
 			queue_id = IPC4_SINK_QUEUE_ID(buf_get_id(buf));
 
 			if (queue_id == probe_point.fields.index)
diff --git a/src/samples/audio/detect_test.c b/src/samples/audio/detect_test.c
index 7a29337f20db..d7067243dad3 100644
--- a/src/samples/audio/detect_test.c
+++ b/src/samples/audio/detect_test.c
@@ -818,8 +818,7 @@ static int test_keyword_params(struct comp_dev *dev,
 	cd->sample_valid_bytes = params->sample_valid_bytes;
 
 	/* keyword components will only ever have 1 source */
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer,
-				  sink_list);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	channels = audio_stream_get_channels(&sourceb->stream);
 	frame_fmt = audio_stream_get_frm_fmt(&sourceb->stream);
 	rate = audio_stream_get_rate(&sourceb->stream);
@@ -895,8 +894,7 @@ static int test_keyword_copy(struct comp_dev *dev)
 	comp_dbg(dev, "test_keyword_copy()");
 
 	/* keyword components will only ever have 1 source */
-	source = list_first_item(&dev->bsource_list,
-				 struct comp_buffer, sink_list);
+	source = comp_dev_get_first_data_producer(dev);
 
 	if (!audio_stream_get_avail(&source->stream))
 		return PPL_STATUS_PATH_STOP;
diff --git a/src/samples/audio/smart_amp_test_ipc3.c b/src/samples/audio/smart_amp_test_ipc3.c
index 9de3bacb76d5..c69805a9d0bf 100644
--- a/src/samples/audio/smart_amp_test_ipc3.c
+++ b/src/samples/audio/smart_amp_test_ipc3.c
@@ -487,7 +487,6 @@ static int smart_amp_prepare(struct comp_dev *dev)
 {
 	struct smart_amp_data *sad = comp_get_drvdata(dev);
 	struct comp_buffer *source_buffer;
-	struct list_item *blist;
 	int ret;
 
 	comp_info(dev, "smart_amp_prepare()");
@@ -500,10 +499,7 @@ static int smart_amp_prepare(struct comp_dev *dev)
 		return PPL_STATUS_PATH_STOP;
 
 	/* searching for stream and feedback source buffers */
-	list_for_item(blist, &dev->bsource_list) {
-		source_buffer = container_of(blist, struct comp_buffer,
-					     sink_list);
-
+	comp_dev_for_each_producer(dev, source_buffer) {
 		/* FIXME: how often can this loop be run? */
 		if (source_buffer->source->ipc_config.type == SOF_COMP_DEMUX)
 			sad->feedback_buf = source_buffer;
@@ -511,8 +507,7 @@ static int smart_amp_prepare(struct comp_dev *dev)
 			sad->source_buf = source_buffer;
 	}
 
-	sad->sink_buf = list_first_item(&dev->bsink_list, struct comp_buffer,
-					source_list);
+	sad->sink_buf = comp_dev_get_first_data_consumer(dev);
 
 	sad->out_channels = audio_stream_get_channels(&sad->sink_buf->stream);
 
diff --git a/test/cmocka/src/audio/eq_fir/eq_fir_process.c b/test/cmocka/src/audio/eq_fir/eq_fir_process.c
index a847a4d34fb6..f281a31603d4 100644
--- a/test/cmocka/src/audio/eq_fir/eq_fir_process.c
+++ b/test/cmocka/src/audio/eq_fir/eq_fir_process.c
@@ -211,7 +211,7 @@ static void fill_source_s16(struct test_data *td, int frames_max)
 	int i;
 	int samples_processed = 0;
 
-	sb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sb = comp_dev_get_first_data_producer(dev);
 	ss = &sb->stream;
 	frames = MIN(audio_stream_get_free_frames(ss), frames_max);
 	samples = frames * audio_stream_get_channels(ss);
@@ -246,7 +246,7 @@ static void verify_sink_s16(struct test_data *td)
 	int samples;
 	int i;
 
-	sb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sb = comp_dev_get_first_data_consumer(dev);
 	ss = &sb->stream;
 	samples = mod->output_buffers[0].size >> 1;
 	for (i = 0; i < samples; i++) {
@@ -277,7 +277,7 @@ static void fill_source_s24(struct test_data *td, int frames_max)
 	int i;
 	int samples_processed = 0;
 
-	sb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sb = comp_dev_get_first_data_producer(dev);
 	ss = &sb->stream;
 	frames = MIN(audio_stream_get_free_frames(ss), frames_max);
 	samples = frames * audio_stream_get_channels(ss);
@@ -312,7 +312,7 @@ static void verify_sink_s24(struct test_data *td)
 	int samples;
 	int i;
 
-	sb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sb = comp_dev_get_first_data_consumer(dev);
 	ss = &sb->stream;
 	samples = mod->output_buffers[0].size >> 2;
 	for (i = 0; i < samples; i++) {
@@ -343,7 +343,7 @@ static void fill_source_s32(struct test_data *td, int frames_max)
 	int i;
 	int samples_processed = 0;
 
-	sb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sb = comp_dev_get_first_data_producer(dev);
 	ss = &sb->stream;
 	frames = MIN(audio_stream_get_free_frames(ss), frames_max);
 	samples = frames * audio_stream_get_channels(ss);
@@ -378,7 +378,7 @@ static void verify_sink_s32(struct test_data *td)
 	int samples;
 	int i;
 
-	sb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sb = comp_dev_get_first_data_consumer(dev);
 	ss = &sb->stream;
 	samples = mod->output_buffers[0].size >> 2;
 	for (i = 0; i < samples; i++) {
diff --git a/test/cmocka/src/audio/eq_iir/eq_iir_process.c b/test/cmocka/src/audio/eq_iir/eq_iir_process.c
index fa18cf5b0c3f..44aabf2d5619 100644
--- a/test/cmocka/src/audio/eq_iir/eq_iir_process.c
+++ b/test/cmocka/src/audio/eq_iir/eq_iir_process.c
@@ -210,7 +210,7 @@ static void fill_source_s16(struct test_data *td, int frames_max)
 	int i;
 	int samples_processed = 0;
 
-	sb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sb = comp_dev_get_first_data_producer(dev);
 	ss = &sb->stream;
 	frames = MIN(audio_stream_get_free_frames(ss), frames_max);
 	samples = frames * audio_stream_get_channels(ss);
@@ -245,7 +245,7 @@ static void verify_sink_s16(struct test_data *td)
 	int samples;
 	int i;
 
-	sb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sb = comp_dev_get_first_data_consumer(dev);
 	ss = &sb->stream;
 	samples = mod->output_buffers[0].size >> 1;
 	for (i = 0; i < samples; i++) {
@@ -273,7 +273,7 @@ static void fill_source_s24(struct test_data *td, int frames_max)
 	int i;
 	int samples_processed = 0;
 
-	sb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sb = comp_dev_get_first_data_producer(dev);
 	ss = &sb->stream;
 	frames = MIN(audio_stream_get_free_frames(ss), frames_max);
 	samples = frames * audio_stream_get_channels(ss);
@@ -308,7 +308,7 @@ static void verify_sink_s24(struct test_data *td)
 	int samples;
 	int i;
 
-	sb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sb = comp_dev_get_first_data_consumer(dev);
 	ss = &sb->stream;
 	samples = mod->output_buffers[0].size >> 2;
 	for (i = 0; i < samples; i++) {
@@ -336,7 +336,7 @@ static void fill_source_s32(struct test_data *td, int frames_max)
 	int i;
 	int samples_processed = 0;
 
-	sb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+	sb = comp_dev_get_first_data_producer(dev);
 	ss = &sb->stream;
 	frames = MIN(audio_stream_get_free_frames(ss), frames_max);
 	samples = frames * audio_stream_get_channels(ss);
@@ -371,7 +371,7 @@ static void verify_sink_s32(struct test_data *td)
 	int samples;
 	int i;
 
-	sb = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	sb = comp_dev_get_first_data_consumer(dev);
 	ss = &sb->stream;
 	samples = mod->output_buffers[0].size >> 2;
 	for (i = 0; i < samples; i++) {
diff --git a/tools/plugin/modules/alsa.c b/tools/plugin/modules/alsa.c
index 8d45597d6227..2d946529d05f 100644
--- a/tools/plugin/modules/alsa.c
+++ b/tools/plugin/modules/alsa.c
@@ -428,7 +428,7 @@ static int arecord_params(struct comp_dev *dev, struct sof_ipc_stream_params *pa
 	memcpy(&cd->params, params, sizeof(*params));
 
 	/* file component sink/source buffer period count */
-	buffer = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	buffer = comp_dev_get_first_data_consumer(dev);
 	buffer_reset_pos(buffer, NULL);
 
 	comp_dbg(dev, "prepare done ret = %d", ret);
@@ -460,8 +460,7 @@ static int aplay_params(struct comp_dev *dev, struct sof_ipc_stream_params *para
 	memcpy(&cd->params, params, sizeof(*params));
 
 	/* file component sink/source buffer period count */
-	buffer = list_first_item(&dev->bsource_list, struct comp_buffer,
-				 sink_list);
+	buffer = comp_dev_get_first_data_producer(dev);
 	buffer_reset_pos(buffer, NULL);
 
 	comp_dbg(dev, "prepare done ret = %d", ret);
@@ -529,7 +528,7 @@ static int arecord_copy(struct comp_dev *dev)
 	}
 
 	/* file component sink buffer */
-	buffer = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+	buffer = comp_dev_get_first_data_consumer(dev);
 	sink = &buffer->stream;
 	pos = sink->w_ptr;
 
@@ -587,8 +586,7 @@ static int aplay_copy(struct comp_dev *dev)
 	}
 
 	/* file component source buffer */
-	buffer = list_first_item(&dev->bsource_list, struct comp_buffer,
-				 sink_list);
+	buffer = comp_dev_get_first_data_producer(dev);
 	source = &buffer->stream;
 	pos = source->r_ptr;
 	avail = audio_stream_get_avail_frames(source);
diff --git a/tools/plugin/modules/shm.c b/tools/plugin/modules/shm.c
index 0e990bfe7cd0..63f9e002e0d2 100644
--- a/tools/plugin/modules/shm.c
+++ b/tools/plugin/modules/shm.c
@@ -247,8 +247,7 @@ static int shmread_copy(struct comp_dev *dev)
 	void *dest;
 
 	/* local SOF source buffer */
-	buffer = list_first_item(&dev->bsource_list, struct comp_buffer,
-				 sink_list);
+	buffer = comp_dev_get_first_data_producer(dev);
 	source = &buffer->stream;
 	rptr = source->r_ptr;
 
@@ -306,8 +305,7 @@ static int shmwrite_copy(struct comp_dev *dev)
 	void *src;
 
 	/* local SOF sink buffer */
-	buffer = list_first_item(&dev->bsink_list, struct comp_buffer,
-				 source_list);
+	buffer = comp_dev_get_first_data_consumer(dev);
 	sink = &buffer->stream;
 	wptr = sink->w_ptr;
 
diff --git a/tools/testbench/file.c b/tools/testbench/file.c
index 846d19c0fb06..bd3584155005 100644
--- a/tools/testbench/file.c
+++ b/tools/testbench/file.c
@@ -686,7 +686,7 @@ static int file_process(struct processing_module *mod,
 	switch (cd->fs.mode) {
 	case FILE_READ:
 		/* read PCM samples from file */
-		buffer = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+		buffer = comp_dev_get_first_data_consumer(dev);
 		sink = &buffer->stream;
 		frames = audio_stream_get_free_frames(sink);
 		frames = MIN(frames, cd->max_frames);
@@ -695,7 +695,7 @@ static int file_process(struct processing_module *mod,
 		break;
 	case FILE_WRITE:
 		/* write PCM samples into file */
-		buffer = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+		buffer = comp_dev_get_first_data_producer(dev);
 		source = &buffer->stream;
 		frames = audio_stream_get_avail_frames(source);
 		frames = MIN(frames, cd->max_frames);
@@ -735,10 +735,10 @@ static int file_prepare(struct processing_module *mod,
 	cd->max_frames = dev->frames;
 	switch (cd->fs.mode) {
 	case FILE_READ:
-		buffer = list_first_item(&dev->bsink_list, struct comp_buffer, source_list);
+		buffer = comp_dev_get_first_data_consumer(dev);
 		break;
 	case FILE_WRITE:
-		buffer = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+		buffer = comp_dev_get_first_data_producer(dev);
 		break;
 	default:
 		/* TODO: duplex mode */