diff --git a/src/audio/mixer/mixer.c b/src/audio/mixer/mixer.c
index e037852b0..c04bc7dff 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,15 +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 = 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;
 
 		/*
@@ -233,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/mux/mux.c b/src/audio/mux/mux.c
index 9b2e882f9..2b8f6c471 100644
--- a/src/audio/mux/mux.c
+++ b/src/audio/mux/mux.c
@@ -287,7 +287,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;
@@ -298,8 +297,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);
@@ -330,8 +328,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++;
@@ -342,7 +339,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;
@@ -350,9 +347,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 */
@@ -432,9 +427,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
@@ -444,8 +436,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_ipc4.c b/src/audio/mux/mux_ipc4.c
index bd0071901..6d183d697 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;
@@ -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/smart_amp/smart_amp.c b/src/audio/smart_amp/smart_amp.c
index 80100022e..e55d9cdb2 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
diff --git a/src/probe/probe.c b/src/probe/probe.c
index 17b9b630a..97f460edf 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 *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)
diff --git a/src/samples/audio/smart_amp_test_ipc3.c b/src/samples/audio/smart_amp_test_ipc3.c
index 5406c41b1..c69805a9d 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;