ipc: icmsg: rx nocopy feature

This patch adds an optional nocopy feature for RX message to the icmsg
IPC library. The nocopy feature can be enabled using project
configuration.

If this feature is not used by icmsg users it is recommended to disable it
to reduce memory usage and improve run-time performance.

Signed-off-by: Hubert Miś <hubert.mis@gmail.com>

include/zephyr/ipc/icmsg.h

@@ -41,6 +41,12 @@ struct icmsg_data_t {
 	struct k_work_delayable notify_work;
 	struct k_work mbox_work;
 	atomic_t state;
+	uint8_t rx_buffer[CONFIG_IPC_SERVICE_ICMSG_CB_BUF_SIZE] __aligned(4);
+
+	/* No-copy */
+#ifdef CONFIG_IPC_SERVICE_ICMSG_NOCOPY_RX
+	atomic_t rx_buffer_held;
+#endif
 };
 
 /** @brief Initialize an icmsg instance
@@ -134,6 +140,50 @@ int icmsg_send(const struct icmsg_config_t *conf,
 	       struct icmsg_data_t *dev_data,
 	       const void *msg, size_t len);
 
+#ifdef CONFIG_IPC_SERVICE_ICMSG_NOCOPY_RX
+/** @brief Hold RX buffer to be used outside of the received callback.
+ *
+ *  @param[in] conf Structure containing configuration parameters for the icmsg
+ *                  instance being created.
+ *  @param[inout] dev_data Structure containing run-time data used by the icmsg
+ *                         instance. The structure is initialized with
+ *                         @ref icmsg_init and its content must be preserved
+ *                         while the icmsg instance is active.
+ *  @param[in] data Pointer to the buffer to be held.
+ *
+ *  @retval 0 on success.
+ *  @retval -EBUSY when the instance has not finished handshake with the remote
+ *                 instance.
+ *  @retval -EINVAL when the @p data argument does not point to a valid RX
+ *                  buffer.
+ *  @retval -EALREADY when the buffer is already held.
+ */
+int icmsg_hold_rx_buffer(const struct icmsg_config_t *conf,
+			 struct icmsg_data_t *dev_data,
+			 const void *data);
+
+/** @brief Release RX buffer for future use.
+ *
+ *  @param[in] conf Structure containing configuration parameters for the icmsg
+ *                  instance being created.
+ *  @param[inout] dev_data Structure containing run-time data used by the icmsg
+ *                         instance. The structure is initialized with
+ *                         @ref icmsg_init and its content must be preserved
+ *                         while the icmsg instance is active.
+ *  @param[in] data Pointer to the buffer to be released.
+ *
+ *  @retval 0 on success.
+ *  @retval -EBUSY when the instance has not finished handshake with the remote
+ *                 instance.
+ *  @retval -EINVAL when the @p data argument does not point to a valid RX
+ *                  buffer.
+ *  @retval -EALREADY when the buffer is not held.
+ */
+int icmsg_release_rx_buffer(const struct icmsg_config_t *conf,
+			    struct icmsg_data_t *dev_data,
+			    const void *data);
+#endif
+
 /** @brief Clear memory in TX buffer.
  *
  *  This function is intended to be called at an early stage of boot process,

include/zephyr/ipc/ipc_service.h

@@ -433,7 +433,7 @@ int ipc_service_send_nocopy(struct ipc_ept *ept, const void *data, size_t len);
  *  using the @ref ipc_service_release_rx_buffer function.
  *
  *  @param[in] ept Registered endpoint by @ref ipc_service_register_endpoint.
- *  @param[in] data Pointer to the RX buffer to release.
+ *  @param[in] data Pointer to the RX buffer to hold.
  *
  *  @retval -EIO when no backend is registered or release hook is missing from
  *		 backend.

subsys/ipc/ipc_service/lib/Kconfig.icmsg

@@ -11,6 +11,13 @@ config IPC_SERVICE_ICMSG_CB_BUF_SIZE
 	  data bigger than some size, you can safely change this option to
 	  reduce RAM consumption in your application.
 
+config IPC_SERVICE_ICMSG_NOCOPY_RX
+	bool
+	depends on IPC_SERVICE_ICMSG
+	help
+	  Enable nocopy feature for the icmsg library that might be used by
+	  backends based on icmsg.
+
 # The Icmsg library in its simplicity requires the system workqueue to execute
 # at a cooperative priority.
 config SYSTEM_WORKQUEUE_PRIORITY

subsys/ipc/ipc_service/lib/icmsg.c

@@ -11,13 +11,15 @@
 #include <zephyr/sys/atomic.h>
 #include <zephyr/sys/spsc_pbuf.h>
 
-#define CB_BUF_SIZE	CONFIG_IPC_SERVICE_ICMSG_CB_BUF_SIZE
+#define RX_BUF_SIZE	CONFIG_IPC_SERVICE_ICMSG_CB_BUF_SIZE
 #define BOND_NOTIFY_REPEAT_TO K_MSEC(CONFIG_IPC_SERVICE_ICMSG_BOND_NOTIFY_REPEAT_TO_MS)
+#define BUFFER_RELEASED 0
+#define BUFFER_HELD     1
 
 static const uint8_t magic[] = {0x45, 0x6d, 0x31, 0x6c, 0x31, 0x4b,
 				0x30, 0x72, 0x6e, 0x33, 0x6c, 0x69, 0x34};
-BUILD_ASSERT(sizeof(magic) <= CB_BUF_SIZE);
+BUILD_ASSERT(sizeof(magic) <= RX_BUF_SIZE);
-BUILD_ASSERT(CB_BUF_SIZE <= UINT16_MAX);
+BUILD_ASSERT(RX_BUF_SIZE <= UINT16_MAX);
 
 static int mbox_deinit(const struct icmsg_config_t *conf,
 		       struct icmsg_data_t *dev_data)
@@ -59,19 +61,72 @@ static void notify_process(struct k_work *item)
 	}
 }
 
+static bool is_endpoint_ready(struct icmsg_data_t *dev_data)
+{
+	return atomic_get(&dev_data->state) == ICMSG_STATE_READY;
+}
+
+static bool is_rx_buffer_free(struct icmsg_data_t *dev_data)
+{
+#ifdef CONFIG_IPC_SERVICE_ICMSG_NOCOPY_RX
+	return atomic_get(&dev_data->rx_buffer_held) == BUFFER_RELEASED;
+#else
+	return true;
+#endif
+}
+
+static bool is_rx_data_available(struct icmsg_data_t *dev_data)
+{
+	int len = spsc_pbuf_read(dev_data->rx_ib, NULL, 0);
+
+	return len > 0;
+}
+
+static void submit_mbox_work(struct icmsg_data_t *dev_data)
+{
+	if (k_work_submit(&dev_data->mbox_work) < 0) {
+		/* The mbox processing work is never canceled.
+		 * The negative error code should never be seen.
+		 */
+		__ASSERT_NO_MSG(false);
+	}
+}
+
+static void submit_work_if_buffer_free(struct icmsg_data_t *dev_data)
+{
+	if (!is_rx_buffer_free(dev_data)) {
+		return;
+	}
+
+	submit_mbox_work(dev_data);
+}
+
+static void submit_work_if_buffer_free_and_data_available(
+		struct icmsg_data_t *dev_data)
+{
+	if (!is_rx_buffer_free(dev_data)) {
+		return;
+	}
+	if (!is_rx_data_available(dev_data)) {
+		return;
+	}
+
+	submit_mbox_work(dev_data);
+}
+
 static void mbox_callback_process(struct k_work *item)
 {
 	struct icmsg_data_t *dev_data = CONTAINER_OF(item, struct icmsg_data_t, mbox_work);
-	uint8_t cb_buffer[CB_BUF_SIZE] __aligned(4);
 
 	atomic_t state = atomic_get(&dev_data->state);
-	int len = spsc_pbuf_read(dev_data->rx_ib, cb_buffer, CB_BUF_SIZE);
+	int len = spsc_pbuf_read(dev_data->rx_ib, dev_data->rx_buffer,
+				 RX_BUF_SIZE);
 
-	__ASSERT_NO_MSG(len <= CB_BUF_SIZE);
+	__ASSERT_NO_MSG(len <= RX_BUF_SIZE);
 
 	if (len == -EAGAIN) {
 		__ASSERT_NO_MSG(false);
-		(void)k_work_submit(&dev_data->mbox_work);
+		submit_mbox_work(dev_data);
 		return;
 	} else if (len <= 0) {
 		return;
@@ -79,13 +134,15 @@ static void mbox_callback_process(struct k_work *item)
 
 	if (state == ICMSG_STATE_READY) {
 		if (dev_data->cb->received) {
-			dev_data->cb->received(cb_buffer, len, dev_data->ctx);
+			dev_data->cb->received(dev_data->rx_buffer, len,
+					       dev_data->ctx);
 		}
 	} else {
 		int ret;
 
 		__ASSERT_NO_MSG(state == ICMSG_STATE_BUSY);
-		if (len != sizeof(magic) || memcmp(magic, cb_buffer, len)) {
+		if (len != sizeof(magic) ||
+		    memcmp(magic, dev_data->rx_buffer, len)) {
 			__ASSERT_NO_MSG(false);
 			return;
 		}
@@ -100,18 +157,7 @@ static void mbox_callback_process(struct k_work *item)
 		(void)ret;
 	}
 
-	/* Reading with NULL buffer to know if there are data in the
+	submit_work_if_buffer_free_and_data_available(dev_data);
-	 * buffer to be read.
-	 */
-	len = spsc_pbuf_read(dev_data->rx_ib, NULL, 0);
-	if (len > 0) {
-		if (k_work_submit(&dev_data->mbox_work) < 0) {
-			/* The mbox processing work is never canceled.
-			 * The negative error code should never be seen.
-			 */
-			__ASSERT_NO_MSG(false);
-		}
-	}
 }
 
 static void mbox_callback(const struct device *instance, uint32_t channel,
@@ -119,7 +165,7 @@ static void mbox_callback(const struct device *instance, uint32_t channel,
 {
 	struct icmsg_data_t *dev_data = user_data;
 
-	(void)k_work_submit(&dev_data->mbox_work);
+	submit_work_if_buffer_free(dev_data);
 }
 
 static int mbox_init(const struct icmsg_config_t *conf,
@@ -212,7 +258,7 @@ int icmsg_send(const struct icmsg_config_t *conf,
 	int ret;
 	int sent_bytes;
 
-	if (atomic_get(&dev_data->state) != ICMSG_STATE_READY) {
+	if (!is_endpoint_ready(dev_data)) {
 		return -EBUSY;
 	}
 
@@ -239,6 +285,56 @@ int icmsg_send(const struct icmsg_config_t *conf,
 	return sent_bytes;
 }
 
+#ifdef CONFIG_IPC_SERVICE_ICMSG_NOCOPY_RX
+int icmsg_hold_rx_buffer(const struct icmsg_config_t *conf,
+			 struct icmsg_data_t *dev_data,
+			 const void *data)
+{
+	bool was_released;
+
+	if (!is_endpoint_ready(dev_data)) {
+		return -EBUSY;
+	}
+
+	if (data != dev_data->rx_buffer) {
+		return -EINVAL;
+	}
+
+	was_released = atomic_cas(&dev_data->rx_buffer_held,
+				  BUFFER_RELEASED, BUFFER_HELD);
+	if (!was_released) {
+		return -EALREADY;
+	}
+
+	return 0;
+}
+
+int icmsg_release_rx_buffer(const struct icmsg_config_t *conf,
+			    struct icmsg_data_t *dev_data,
+			    const void *data)
+{
+	bool was_held;
+
+	if (!is_endpoint_ready(dev_data)) {
+		return -EBUSY;
+	}
+
+	if (data != dev_data->rx_buffer) {
+		return -EINVAL;
+	}
+
+	was_held = atomic_cas(&dev_data->rx_buffer_held,
+			      BUFFER_HELD, BUFFER_RELEASED);
+	if (!was_held) {
+		return -EALREADY;
+	}
+
+	submit_work_if_buffer_free_and_data_available(dev_data);
+
+	return 0;
+}
+#endif /* CONFIG_IPC_SERVICE_ICMSG_NOCOPY_RX */
+
 int icmsg_clear_tx_memory(const struct icmsg_config_t *conf)
 {
 	/* Clear spsc_pbuf header and a part of the magic number. */