openamp: change rx buffer hold flag to count

This commit fix one issue, before this commit, if rpmsg_hold_rx_buffer()
is called in the endpoint callback to hold current received buffer and
call rpmsg_release_rx_buffer() to release this buffer immediately.
This buffer will be returned to the virtqueue in rpmsg_virtio_release_rx_buffer()
and returned to virtqueue again in rpmsg_virtio_return_buffer() after the ept->cb()
So same buffer returned to the virtqueue twice, error happened.
Follow shows this process:

rpmsg_virtio_rx_callback()
  - get rp_hdr
  - ept->cb(data = RPMSG_LOCATE_DATA(rp_hdr))
    - rpsmg_hold_rx_buffer(data)
    - rpmsg_release_rx_buffer(data) here has returned buffer to virtqueue
  - rpmsg_virtio_return_buffer() return same buffer to virtqueue again

So this commit change the HELD flag in rp_hdr to count to fix this issue and also
support hold/release rx buffer recursively.

Signed-off-by: Guiding Li <[email protected]>
Signed-off-by: Yin Tao <[email protected]>

lib/rpmsg/rpmsg_internal.h

@@ -28,7 +28,9 @@ extern "C" {
 #define RPMSG_ASSERT(_exp, _msg) metal_assert(_exp)
 #endif
 
-#define RPMSG_BUF_HELD (1U << 31) /* Flag to suggest to hold the buffer */
+/* Flag to suggest to hold the buffer */
+#define RPMSG_BUF_HELD_SHIFT (16)
+#define RPMSG_BUF_HELD_MASK ((uint32_t)0xFFFF << RPMSG_BUF_HELD_SHIFT)
 
 #define RPMSG_LOCATE_HDR(p) \
 	((struct rpmsg_hdr *)((unsigned char *)(p) - sizeof(struct rpmsg_hdr)))

lib/rpmsg/rpmsg_virtio.c

@@ -313,37 +313,54 @@ static int _rpmsg_virtio_get_buffer_size(struct rpmsg_virtio_device *rvdev)
 	return length;
 }
 
-static void rpmsg_virtio_hold_rx_buffer(struct rpmsg_device *rdev, void *rxbuf)
+static void rpmsg_virtio_hold_rx_buffer_nolock(struct rpmsg_device *rdev,
+					       struct rpmsg_hdr *rp_hdr)
 {
-	struct rpmsg_hdr *rp_hdr;
-
 	(void)rdev;
 
-	rp_hdr = RPMSG_LOCATE_HDR(rxbuf);
-
 	/* Set held status to keep buffer */
-	rp_hdr->reserved |= RPMSG_BUF_HELD;
+	rp_hdr->reserved += 1 << RPMSG_BUF_HELD_SHIFT;
 }
 
-static void rpmsg_virtio_release_rx_buffer(struct rpmsg_device *rdev,
-					   void *rxbuf)
+static void rpmsg_virtio_hold_rx_buffer(struct rpmsg_device *rdev, void *rxbuf)
+{
+	metal_mutex_acquire(&rdev->lock);
+	rpmsg_virtio_hold_rx_buffer_nolock(rdev, RPMSG_LOCATE_HDR(rxbuf));
+	metal_mutex_release(&rdev->lock);
+}
+
+static bool rpmsg_virtio_release_rx_buffer_nolock(struct rpmsg_device *rdev,
+	    					  struct rpmsg_hdr *rp_hdr)
 {
 	struct rpmsg_virtio_device *rvdev;
-	struct rpmsg_hdr *rp_hdr;
 	uint16_t idx;
 	uint32_t len;
 
 	rvdev = metal_container_of(rdev, struct rpmsg_virtio_device, rdev);
-	rp_hdr = RPMSG_LOCATE_HDR(rxbuf);
-	/* The reserved field contains buffer index */
-	idx = (uint16_t)(rp_hdr->reserved & ~RPMSG_BUF_HELD);
 
-	metal_mutex_acquire(&rdev->lock);
+	rp_hdr->reserved -= 1 << RPMSG_BUF_HELD_SHIFT;
+	if ((rp_hdr->reserved & RPMSG_BUF_HELD_MASK) > 0) {
+		return false;
+	}
+	/* The reserved field contains buffer index */
+	idx = (uint16_t)(rp_hdr->reserved & ~RPMSG_BUF_HELD_MASK);
 	/* Return buffer on virtqueue. */
 	len = virtqueue_get_buffer_length(rvdev->rvq, idx);
 	rpmsg_virtio_return_buffer(rvdev, rp_hdr, len, idx);
-	/* Tell peer we return some rx buffers */
-	virtqueue_kick(rvdev->rvq);
+	return true;
+}
+
+static void rpmsg_virtio_release_rx_buffer(struct rpmsg_device *rdev,
+					   void *rxbuf)
+{
+	struct rpmsg_virtio_device *rvdev;
+	rvdev = metal_container_of(rdev, struct rpmsg_virtio_device, rdev);
+
+	metal_mutex_acquire(&rdev->lock);
+	if (rpmsg_virtio_release_rx_buffer_nolock(rdev, RPMSG_LOCATE_HDR(rxbuf))) {
+		/* Tell peer we return some rx buffers */
+		virtqueue_kick(rvdev->rvq);
+	}
 	metal_mutex_release(&rdev->lock);
 }
 
@@ -558,6 +575,7 @@ static void rpmsg_virtio_rx_callback(struct virtqueue *vq)
 		/* 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_virtio_hold_rx_buffer_nolock(rdev, rp_hdr);
 		metal_mutex_release(&rdev->lock);
 
 		if (ept) {
@@ -576,14 +594,8 @@ static void rpmsg_virtio_rx_callback(struct virtqueue *vq)
 		}
 
 		metal_mutex_acquire(&rdev->lock);
-
-		/* Check whether callback wants to hold buffer */
-		if (!(rp_hdr->reserved & RPMSG_BUF_HELD)) {
-			/* No, return used buffers. */
-			rpmsg_virtio_return_buffer(rvdev, rp_hdr, len, idx);
-		}
-
-		rp_hdr = rpmsg_virtio_get_rx_buffer(rvdev, &len, &idx);
+		rpmsg_virtio_release_rx_buffer_nolock(rdev, rp_hdr);
+		rp_hdr = rpmsg_virtio_get_rx_buffer(rvdev, &next_len, &next_idx);
 		if (!rp_hdr) {
 			/* tell peer we return some rx buffer */
 			virtqueue_kick(rvdev->rvq);