ipc4: pipeline: Add notification pool and data process error notification

src/audio/pipeline/pipeline-schedule.c

@@ -372,8 +372,10 @@ static enum task_state dp_task_run(void *data)
 {
 	struct processing_module *mod = data;
 
-	module_process_sink_src(mod, mod->sources, mod->num_of_sources,
-				mod->sinks, mod->num_of_sinks);
+	int ret = module_process_sink_src(mod, mod->sources, mod->num_of_sources,
+					  mod->sinks, mod->num_of_sinks);
+	if (ret)
+		pipeline_comp_copy_error_notify(mod->dev, ret);
 
 	return SOF_TASK_STATE_RESCHEDULE;
 }

src/audio/pipeline/pipeline-stream.c

@@ -21,6 +21,11 @@
 #include <rtos/kernel.h>
 #include <sof/audio/module_adapter/module/generic.h>
 #include <sof/lib/cpu-clk-manager.h>
+#include <sof/ipc/notification_pool.h>
+
+#ifdef CONFIG_IPC_MAJOR_4
+#include <ipc4/notification.h>
+#endif
 
 #include <errno.h>
 #include <stdbool.h>
@@ -88,6 +93,20 @@ pipeline_should_report_enodata_on_trigger(struct comp_dev *rsrc,
 	return false;
 }
 
+void pipeline_comp_copy_error_notify(const struct comp_dev *component, int err)
+{
+#ifdef CONFIG_IPC_MAJOR_4
+	struct ipc_msg *notify;
+
+	notify = ipc_notification_pool_get(IPC4_RESOURCE_EVENT_SIZE);
+	if (!notify)
+		return;
+
+	process_data_error_notif_msg_init(notify, component->ipc_config.id, err);
+	ipc_msg_send(notify, notify->tx_data, false);
+#endif
+}
+
 static int pipeline_comp_copy(struct comp_dev *current,
 			      struct comp_buffer *calling_buf,
 			      struct pipeline_walk_context *ctx, int dir)
@@ -113,16 +132,23 @@ static int pipeline_comp_copy(struct comp_dev *current,
 	/* copy to downstream immediately */
 	if (dir == PPL_DIR_DOWNSTREAM) {
 		err = comp_copy(current);
-		if (err < 0 || err == PPL_STATUS_PATH_STOP)
+		if (err < 0) {
+			pipeline_comp_copy_error_notify(current, err);
+			return err;
+		}
+		if (err == PPL_STATUS_PATH_STOP)
 			return err;
 	}
 
 	err = pipeline_for_each_comp(current, ctx, dir);
 	if (err < 0 || err == PPL_STATUS_PATH_STOP)
 		return err;
 
-	if (dir == PPL_DIR_UPSTREAM)
+	if (dir == PPL_DIR_UPSTREAM) {
 		err = comp_copy(current);
+		if (err < 0)
+			pipeline_comp_copy_error_notify(current, err);
+	}
 
 	return err;
 }

src/include/ipc4/notification.h

@@ -25,6 +25,7 @@
 
 #include <stdint.h>
 #include <ipc/header.h>
+#include <sof/ipc/msg.h>
 #include <sof/compiler_attributes.h>
 
 /* ipc4 notification msg */
@@ -37,7 +38,7 @@ enum sof_ipc4_notification_type {
 	SOF_IPC4_FW_AUD_CLASS_RESULT		= 9,
 	SOF_IPC4_EXCEPTION_CAUGHT		= 10,
 	SOF_IPC4_MODULE_NOTIFICATION		= 12,
-	SOF_IPC4_UAOL_RSVD_		= 13,
+	SOF_IPC4_UAOL_EVENT			= 13,
 	SOF_IPC4_PROBE_DATA_AVAILABLE		= 14,
 	SOF_IPC4_WATCHDOG_TIMEOUT		= 15,
 	SOF_IPC4_MANAGEMENT_SERVICE		= 16,
@@ -117,8 +118,6 @@ enum sof_ipc4_resource_type {
 		((notif_type) << (SOF_IPC4_GLB_NOTIFY_TYPE_SHIFT) | \
 		((SOF_IPC4_GLB_NOTIFICATION) << (SOF_IPC4_GLB_NOTIFY_MSG_TYPE_SHIFT)))
 
-#endif
-
 /**
  * \brief IPC MAJOR 4 notification header. All IPC4 notifications use this header.
  */
@@ -231,6 +230,18 @@ static inline void ipc4_notification_watchdog_init(struct ipc4_watchdog_timeout_
 	notif->primary.r.msg_tgt = SOF_IPC4_MESSAGE_TARGET_FW_GEN_MSG;
 }
 
+/**
+ * \brief This notification is sent by a shim of module instance on error raised by data processing
+ * function.
+ *
+ * In case of 3rd party IP error_code is set to its native error code returned by the 3rd party
+ * library.
+ */
+struct ipc4_process_data_error_event_data {
+	/* Error code returned by data processing function */
+	uint32_t error_code;
+};
+
 /**
  * \brief Input data payload is reserved field in parent technical spec which can be easily
  * extendable if needed by specific resource event types in the future. For backward compatibility
@@ -239,6 +250,8 @@ static inline void ipc4_notification_watchdog_init(struct ipc4_watchdog_timeout_
 union ipc4_resource_event_data {
 	/* Raw data */
 	uint32_t dws[6];
+	/* Process Data Error Data (res type = MODULE_INSTANCE) */
+	struct ipc4_process_data_error_event_data process_data_error;
 };
 
 struct ipc4_resource_event_data_notification {
@@ -255,3 +268,10 @@ struct ipc4_resource_event_data_notification {
 	/* Detailed event data */
 	union ipc4_resource_event_data event_data;
 } __packed __aligned(8);
+
+#define IPC4_RESOURCE_EVENT_SIZE	sizeof(struct ipc4_resource_event_data_notification)
+
+void process_data_error_notif_msg_init(struct ipc_msg *msg, uint32_t resource_id,
+				       uint32_t error_code);
+
+#endif /* __IPC4_NOTIFICATION_H__ */

src/include/sof/audio/pipeline.h

@@ -431,4 +431,11 @@ void pipeline_xrun(struct pipeline *p, struct comp_dev *dev, int32_t bytes);
  */
 int pipeline_xrun_set_limit(struct pipeline *p, uint32_t xrun_limit_usecs);
 
+/**
+ * \brief Sends an ipc notification that an error occurred in the module's processing function.
+ * \param[in] component The component in which the error occurred.
+ * \param[in] error_code Error code.
+ */
+void pipeline_comp_copy_error_notify(const struct comp_dev *component, int error_code);
+
 #endif /* __SOF_AUDIO_PIPELINE_H__ */

src/include/sof/ipc/msg.h

@@ -36,6 +36,7 @@ struct ipc_msg {
 	uint32_t tx_size;	/* payload size in bytes */
 	void *tx_data;		/* pointer to payload data, must be in a non-cached memory */
 	struct list_item list;
+	void (*callback)(struct ipc_msg *msg); /* Function called after sending the message */
 };
 
 /**

src/include/sof/ipc/notification_pool.h

@@ -0,0 +1,26 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ *
+ * Copyright(c) 2025 Intel Corporation. All rights reserved.
+ *
+ * Author: Adrian Warecki <adrian.warecki@intel.com>
+ */
+
+#ifndef __SOF_IPC_NOTIFICATION_POOL_H__
+#define __SOF_IPC_NOTIFICATION_POOL_H__
+
+#include <stdint.h>
+#include <sof/ipc/msg.h>
+
+/**
+ * @brief Retrieves an IPC notification message from the pool.
+ *
+ * This function retrieves and returns an IPC notification message
+ * of the specified size from the notification pool. The size of the
+ * message is limited by the maximum size available in the pool.
+ *
+ * @param size The size of the IPC message to retrieve.
+ * @return A pointer to the retrieved IPC message, or NULL if retrieval fails.
+ */
+struct ipc_msg *ipc_notification_pool_get(size_t size);
+
+#endif /* __SOF_IPC_NOTIFICATION_POOL_H__ */

src/ipc/CMakeLists.txt

@@ -10,6 +10,7 @@ endif()
 add_local_sources(sof
 	ipc-common.c
 	ipc-helper.c
+	notification_pool.c
 )
 
 is_zephyr(it_is)

src/ipc/ipc-common.c

@@ -162,6 +162,9 @@ void ipc_send_queued_msg(void)
 	if (ipc_platform_send_msg(msg) == 0) {
 		/* Remove the message from the list if it has been successfully sent. */
 		list_item_del(&msg->list);
+		/* Invoke a callback to notify that the message has been sent. */
+		if (msg->callback)
+			msg->callback(msg);
 #ifdef CONFIG_SOF_TELEMETRY_IO_PERFORMANCE_MEASUREMENTS
 		/* Increment performance counters */
 		io_perf_monitor_update_data(ipc->io_perf_out_msg_count, 1);

src/ipc/ipc4/notification.c

@@ -3,27 +3,47 @@
  * Copyright(c) 2023 Intel Corporation. All rights reserved.
  *
  * Author: Piotr Makaruk <piotr.makaruk@intel.com>
+ *	   Adrian Warecki <adrian.warecki@intel.com>
  */
 
 #include <sof/common.h>
+#include <sof/ipc/msg.h>
 #include <stdbool.h>
 #include <ipc4/notification.h>
 
-#if CONFIG_XRUN_NOTIFICATIONS_ENABLE
-void xrun_notif_msg_init(struct ipc_msg *msg_xrun, uint32_t resource_id, uint32_t event_type)
+static void resource_notif_header_init(struct ipc_msg *msg)
 {
-	struct ipc4_resource_event_data_notification *notif_data = msg_xrun->tx_data;
+	struct ipc4_resource_event_data_notification *notif_data = msg->tx_data;
 	union ipc4_notification_header header;
 
 	header.r.notif_type = SOF_IPC4_NOTIFY_RESOURCE_EVENT;
 	header.r.type = SOF_IPC4_GLB_NOTIFICATION;
 	header.r.rsp = SOF_IPC4_MESSAGE_DIR_MSG_REQUEST;
 	header.r.msg_tgt = SOF_IPC4_MESSAGE_TARGET_FW_GEN_MSG;
-	msg_xrun->header = header.dat;
+	msg->header = header.dat;
+	memset(&notif_data->event_data, 0, sizeof(notif_data->event_data));
+}
+
+#if CONFIG_XRUN_NOTIFICATIONS_ENABLE
+void xrun_notif_msg_init(struct ipc_msg *msg_xrun, uint32_t resource_id, uint32_t event_type)
+{
+	struct ipc4_resource_event_data_notification *notif_data = msg_xrun->tx_data;
 
+	resource_notif_header_init(msg_xrun);
 	notif_data->resource_id = resource_id;
 	notif_data->event_type = event_type;
 	notif_data->resource_type = SOF_IPC4_GATEWAY;
-	memset(&notif_data->event_data, 0, sizeof(notif_data->event_data));
 }
 #endif
+
+void process_data_error_notif_msg_init(struct ipc_msg *msg, uint32_t resource_id,
+				       uint32_t error_code)
+{
+	struct ipc4_resource_event_data_notification *notif_data = msg->tx_data;
+
+	resource_notif_header_init(msg);
+	notif_data->resource_id = resource_id;
+	notif_data->event_type = SOF_IPC4_PROCESS_DATA_ERROR;
+	notif_data->resource_type = SOF_IPC4_MODULE_INSTANCE;
+	notif_data->event_data.process_data_error.error_code = error_code;
+}

src/ipc/notification_pool.c

@@ -0,0 +1,100 @@
+// SPDX-License-Identifier: BSD-3-Clause
+//
+// Copyright(c) 2025 Intel Corporation. All rights reserved.
+//
+// Author: Adrian Warecki <adrian.warecki@intel.com>
+//
+
+#include <stdint.h>
+#include <sof/common.h>
+#include <sof/list.h>
+#include <sof/ipc/notification_pool.h>
+
+#define NOTIFICATION_POOL_MAX_PAYLOAD_SIZE	40	/* IPC4 Resource Event needs 10dw */
+#define NOTIFICATION_POOL_MAX_DEPTH		8	/* Maximum number of notifications
+							 * in the pool
+							 */
+
+LOG_MODULE_REGISTER(notification_pool, CONFIG_SOF_LOG_LEVEL);
+
+SOF_DEFINE_REG_UUID(notification_pool);
+
+DECLARE_TR_CTX(notif_tr, SOF_UUID(notification_pool_uuid), LOG_LEVEL_INFO);
+
+struct ipc_notif_pool_item {
+	struct ipc_msg msg;
+	uint32_t payload[SOF_DIV_ROUND_UP(NOTIFICATION_POOL_MAX_PAYLOAD_SIZE, sizeof(uint32_t))];
+};
+
+static struct list_item pool_free_list = LIST_INIT(pool_free_list);
+static struct k_spinlock pool_free_list_lock;
+static int pool_depth;
+
+static void ipc_notif_free(struct ipc_msg *msg)
+{
+	struct ipc_notif_pool_item *item = container_of(msg, struct ipc_notif_pool_item, msg);
+	k_spinlock_key_t key;
+
+	key = k_spin_lock(&pool_free_list_lock);
+	list_item_append(&item->msg.list, &pool_free_list);
+	k_spin_unlock(&pool_free_list_lock, key);
+}
+
+static struct ipc_msg *ipc_notif_new(size_t size)
+{
+	struct ipc_notif_pool_item *item;
+
+	item = rzalloc(SOF_MEM_ZONE_RUNTIME_SHARED, 0, SOF_MEM_CAPS_RAM, sizeof(*item));
+	if (!item) {
+		tr_err(&notif_tr, "Unable to allocate memory for notification message");
+		return NULL;
+	}
+
+	list_init(&item->msg.list);
+	item->msg.tx_data = item->payload;
+	item->msg.tx_size = size;
+	item->msg.callback = ipc_notif_free;
+	return &item->msg;
+}
+
+struct ipc_msg *ipc_notification_pool_get(size_t size)
+{
+	struct ipc_notif_pool_item *item;
+	k_spinlock_key_t key;
+	struct ipc_msg *new_msg;
+
+	if (size > NOTIFICATION_POOL_MAX_PAYLOAD_SIZE) {
+		tr_err(&notif_tr, "Requested size %zu exceeds maximum payload size %u",
+		       size, NOTIFICATION_POOL_MAX_PAYLOAD_SIZE);
+		return NULL;
+	}
+
+	/* check if we have a free message */
+	key = k_spin_lock(&pool_free_list_lock);
+	if (list_is_empty(&pool_free_list)) {
+		/* allocate a new message */
+		if (pool_depth >= NOTIFICATION_POOL_MAX_DEPTH) {
+			k_spin_unlock(&pool_free_list_lock, key);
+			tr_err(&notif_tr, "Pool depth exceeded");
+			return NULL;
+		}
+		++pool_depth;
+		k_spin_unlock(&pool_free_list_lock, key);
+
+		new_msg = ipc_notif_new(size);
+		if (!new_msg) {
+			key = k_spin_lock(&pool_free_list_lock);
+			--pool_depth;
+			k_spin_unlock(&pool_free_list_lock, key);
+		}
+		return new_msg;
+	}
+
+	/* take the first free message */
+	item = list_first_item(&pool_free_list, struct ipc_notif_pool_item, msg.list);
+	list_item_del(&item->msg.list);
+	k_spin_unlock(&pool_free_list_lock, key);
+
+	item->msg.tx_size = size;
+	return &item->msg;
+}

uuid-registry.txt

@@ -172,3 +172,4 @@ d944281a-afe9-4695-a043d7f62b89538e waves
 2B79E4F3-4675-F649-89DF3BC194A91AEB brngup
 D406D134-C3C1-402C-8AEC6821C0C2B0E6 cold
 c51dc642-a2e1-48df-a490e2748cb6363e tflmcly
+f36BF24B-9AAF-83f4-8677E072E8AEADB7 notification_pool