diff --git a/src/audio/aria/aria.c b/src/audio/aria/aria.c
index c375ea8bb61f..2c703fc277bc 100644
--- a/src/audio/aria/aria.c
+++ b/src/audio/aria/aria.c
@@ -178,7 +178,7 @@ 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 = comp_dev_get_first_data_consumer(dev);
diff --git a/src/audio/asrc/asrc.c b/src/audio/asrc/asrc.c
index 9adcd18e8baa..8300a52250c8 100644
--- a/src/audio/asrc/asrc.c
+++ b/src/audio/asrc/asrc.c
@@ -396,8 +396,7 @@ static int asrc_params(struct processing_module *mod)
 		return -EINVAL;
 	}
 
-	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer,
-				  sink_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 */
@@ -469,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;
 
@@ -544,8 +543,7 @@ 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);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* get source data format and period bytes */
@@ -794,8 +792,7 @@ 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);
+	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);
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 248879de015c..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,
@@ -561,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);
diff --git a/src/audio/copier/copier_ipcgtw.c b/src/audio/copier/copier_ipcgtw.c
index 81d39d3bdfa6..604d9fecced4 100644
--- a/src/audio/copier/copier_ipcgtw.c
+++ b/src/audio/copier/copier_ipcgtw.c
@@ -82,7 +82,7 @@ static inline struct comp_buffer *get_buffer(struct comp_dev *dev)
 
 	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 a8578fee6a94..ffeef13aae1f 100644
--- a/src/audio/crossover/crossover.c
+++ b/src/audio/crossover/crossover.c
@@ -536,7 +536,7 @@ 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);
diff --git a/src/audio/crossover/crossover_ipc4.c b/src/audio/crossover/crossover_ipc4.c
index 2aa1a1ba5412..c928a5945034 100644
--- a/src/audio/crossover/crossover_ipc4.c
+++ b/src/audio/crossover/crossover_ipc4.c
@@ -118,7 +118,7 @@ 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);
 
 	comp_dev_for_each_consumer(dev, sinkb) {
diff --git a/src/audio/dai-legacy.c b/src/audio/dai-legacy.c
index 10c1aeb586a2..189926f6f3d9 100644
--- a/src/audio/dai-legacy.c
+++ b/src/audio/dai-legacy.c
@@ -493,9 +493,7 @@ 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 = comp_dev_get_first_data_consumer(dev);
 
diff --git a/src/audio/dai-zephyr.c b/src/audio/dai-zephyr.c
index 52d9a54bf9c4..132cc43f1453 100644
--- a/src/audio/dai-zephyr.c
+++ b/src/audio/dai-zephyr.c
@@ -848,8 +848,7 @@ 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 = comp_dev_get_first_data_consumer(dev);
 
@@ -1519,9 +1518,7 @@ 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 = comp_dev_get_first_data_consumer(dev);
 
diff --git a/src/audio/dcblock/dcblock.c b/src/audio/dcblock/dcblock.c
index 20b051eb0f63..de8a958733e5 100644
--- a/src/audio/dcblock/dcblock.c
+++ b/src/audio/dcblock/dcblock.c
@@ -199,7 +199,7 @@ 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);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* get source data format */
diff --git a/src/audio/dcblock/dcblock_ipc4.c b/src/audio/dcblock/dcblock_ipc4.c
index c9da0d6f902c..534510841b85 100644
--- a/src/audio/dcblock/dcblock_ipc4.c
+++ b/src/audio/dcblock/dcblock_ipc4.c
@@ -64,7 +64,7 @@ void dcblock_params(struct processing_module *mod)
 	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 eda040d6b869..f02c8ff2a4fe 100644
--- a/src/audio/drc/drc.c
+++ b/src/audio/drc/drc.c
@@ -306,7 +306,7 @@ static void drc_params(struct processing_module *mod)
 	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,7 +329,7 @@ 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);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* get source data format */
diff --git a/src/audio/eq_fir/eq_fir.c b/src/audio/eq_fir/eq_fir.c
index 8587686b3021..0fb40145aa97 100644
--- a/src/audio/eq_fir/eq_fir.c
+++ b/src/audio/eq_fir/eq_fir.c
@@ -424,7 +424,7 @@ 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);
+	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);
diff --git a/src/audio/eq_fir/eq_fir_ipc4.c b/src/audio/eq_fir/eq_fir_ipc4.c
index c23c00f56d52..eec4165b748b 100644
--- a/src/audio/eq_fir/eq_fir_ipc4.c
+++ b/src/audio/eq_fir/eq_fir_ipc4.c
@@ -59,7 +59,7 @@ 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 = comp_dev_get_first_data_consumer(dev);
diff --git a/src/audio/eq_iir/eq_iir.c b/src/audio/eq_iir/eq_iir.c
index 10d0cbeb3c16..7e2e54df7df6 100644
--- a/src/audio/eq_iir/eq_iir.c
+++ b/src/audio/eq_iir/eq_iir.c
@@ -193,7 +193,7 @@ 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);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	sinkb = comp_dev_get_first_data_consumer(dev);
 	eq_iir_set_alignment(&sourceb->stream, &sinkb->stream);
 
diff --git a/src/audio/eq_iir/eq_iir_ipc3.c b/src/audio/eq_iir/eq_iir_ipc3.c
index 2ce335a0cd27..7ccc375021fe 100644
--- a/src/audio/eq_iir/eq_iir_ipc3.c
+++ b/src/audio/eq_iir/eq_iir_ipc3.c
@@ -274,8 +274,7 @@ 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);
+	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
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 94a39296f264..73fa30e1b61c 100644
--- a/src/audio/host-legacy.c
+++ b/src/audio/host-legacy.c
@@ -714,9 +714,7 @@ int host_common_params(struct host_data *hd, struct comp_dev *dev,
 	if (dev->direction == SOF_IPC_STREAM_PLAYBACK)
 		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 cecf9d1547b5..d3bc506d2fbf 100644
--- a/src/audio/host-zephyr.c
+++ b/src/audio/host-zephyr.c
@@ -795,9 +795,7 @@ int host_common_params(struct host_data *hd, struct comp_dev *dev,
 	if (params->direction == SOF_IPC_STREAM_PLAYBACK)
 		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 bd37d0ba3e92..6e83747831eb 100644
--- a/src/audio/kpb.c
+++ b/src/audio/kpb.c
@@ -1182,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 a2f88850991d..3a56f2f95b46 100644
--- a/src/audio/mfcc/mfcc.c
+++ b/src/audio/mfcc/mfcc.c
@@ -192,7 +192,7 @@ 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);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* get source data format */
diff --git a/src/audio/module_adapter/module/waves/waves.c b/src/audio/module_adapter/module/waves/waves.c
index 7569873a2ac0..8ea54c8e21ed 100644
--- a/src/audio/module_adapter/module/waves/waves.c
+++ b/src/audio/module_adapter/module/waves/waves.c
@@ -216,8 +216,7 @@ static int waves_effect_allocate(struct processing_module *mod)
 static int waves_effect_check(struct comp_dev *dev)
 {
 	struct comp_buffer *sink = comp_dev_get_first_data_consumer(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);
 	const struct audio_stream *src_fmt = &source->stream;
 	const struct audio_stream *snk_fmt = &sink->stream;
 
@@ -283,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/mux/mux_generic.c b/src/audio/mux/mux_generic.c
index 2bf3afe010cf..991b79097fec 100644
--- a/src/audio/mux/mux_generic.c
+++ b/src/audio/mux/mux_generic.c
@@ -549,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/rtnr/rtnr.c b/src/audio/rtnr/rtnr.c
index 5e4bd2cb7e39..4ac99d1b35f9 100644
--- a/src/audio/rtnr/rtnr.c
+++ b/src/audio/rtnr/rtnr.c
@@ -781,7 +781,7 @@ 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 = comp_dev_get_first_data_consumer(dev);
@@ -817,7 +817,7 @@ static int rtnr_prepare(struct processing_module *mod,
 	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 1e389ca37663..a14e584007bb 100644
--- a/src/audio/selector/selector.c
+++ b/src/audio/selector/selector.c
@@ -90,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");
@@ -353,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);
 
@@ -383,8 +381,7 @@ 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);
+	source = comp_dev_get_first_data_producer(dev);
 	sink = comp_dev_get_first_data_consumer(dev);
 
 	if (!audio_stream_get_avail(&source->stream))
@@ -431,8 +428,7 @@ 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);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	/* get source data format and period bytes */
@@ -677,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);
@@ -712,7 +708,7 @@ static int selector_verify_params(struct processing_module *mod,
 		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);
 
@@ -837,7 +833,7 @@ 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);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	sinkb = comp_dev_get_first_data_consumer(dev);
 
 	audio_stream_set_align(4, 1, &sourceb->stream);
diff --git a/src/audio/tdfb/tdfb.c b/src/audio/tdfb/tdfb.c
index 7b5202adeecb..f4821623966a 100644
--- a/src/audio/tdfb/tdfb.c
+++ b/src/audio/tdfb/tdfb.c
@@ -739,7 +739,7 @@ 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);
+	sourceb = comp_dev_get_first_data_producer(dev);
 	sinkb = comp_dev_get_first_data_consumer(dev);
 	tdfb_set_alignment(&sourceb->stream, &sinkb->stream);
 
diff --git a/src/audio/tdfb/tdfb_ipc4.c b/src/audio/tdfb/tdfb_ipc4.c
index ad262fedd7e0..7a16328d952d 100644
--- a/src/audio/tdfb/tdfb_ipc4.c
+++ b/src/audio/tdfb/tdfb_ipc4.c
@@ -198,7 +198,7 @@ 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 = comp_dev_get_first_data_consumer(dev);
diff --git a/src/audio/tone.c b/src/audio/tone.c
index 5906f3ac9e71..d289c12d5f57 100644
--- a/src/audio/tone.c
+++ b/src/audio/tone.c
@@ -428,8 +428,7 @@ 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 = comp_dev_get_first_data_consumer(dev);
 
@@ -673,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 e94bd8822671..c63b1cfde724 100644
--- a/src/audio/volume/volume.c
+++ b/src/audio/volume/volume.c
@@ -685,8 +685,7 @@ static int volume_prepare(struct processing_module *mod,
 
 	/* volume component will only ever have 1 sink and source buffer */
 	sinkb = comp_dev_get_first_data_consumer(dev);
-	sourceb = list_first_item(&dev->bsource_list,
-				  struct comp_buffer, sink_list);
+	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 f02ee74109a1..2ae1d362a787 100644
--- a/src/audio/volume/volume_ipc3.c
+++ b/src/audio/volume/volume_ipc3.c
@@ -39,8 +39,7 @@ 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 = comp_dev_get_first_data_consumer(dev);
 
diff --git a/src/audio/volume/volume_ipc4.c b/src/audio/volume/volume_ipc4.c
index b495c6868666..9457b9f57359 100644
--- a/src/audio/volume/volume_ipc4.c
+++ b/src/audio/volume/volume_ipc4.c
@@ -416,7 +416,7 @@ static int volume_params(struct processing_module *mod)
 	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/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/test/cmocka/src/audio/eq_fir/eq_fir_process.c b/test/cmocka/src/audio/eq_fir/eq_fir_process.c
index 22f2a9103c64..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);
@@ -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);
@@ -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);
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 b22d14b396e9..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);
@@ -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);
@@ -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);
diff --git a/tools/plugin/modules/alsa.c b/tools/plugin/modules/alsa.c
index a3543260b721..2d946529d05f 100644
--- a/tools/plugin/modules/alsa.c
+++ b/tools/plugin/modules/alsa.c
@@ -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);
@@ -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 c9aaa5e3b7e6..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;
 
diff --git a/tools/testbench/file.c b/tools/testbench/file.c
index b70d04cf1934..957ae0ba0091 100644
--- a/tools/testbench/file.c
+++ b/tools/testbench/file.c
@@ -706,7 +706,7 @@ static int file_params(struct comp_dev *dev,
 		periods = dev->ipc_config.periods_sink;
 		break;
 	case FILE_WRITE:
-		buffer = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
+		buffer = comp_dev_get_first_data_producer(dev);
 		periods = dev->ipc_config.periods_source;
 		break;
 	default:
@@ -829,8 +829,7 @@ static int file_copy(struct comp_dev *dev)
 		break;
 	case FILE_WRITE:
 		/* file component source buffer */
-		buffer = list_first_item(&dev->bsource_list,
-					 struct comp_buffer, sink_list);
+		buffer = comp_dev_get_first_data_producer(dev);
 
 		/* test source has enough free frames */
 		src_frames = audio_stream_get_avail_frames(&buffer->stream);