rpmsg: Allow to send virqueue_kick only when RX queue is empty

Add VQ_RX_EMPTY_NOTIFY config to define the behavior. If
VQ_RX_EMPTY_NOTIFY is disabled, notify the other side each
time a buffer is released. If VQ_RX_EMPTY_NOTIFY is enabled,
only send one notification when the RX queue is empty to
improve performance.

Signed-off-by: Yongrong Wang <[email protected]>

README.md

@@ -86,6 +86,9 @@ library for it project:
   This option can be set to OFF if the only the remote mode is implemented.
 * **WITH_VIRTIO_DEVICE** (default ON): Build with virtio device enabled.
   This option can be set to OFF if the only the driver mode is implemented.
+* **WITH_VQ_RX_EMPTY_NOTIFY** (default OFF): Choose notify mode. When set to
+  ON, only notify when there are no more Message in the RX queue. When set to
+  OFF, notify for each RX buffer released.
 * **WITH_STATIC_LIB** (default ON): Build with a static library.
 * **WITH_SHARED_LIB** (default ON): Build with a shared library.
 * **WITH_ZEPHYR** (default OFF): Build open-amp as a zephyr library. This option

cmake/options.cmake

@@ -80,6 +80,14 @@ if (WITH_VIRTIO_MMIO_DRV)
   add_definitions(-DWITH_VIRTIO_MMIO_DRV)
 endif (WITH_VIRTIO_MMIO_DRV)
 
+option (WITH_VQ_RX_EMPTY_NOTIFY "Build with virtqueue rx empty notify enabled" OFF)
+
+if (NOT WITH_VQ_RX_EMPTY_NOTIFY)
+  add_definitions(-DVQ_RX_EMPTY_NOTIFY=0)
+else (NOT WITH_VQ_RX_EMPTY_NOTIFY)
+  add_definitions(-DVQ_RX_EMPTY_NOTIFY=1)
+endif (NOT WITH_VQ_RX_EMPTY_NOTIFY)
+
 option (WITH_DCACHE "Build with all cache operations enabled" OFF)
 
 if (WITH_DCACHE)

lib/rpmsg/rpmsg_virtio.c

@@ -321,8 +321,6 @@ static bool rpmsg_virtio_release_rx_buffer_nolock(struct rpmsg_virtio_device *rv
 	/* Return buffer on virtqueue. */
 	len = virtqueue_get_buffer_length(rvdev->rvq, idx);
 	rpmsg_virtio_return_buffer(rvdev, rp_hdr, len, idx);
-	/* Tell peer we returned an rx buffer */
-	virtqueue_kick(rvdev->rvq);
 
 	return true;
 }
@@ -337,8 +335,11 @@ static void rpmsg_virtio_release_rx_buffer(struct rpmsg_device *rdev,
 	rp_hdr = RPMSG_LOCATE_HDR(rxbuf);
 
 	metal_mutex_acquire(&rdev->lock);
-	if (rpmsg_virtio_buf_held_dec_test(rp_hdr))
+	if (rpmsg_virtio_buf_held_dec_test(rp_hdr)) {
 		rpmsg_virtio_release_rx_buffer_nolock(rvdev, rp_hdr);
+		/* Tell peer we returned an rx buffer */
+		virtqueue_kick(rvdev->rvq);
+	}
 	metal_mutex_release(&rdev->lock);
 }
 
@@ -560,6 +561,7 @@ static void rpmsg_virtio_rx_callback(struct virtqueue *vq)
 	struct rpmsg_device *rdev = &rvdev->rdev;
 	struct rpmsg_endpoint *ept;
 	struct rpmsg_hdr *rp_hdr;
+	bool release = false;
 	uint32_t len;
 	uint16_t idx;
 	int status;
@@ -568,16 +570,19 @@ static void rpmsg_virtio_rx_callback(struct virtqueue *vq)
 		/* Process the received data from remote node */
 		metal_mutex_acquire(&rdev->lock);
 		rp_hdr = rpmsg_virtio_get_rx_buffer(rvdev, &len, &idx);
-		metal_mutex_release(&rdev->lock);
 
 		/* No more filled rx buffers */
-		if (!rp_hdr)
+		if (!rp_hdr) {
+			if (VIRTIO_ENABLED(VQ_RX_EMPTY_NOTIFY) && release)
+				/* Tell peer we returned some rx buffer */
+				virtqueue_kick(rvdev->rvq);
+			metal_mutex_release(&rdev->lock);
 			break;
+		}
 
 		rp_hdr->reserved = idx;
 
 		/* Get the channel node from the remote device channels list. */
-		metal_mutex_acquire(&rdev->lock);
 		ept = rpmsg_get_ept_from_addr(rdev, rp_hdr->dst);
 		rpmsg_ept_incref(ept);
 		RPMSG_BUF_HELD_INC(rp_hdr);
@@ -600,8 +605,15 @@ static void rpmsg_virtio_rx_callback(struct virtqueue *vq)
 
 		metal_mutex_acquire(&rdev->lock);
 		rpmsg_ept_decref(ept);
-		if (rpmsg_virtio_buf_held_dec_test(rp_hdr))
+		if (rpmsg_virtio_buf_held_dec_test(rp_hdr)) {
 			rpmsg_virtio_release_rx_buffer_nolock(rvdev, rp_hdr);
+			if (VIRTIO_ENABLED(VQ_RX_EMPTY_NOTIFY))
+				/* Kick will be sent only when last buffer is released */
+				release = true;
+			else
+				/* Tell peer we returned an rx buffer */
+				virtqueue_kick(rvdev->rvq);
+		}
 		metal_mutex_release(&rdev->lock);
 	}
 }