diff --git a/src/audio/copier/copier.c b/src/audio/copier/copier.c
index 7c8d53726d20..248879de015c 100644
--- a/src/audio/copier/copier.c
+++ b/src/audio/copier/copier.c
@@ -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) {
@@ -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/crossover/crossover.c b/src/audio/crossover/crossover.c
index 9fffda1858c5..a8578fee6a94 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;
 
@@ -546,8 +543,7 @@ static int crossover_prepare(struct processing_module *mod,
 	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..2aa1a1ba5412 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()");
@@ -122,8 +121,7 @@ void crossover_params(struct processing_module *mod)
 	sourceb = list_first_item(&dev->bsource_list, struct comp_buffer, sink_list);
 	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-zephyr.c b/src/audio/dai-zephyr.c
index d2e672e700c1..52d9a54bf9c4 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;
@@ -1592,17 +1588,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 &&
@@ -1613,8 +1608,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);
 
 		/*
@@ -1630,11 +1623,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/kpb.c b/src/audio/kpb.c
index 542984e88867..bd37d0ba3e92 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)
diff --git a/src/audio/mux/mux.c b/src/audio/mux/mux.c
index 9ac9365297e6..f3fd44ddb548 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 */
diff --git a/src/audio/selector/selector.c b/src/audio/selector/selector.c
index 506853043420..1e389ca37663 100644
--- a/src/audio/selector/selector.c
+++ b/src/audio/selector/selector.c
@@ -644,7 +644,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)
@@ -664,10 +663,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);
diff --git a/src/probe/probe.c b/src/probe/probe.c
index 721004555636..17b9b630a7b2 100644
--- a/src/probe/probe.c
+++ b/src/probe/probe.c
@@ -1061,7 +1061,7 @@ 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;
+	struct list_item *source_list;
 	unsigned int queue_id;
 
 	switch (probe_point.fields.type) {
@@ -1075,8 +1075,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)