[dpdk-dev] [RFC PATCH 11/13] replace vhost dequeue packed ring function

[Marvin Liu]

Vhost dequeue function will try to handle packed descriptors four by
four. If packed descriptors can't be handled by fast path, will try
normal path. Loop will skip skip when normal path can't receive packet.
Zero copy function is following same logic.

Signed-off-by: Marvin Liu <yong.liu at intel.com>

diff --git a/lib/librte_vhost/virtio_net.c b/lib/librte_vhost/virtio_net.c
index 7094944cf..0f9292eb0 100644
--- a/lib/librte_vhost/virtio_net.c
+++ b/lib/librte_vhost/virtio_net.c
@@ -225,6 +225,50 @@ flush_enqueue_used_packed(struct virtio_net *dev,
 	vhost_log_cache_sync(dev, vq);
 }
 
+static __rte_always_inline void
+flush_dequeue_shadow_used_packed(struct virtio_net *dev,
+			struct vhost_virtqueue *vq)
+{
+	int i;
+	uint16_t used_idx;
+	uint16_t head_idx = vq->dequeue_shadow_head;
+	uint16_t head_flags;
+	uint16_t flags;
+
+	if (vq->shadow_used_packed[0].used_wrap_counter)
+		head_flags = VIRTIO_TX_FLAG_PACKED;
+	else
+		head_flags = VIRTIO_TX_WRAP_FLAG_PACKED;
+
+	if (vq->shadow_used_packed[0].len)
+		head_flags |= VRING_DESC_F_WRITE;
+
+	for (i = 1; i < vq->shadow_used_idx; i++) {
+		used_idx = vq->shadow_used_packed[i].used_idx;
+		if (vq->shadow_used_packed[i].used_wrap_counter)
+			flags = VIRTIO_TX_FLAG_PACKED;
+		else
+			flags = VIRTIO_TX_WRAP_FLAG_PACKED;
+
+		vq->desc_packed[used_idx].flags = flags;
+		vhost_log_cache_used_vring(dev, vq,
+					used_idx *
+					sizeof(struct vring_packed_desc),
+					sizeof(struct vring_packed_desc));
+	}
+
+	rte_smp_wmb();
+	vq->desc_packed[head_idx].flags = head_flags;
+
+	vhost_log_cache_used_vring(dev, vq,
+				head_idx *
+				sizeof(struct vring_packed_desc),
+				sizeof(struct vring_packed_desc));
+
+	vq->shadow_used_idx = 0;
+	vhost_log_cache_sync(dev, vq);
+}
+
 /* flags are same when flushing used ring in fast path */
 static __rte_always_inline void
 flush_used_fast_packed(struct virtio_net *dev, struct vhost_virtqueue *vq,
@@ -393,6 +437,24 @@ flush_enqueue_shadow_used_packed(struct virtio_net *dev,
 	}
 }
 
+static __rte_always_inline void
+flush_dequeue_shadow_used(struct virtio_net *dev, struct vhost_virtqueue *vq)
+{
+	if (!vq->shadow_used_idx)
+		return;
+
+	int16_t shadow_count = vq->last_used_idx - vq->dequeue_shadow_head;
+	if (shadow_count <= 0)
+		shadow_count += vq->size;
+
+	/* buffer shadow used ring as many as possible when doing dequeue */
+	shadow_count += vq->last_used_idx & 0x3;
+	if ((uint16_t)shadow_count >= (vq->size >> 1)) {
+		do_data_copy_dequeue(vq);
+		flush_dequeue_shadow_used_packed(dev, vq);
+		vhost_vring_call_packed(dev, vq);
+	}
+}
 
 /* avoid write operation when necessary, to lessen cache issues */
 #define ASSIGN_UNLESS_EQUAL(var, val) do {	\
@@ -2027,120 +2089,49 @@ virtio_dev_tx_normal_packed(struct virtio_net *dev, struct vhost_virtqueue *vq,
 
 	return 0;
 }
-
 static __rte_noinline uint16_t
 virtio_dev_tx_packed(struct virtio_net *dev, struct vhost_virtqueue *vq,
-	struct rte_mempool *mbuf_pool, struct rte_mbuf **pkts, uint16_t count)
+		struct rte_mempool *mbuf_pool, struct rte_mbuf **pkts,
+		uint32_t count)
 {
-	uint16_t i;
-
-	if (unlikely(dev->dequeue_zero_copy)) {
-		struct zcopy_mbuf *zmbuf, *next;
-
-		for (zmbuf = TAILQ_FIRST(&vq->zmbuf_list);
-		     zmbuf != NULL; zmbuf = next) {
-			next = TAILQ_NEXT(zmbuf, next);
+	uint32_t pkt_idx = 0;
+	uint32_t pkt_num;
+	uint32_t remained = count;
+	int ret;
 
-			if (mbuf_is_consumed(zmbuf->mbuf)) {
-				update_shadow_used_ring_packed(vq,
-						zmbuf->desc_idx,
-						0,
-						zmbuf->desc_count);
+	for (pkt_idx = 0; remained; pkt_idx += pkt_num, remained -= pkt_num) {
+		if (remained >= PACKED_DESC_PER_CACHELINE) {
+			ret = virtio_dev_tx_fast_packed(dev, vq, mbuf_pool,
+							&pkts[pkt_idx]);
 
-				TAILQ_REMOVE(&vq->zmbuf_list, zmbuf, next);
-				restore_mbuf(zmbuf->mbuf);
-				rte_pktmbuf_free(zmbuf->mbuf);
-				put_zmbuf(zmbuf);
-				vq->nr_zmbuf -= 1;
+			if (!ret) {
+				pkt_num = PACKED_DESC_PER_CACHELINE;
+				flush_dequeue_shadow_used(dev, vq);
+				continue;
 			}
 		}
 
-		if (likely(vq->shadow_used_idx)) {
-			flush_shadow_used_ring_packed(dev, vq);
-			vhost_vring_call_packed(dev, vq);
-		}
-	}
-
-	VHOST_LOG_DEBUG(VHOST_DATA, "(%d) %s\n", dev->vid, __func__);
-
-	count = RTE_MIN(count, MAX_PKT_BURST);
-	VHOST_LOG_DEBUG(VHOST_DATA, "(%d) about to dequeue %u buffers\n",
-			dev->vid, count);
-
-	for (i = 0; i < count; i++) {
-		struct buf_vector buf_vec[BUF_VECTOR_MAX];
-		uint16_t buf_id;
-		uint32_t dummy_len;
-		uint16_t desc_count, nr_vec = 0;
-		int err;
-
-		if (unlikely(fill_vec_buf_packed(dev, vq,
-						vq->last_avail_idx, &desc_count,
-						buf_vec, &nr_vec,
-						&buf_id, &dummy_len,
-						VHOST_ACCESS_RO) < 0))
-			break;
-
-		if (likely(dev->dequeue_zero_copy == 0))
-			update_shadow_used_ring_packed(vq, buf_id, 0,
-					desc_count);
-
-		pkts[i] = rte_pktmbuf_alloc(mbuf_pool);
-		if (unlikely(pkts[i] == NULL)) {
-			RTE_LOG(ERR, VHOST_DATA,
-				"Failed to allocate memory for mbuf.\n");
-			break;
-		}
-
-		err = copy_desc_to_mbuf(dev, vq, buf_vec, nr_vec, pkts[i],
-				mbuf_pool);
-		if (unlikely(err)) {
-			rte_pktmbuf_free(pkts[i]);
+		/* if remained desc not cache aligned, skip to next round */
+		if (((vq->last_avail_idx & 0x3) + remained) <
+			PACKED_DESC_PER_CACHELINE)
 			break;
-		}
 
-		if (unlikely(dev->dequeue_zero_copy)) {
-			struct zcopy_mbuf *zmbuf;
-
-			zmbuf = get_zmbuf(vq);
-			if (!zmbuf) {
-				rte_pktmbuf_free(pkts[i]);
+		if (virtio_dev_tx_normal_packed(dev, vq, mbuf_pool,
+						&pkts[pkt_idx]))
 				break;
-			}
-			zmbuf->mbuf = pkts[i];
-			zmbuf->desc_idx = buf_id;
-			zmbuf->desc_count = desc_count;
-
-			/*
-			 * Pin lock the mbuf; we will check later to see
-			 * whether the mbuf is freed (when we are the last
-			 * user) or not. If that's the case, we then could
-			 * update the used ring safely.
-			 */
-			rte_mbuf_refcnt_update(pkts[i], 1);
 
-			vq->nr_zmbuf += 1;
-			TAILQ_INSERT_TAIL(&vq->zmbuf_list, zmbuf, next);
-		}
-
-		vq->last_avail_idx += desc_count;
-		if (vq->last_avail_idx >= vq->size) {
-			vq->last_avail_idx -= vq->size;
-			vq->avail_wrap_counter ^= 1;
-		}
+		pkt_num = 1;
+		flush_dequeue_shadow_used(dev, vq);
 	}
 
-	if (likely(dev->dequeue_zero_copy == 0)) {
-		do_data_copy_dequeue(vq);
-		if (unlikely(i < count))
-			vq->shadow_used_idx = i;
-		if (likely(vq->shadow_used_idx)) {
-			flush_shadow_used_ring_packed(dev, vq);
-			vhost_vring_call_packed(dev, vq);
-		}
+	if (pkt_idx) {
+		if (vq->shadow_used_idx)
+			do_data_copy_dequeue(vq);
+
+		vhost_vring_call_packed(dev, vq);
 	}
 
-	return i;
+	return pkt_idx;
 }
 
 static __rte_always_inline void
-- 
2.17.1