rpmsg virtio: Implement the Tx buffer reclaimer

With the no-copy feature a tx buffer can be get, filled and then
sent to the remote side.
In Some error cases the application can need to release it instead
of sending it to the remote side.
As the virtqueue is updated when the buffer it get, it is not
possible to manage this use case at virtqueue level.

This patchset implements the release based on a buffer recycler.
The principle is to store the released buffer in a 'reclaimer' list.
On next get pmsg_virtio_get_tx_buffer call the buffer is reused.

Signed-off-by: Arnaud Pouliquen <arnaud.pouliquen@foss.st.com>

lib/include/openamp/rpmsg_virtio.h

@@ -64,6 +64,8 @@ struct rpmsg_virtio_config {
  * @svq: pointer to send virtqueue
  * @shbuf_io: pointer to the shared buffer I/O region
  * @shpool: pointer to the shared buffers pool
+ * @reclaimer: Rpmsg buffer reclaimer that contains buffers released by
+ *             the rpmsg_virtio_release_tx_buffer function.
  */
 struct rpmsg_virtio_device {
 	struct rpmsg_device rdev;
@@ -73,6 +75,7 @@ struct rpmsg_virtio_device {
 	struct virtqueue *svq;
 	struct metal_io_region *shbuf_io;
 	struct rpmsg_virtio_shm_pool *shpool;
+	struct metal_list reclaimer;
 };
 
 #define RPMSG_REMOTE	VIRTIO_DEV_DEVICE

lib/rpmsg/rpmsg_virtio.c

@@ -25,6 +25,19 @@
 /* Time to wait - In multiple of 1 msecs. */
 #define RPMSG_TICKS_PER_INTERVAL                1000
 
+/**
+ * struct vbuff_reclaimer_t - vring buffer recycler
+ *
+ * This structure is used by the rpmsg virtio to store unused virtio buffer, as the
+ * virtqueue structure has been already updated and memory allocated.
+ *
+ * @node: node in reclaimer list.
+ * @idx:  virtio descriptor index containing the buffer information.
+ */
+struct vbuff_reclaimer_t {
+	struct metal_list node;
+	uint16_t idx;
+};
 
 /* Default configuration */
 #ifndef VIRTIO_DEVICE_ONLY
@@ -157,25 +170,42 @@ static void *rpmsg_virtio_get_tx_buffer(struct rpmsg_virtio_device *rvdev,
 					uint32_t *len, uint16_t *idx)
 {
 	unsigned int role = rpmsg_virtio_get_role(rvdev);
+	struct metal_list *node;
+	struct vbuff_reclaimer_t *r_desc;
 	void *data = NULL;
 
+	/* Try first to recycle a buffer that has been freed without been used */
+	node = metal_list_first(&rvdev->reclaimer);
+	if (node) {
+		r_desc = metal_container_of(node, struct vbuff_reclaimer_t, node);
+		metal_list_del(node);
+		data = r_desc;
+
 #ifndef VIRTIO_DEVICE_ONLY
-	if (role == RPMSG_HOST) {
+		if (role == RPMSG_HOST)
+			*len = rvdev->config.h2r_buf_size;
+#endif /*!VIRTIO_DEVICE_ONLY*/
+#ifndef VIRTIO_DRIVER_ONLY
+		if (role == RPMSG_REMOTE) {
+			*idx = r_desc->idx;
+			*len = virtqueue_get_buffer_length(rvdev->svq, *idx);
+		}
+#endif /*!VIRTIO_DRIVER_ONLY*/
+#ifndef VIRTIO_DEVICE_ONLY
+	} else if (role == RPMSG_HOST) {
 		data = virtqueue_get_buffer(rvdev->svq, len, idx);
 		if (!data && rvdev->svq->vq_free_cnt) {
 			data = rpmsg_virtio_shm_pool_get_buffer(rvdev->shpool,
 					rvdev->config.h2r_buf_size);
 			*len = rvdev->config.h2r_buf_size;
 			*idx = 0;
 		}
-	}
 #endif /*!VIRTIO_DEVICE_ONLY*/
-
 #ifndef VIRTIO_DRIVER_ONLY
-	if (role == RPMSG_REMOTE) {
+	} else if (role == RPMSG_REMOTE) {
 		data = virtqueue_get_available_buffer(rvdev->svq, idx, len);
-	}
 #endif /*!VIRTIO_DRIVER_ONLY*/
+	}
 
 	return data;
 }
@@ -414,6 +444,32 @@ static int rpmsg_virtio_send_offchannel_nocopy(struct rpmsg_device *rdev,
 	return len;
 }
 
+static int rpmsg_virtio_release_tx_buffer(struct rpmsg_device *rdev, void *txbuf)
+{
+	struct rpmsg_virtio_device *rvdev;
+	struct rpmsg_hdr *rp_hdr = RPMSG_LOCATE_HDR(txbuf);
+	void *vbuff = rp_hdr;  /* only used to avoid warning on the cast of a packed structure */
+	struct vbuff_reclaimer_t *r_desc = (struct vbuff_reclaimer_t *)vbuff;
+	uint16_t idx;
+
+	/*
+	 * Reuse the RPMsg buffer to temporary store the vbuff_reclaimer_t structure.
+	 * Stores the index locally before overwriting the RPMsg header.
+	 */
+	idx = rp_hdr->reserved;
+
+	rvdev = metal_container_of(rdev, struct rpmsg_virtio_device, rdev);
+
+	metal_mutex_acquire(&rdev->lock);
+
+	r_desc->idx = idx;
+	metal_list_add_tail(&rvdev->reclaimer, &r_desc->node);
+
+	metal_mutex_release(&rdev->lock);
+
+	return RPMSG_SUCCESS;
+}
+
 /**
  * This function sends rpmsg "message" to remote device.
  *
@@ -655,6 +711,7 @@ int rpmsg_init_vdev_with_config(struct rpmsg_virtio_device *rvdev,
 	rdev->ops.release_rx_buffer = rpmsg_virtio_release_rx_buffer;
 	rdev->ops.get_tx_payload_buffer = rpmsg_virtio_get_tx_payload_buffer;
 	rdev->ops.send_offchannel_nocopy = rpmsg_virtio_send_offchannel_nocopy;
+	rdev->ops.release_tx_buffer = rpmsg_virtio_release_tx_buffer;
 	role = rpmsg_virtio_get_role(rvdev);
 
 #ifndef VIRTIO_DEVICE_ONLY
@@ -716,6 +773,7 @@ int rpmsg_init_vdev_with_config(struct rpmsg_virtio_device *rvdev,
 	}
 #endif /*!VIRTIO_DRIVER_ONLY*/
 	rvdev->shbuf_io = shm_io;
+	metal_list_init(&rvdev->reclaimer);
 
 	/* Create virtqueues for remote device */
 	status = rpmsg_virtio_create_virtqueues(rvdev, 0, RPMSG_NUM_VRINGS,