[PATCH v6 00/11] implement packed virtqueues

[PATCH v6 00/11] implement packed virtqueues

[Jens Freimann]

This is a basic implementation of packed virtqueues as specified in the
Virtio 1.1 draft. A compiled version of the current draft is available
at https://github.com/oasis-tcs/virtio-docs.git (or as .pdf at
https://github.com/oasis-tcs/virtio-docs/blob/master/virtio-v1.1-packed-wd10.pdf

A packed virtqueue is different from a split virtqueue in that it
consists of only a single descriptor ring that replaces available and
used ring, index and descriptor buffer.

Each descriptor is readable and writable and has a flags field. These flags
will mark if a descriptor is available or used.  To detect new available descriptors
even after the ring has wrapped, device and driver each have a
single-bit wrap counter that is flipped from 0 to 1 and vice versa every time
the last descriptor in the ring is used/made available.

The idea behind this is to 1. improve performance by avoiding cache misses
and 2. be easier for devices to implement.

Regarding performance: with these patches I get 21.13 Mpps on my system
as compared to 18.8 Mpps with the virtio 1.0 code. Packet size was 64
bytes, 0.05% acceptable loss.  Test setup is described as in
http://dpdk.org/doc/guides/howto/pvp_reference_benchmark.html

Packet generator:
MoonGen
Intel(R) Xeon(R) CPU E5-2665 0 @ 2.40GHz
Intel X710 NIC
RHEL 7.4

Device under test:
Intel(R) Xeon(R) CPU E5-2667 v4 @ 3.20GHz
Intel X710 NIC
RHEL 7.4

VM on DuT: RHEL7.4

I plan to do more performance test with bigger frame sizes.

changes from v5->v6:
* fix VIRTQUEUE_DUMP macro
* rework mergeable rx buffer path, support out of order (not sure if I
  need a .next field to support chains) 
* move wmb in virtio_receive_pkts_packed() (Gavin)
* rename to virtio_init_split/_packed (Maxime)
* add support for ctrl virtqueues (Tiwei, thx Max for fixing)
* rework tx path to support update_packet_stats and
  virtqueue_xmit_offload, TODO: merge with split-ring code to
  avoid a lot of duplicate code
* remove unnecessary check for avoiding to call VIRTQUEUE_DUMP (Maxime)

changes from v4->v5:
* fix VIRTQUEUE_DUMP macro
* fix wrap counter logic in transmit and receive functions  

changes from v3->v4:
* added helpers to increment index and set available/used flags
* driver keeps track of number of descriptors used
* change logic in set_rxtx_funcs()
* add patch for ctrl virtqueue with support for packed virtqueues
* rename virtio-1.1.h to virtio-packed.h
* fix wrong sizeof() in "vhost: vring address setup for packed queues"
* fix coding style of function definition in "net/virtio: add packed
  virtqueue helpers"
* fix padding in vring_size()
* move patches to enable packed virtqueues end of series
* v4 has two open problems: I'm sending it out anyway for feedback/help:
 * when VIRTIO_NET_F_MRG_RXBUF enabled only 128 packets are send in
   guest, i.e. when ring is full for the first time. I suspect a bug in
   setting the avail/used flags

changes from v2->v3:
* implement event suppression
* add code do dump packed virtqueues
* don't use assert in vhost code
* rename virtio-user parameter to packed-vq
* support rxvf flush

changes from v1->v2:
* don't use VIRTQ_DESC_F_NEXT in used descriptors (Jason)
* no rte_panice() in guest triggerable code (Maxime)
* use unlikely when checking for vq (Maxime)
* rename everything from _1_1 to _packed  (Yuanhan)
* add two more patches to implement mergeable receive buffers

*** BLURB HERE ***

Jens Freimann (10):
  net/virtio: vring init for packed queues
  net/virtio: add packed virtqueue defines
  net/virtio: add packed virtqueue helpers
  net/virtio: flush packed receive virtqueues
  net/virtio: dump packed virtqueue data
  net/virtio: implement transmit path for packed queues
  net/virtio: implement receive path for packed queues
  net/virtio: add support for mergeable buffers with packed virtqueues
  net/virtio: add virtio send command packed queue support
  net/virtio: enable packed virtqueues by default

Yuanhan Liu (1):
  net/virtio-user: add option to use packed queues

 drivers/net/virtio/virtio_ethdev.c            | 135 ++++-
 drivers/net/virtio/virtio_ethdev.h            |   5 +
 drivers/net/virtio/virtio_pci.h               |   8 +
 drivers/net/virtio/virtio_ring.h              |  96 +++-
 drivers/net/virtio/virtio_rxtx.c              | 490 +++++++++++++++++-
 .../net/virtio/virtio_user/virtio_user_dev.c  |  10 +-
 .../net/virtio/virtio_user/virtio_user_dev.h  |   2 +-
 drivers/net/virtio/virtio_user_ethdev.c       |  14 +-
 drivers/net/virtio/virtqueue.c                |  21 +
 drivers/net/virtio/virtqueue.h                |  50 +-
 10 files changed, 796 insertions(+), 35 deletions(-)

-- 
2.17.1

[PATCH v6 01/11] net/virtio: vring init for packed queues

[Jens Freimann]

Add and initialize descriptor data structures.

Signed-off-by: Jens Freimann <jfreimann@redhat.com>
---
 drivers/net/virtio/virtio_ethdev.c | 24 +++++++------
 drivers/net/virtio/virtio_pci.h    |  8 +++++
 drivers/net/virtio/virtio_ring.h   | 55 +++++++++++++++++++++++++++---
 drivers/net/virtio/virtqueue.h     | 12 ++++++-
 4 files changed, 82 insertions(+), 17 deletions(-)

diff --git a/drivers/net/virtio/virtio_ethdev.c b/drivers/net/virtio/virtio_ethdev.c
index b81df0a99..29f3e1043 100644
--- a/drivers/net/virtio/virtio_ethdev.c
+++ b/drivers/net/virtio/virtio_ethdev.c
@@ -299,19 +299,21 @@ virtio_init_vring(struct virtqueue *vq)
 
 	PMD_INIT_FUNC_TRACE();
 
-	/*
-	 * Reinitialise since virtio port might have been stopped and restarted
-	 */
 	memset(ring_mem, 0, vq->vq_ring_size);
-	vring_init(vr, size, ring_mem, VIRTIO_PCI_VRING_ALIGN);
-	vq->vq_used_cons_idx = 0;
-	vq->vq_desc_head_idx = 0;
-	vq->vq_avail_idx = 0;
-	vq->vq_desc_tail_idx = (uint16_t)(vq->vq_nentries - 1);
+	vring_init(vq->hw, vr, size, ring_mem, VIRTIO_PCI_VRING_ALIGN);
+
 	vq->vq_free_cnt = vq->vq_nentries;
 	memset(vq->vq_descx, 0, sizeof(struct vq_desc_extra) * vq->vq_nentries);
+	vq->vq_used_cons_idx = 0;
+	vq->vq_avail_idx     = 0;
+	if (vtpci_packed_queue(vq->hw)) {
+		vring_desc_init_packed(vr, size);
+	} else {
+		vq->vq_desc_head_idx = 0;
+		vq->vq_desc_tail_idx = (uint16_t)(vq->vq_nentries - 1);
 
-	vring_desc_init(vr->desc, size);
+		vring_desc_init_split(vr->desc, size);
+	}
 
 	/*
 	 * Disable device(host) interrupting guest
@@ -386,7 +388,7 @@ virtio_init_queue(struct rte_eth_dev *dev, uint16_t vtpci_queue_idx)
 	/*
 	 * Reserve a memzone for vring elements
 	 */
-	size = vring_size(vq_size, VIRTIO_PCI_VRING_ALIGN);
+	size = vring_size(hw, vq_size, VIRTIO_PCI_VRING_ALIGN);
 	vq->vq_ring_size = RTE_ALIGN_CEIL(size, VIRTIO_PCI_VRING_ALIGN);
 	PMD_INIT_LOG(DEBUG, "vring_size: %d, rounded_vring_size: %d",
 		     size, vq->vq_ring_size);
@@ -489,7 +491,7 @@ virtio_init_queue(struct rte_eth_dev *dev, uint16_t vtpci_queue_idx)
 		for (i = 0; i < vq_size; i++) {
 			struct vring_desc *start_dp = txr[i].tx_indir;
 
-			vring_desc_init(start_dp, RTE_DIM(txr[i].tx_indir));
+			vring_desc_init_split(start_dp, RTE_DIM(txr[i].tx_indir));
 
 			/* first indirect descriptor is always the tx header */
 			start_dp->addr = txvq->virtio_net_hdr_mem
diff --git a/drivers/net/virtio/virtio_pci.h b/drivers/net/virtio/virtio_pci.h
index 58fdd3d45..90204d281 100644
--- a/drivers/net/virtio/virtio_pci.h
+++ b/drivers/net/virtio/virtio_pci.h
@@ -113,6 +113,8 @@ struct virtnet_ctl;
 
 #define VIRTIO_F_VERSION_1		32
 #define VIRTIO_F_IOMMU_PLATFORM	33
+#define VIRTIO_F_RING_PACKED		34
+#define VIRTIO_F_IN_ORDER		35
 
 /*
  * Some VirtIO feature bits (currently bits 28 through 31) are
@@ -314,6 +316,12 @@ vtpci_with_feature(struct virtio_hw *hw, uint64_t bit)
 	return (hw->guest_features & (1ULL << bit)) != 0;
 }
 
+static inline int
+vtpci_packed_queue(struct virtio_hw *hw)
+{
+	return vtpci_with_feature(hw, VIRTIO_F_RING_PACKED);
+}
+
 /*
  * Function declaration from virtio_pci.c
  */
diff --git a/drivers/net/virtio/virtio_ring.h b/drivers/net/virtio/virtio_ring.h
index 9e3c2a015..309069fdb 100644
--- a/drivers/net/virtio/virtio_ring.h
+++ b/drivers/net/virtio/virtio_ring.h
@@ -54,11 +54,38 @@ struct vring_used {
 	struct vring_used_elem ring[0];
 };
 
+/* For support of packed virtqueues in Virtio 1.1 the format of descriptors
+ * looks like this.
+ */
+struct vring_desc_packed {
+	uint64_t addr;
+	uint32_t len;
+	uint16_t index;
+	uint16_t flags;
+} __attribute__ ((aligned(16)));
+
+#define RING_EVENT_FLAGS_ENABLE 0x0
+#define RING_EVENT_FLAGS_DISABLE 0x1
+#define RING_EVENT_FLAGS_DESC 0x2
+struct vring_packed_desc_event {
+	uint16_t desc_event_off_wrap;
+	uint16_t desc_event_flags;
+};
+
 struct vring {
 	unsigned int num;
-	struct vring_desc  *desc;
-	struct vring_avail *avail;
-	struct vring_used  *used;
+	union {
+		struct vring_desc_packed *desc_packed;
+		struct vring_desc *desc;
+	};
+	union {
+		struct vring_avail *avail;
+		struct vring_packed_desc_event *driver_event;
+	};
+	union {
+		struct vring_used  *used;
+		struct vring_packed_desc_event *device_event;
+	};
 };
 
 /* The standard layout for the ring is a continuous chunk of memory which
@@ -95,10 +122,18 @@ struct vring {
 #define vring_avail_event(vr) (*(uint16_t *)&(vr)->used->ring[(vr)->num])
 
 static inline size_t
-vring_size(unsigned int num, unsigned long align)
+vring_size(struct virtio_hw *hw, unsigned int num, unsigned long align)
 {
 	size_t size;
 
+	if (vtpci_packed_queue(hw)) {
+		size = num * sizeof(struct vring_desc_packed);
+		size += sizeof(struct vring_packed_desc_event);
+		size = RTE_ALIGN_CEIL(size, align);
+		size += sizeof(struct vring_packed_desc_event);
+		return size;
+	}
+
 	size = num * sizeof(struct vring_desc);
 	size += sizeof(struct vring_avail) + (num * sizeof(uint16_t));
 	size = RTE_ALIGN_CEIL(size, align);
@@ -108,10 +143,20 @@ vring_size(unsigned int num, unsigned long align)
 }
 
 static inline void
-vring_init(struct vring *vr, unsigned int num, uint8_t *p,
+vring_init(struct virtio_hw *hw, struct vring *vr, unsigned int num, uint8_t *p,
 	unsigned long align)
 {
 	vr->num = num;
+	if (vtpci_packed_queue(hw)) {
+		vr->desc_packed = (struct vring_desc_packed *)p;
+		vr->driver_event = (struct vring_packed_desc_event *)(p +
+			num * sizeof(struct vring_desc_packed));
+		vr->device_event = (struct vring_packed_desc_event *)
+				RTE_ALIGN_CEIL((uintptr_t)(vr->driver_event +
+				sizeof(struct vring_packed_desc_event)), align);
+		return;
+	}
+
 	vr->desc = (struct vring_desc *) p;
 	vr->avail = (struct vring_avail *) (p +
 		num * sizeof(struct vring_desc));
diff --git a/drivers/net/virtio/virtqueue.h b/drivers/net/virtio/virtqueue.h
index 26518ed98..7f337cd42 100644
--- a/drivers/net/virtio/virtqueue.h
+++ b/drivers/net/virtio/virtqueue.h
@@ -245,9 +245,19 @@ struct virtio_tx_region {
 			   __attribute__((__aligned__(16)));
 };
 
+static inline void
+vring_desc_init_packed(struct vring *vr, int n)
+{
+	int i;
+	for (i = 0; i < n; i++) {
+		struct vring_desc_packed *desc = &vr->desc_packed[i];
+		desc->index = i;
+	}
+}
+
 /* Chain all the descriptors in the ring with an END */
 static inline void
-vring_desc_init(struct vring_desc *dp, uint16_t n)
+vring_desc_init_split(struct vring_desc *dp, uint16_t n)
 {
 	uint16_t i;
 
-- 
2.17.1

[PATCH v6 02/11] net/virtio: add packed virtqueue defines

[Jens Freimann]

Signed-off-by: Jens Freimann <jfreimann@redhat.com>
---
 drivers/net/virtio/virtio_ring.h | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/drivers/net/virtio/virtio_ring.h b/drivers/net/virtio/virtio_ring.h
index 309069fdb..36a65f9b3 100644
--- a/drivers/net/virtio/virtio_ring.h
+++ b/drivers/net/virtio/virtio_ring.h
@@ -15,7 +15,11 @@
 #define VRING_DESC_F_WRITE      2
 /* This means the buffer contains a list of buffer descriptors. */
 #define VRING_DESC_F_INDIRECT   4
+/* This flag means the descriptor was made available by the driver */
 
+#define VRING_DESC_F_AVAIL(b)   ((uint16_t)(b) << 7)
+/* This flag means the descriptor was used by the device */
+#define VRING_DESC_F_USED(b)    ((uint16_t)(b) << 15)
 /* The Host uses this in used->flags to advise the Guest: don't kick me
  * when you add a buffer.  It's unreliable, so it's simply an
  * optimization.  Guest will still kick if it's out of buffers. */
-- 
2.17.1

[PATCH v6 03/11] net/virtio: add packed virtqueue helpers

[Jens Freimann]

Add helper functions to set/clear and check descriptor flags.

Signed-off-by: Jens Freimann <jfreimann@redhat.com>
---
 drivers/net/virtio/virtio_ring.h | 26 ++++++++++++++++++++++++++
 drivers/net/virtio/virtqueue.h   | 11 +++++++++++
 2 files changed, 37 insertions(+)

diff --git a/drivers/net/virtio/virtio_ring.h b/drivers/net/virtio/virtio_ring.h
index 36a65f9b3..b9e63d4d4 100644
--- a/drivers/net/virtio/virtio_ring.h
+++ b/drivers/net/virtio/virtio_ring.h
@@ -78,6 +78,8 @@ struct vring_packed_desc_event {
 
 struct vring {
 	unsigned int num;
+	unsigned int avail_wrap_counter;
+	unsigned int used_wrap_counter;
 	union {
 		struct vring_desc_packed *desc_packed;
 		struct vring_desc *desc;
@@ -92,6 +94,30 @@ struct vring {
 	};
 };
 
+static inline void
+_set_desc_avail(struct vring_desc_packed *desc, int wrap_counter)
+{
+	desc->flags |= VRING_DESC_F_AVAIL(wrap_counter) |
+		       VRING_DESC_F_USED(!wrap_counter);
+}
+
+static inline void
+set_desc_avail(struct vring *vr, struct vring_desc_packed *desc)
+{
+	_set_desc_avail(desc, vr->avail_wrap_counter);
+}
+
+static inline int
+desc_is_used(struct vring_desc_packed *desc, struct vring *vr)
+{
+	uint16_t used, avail;
+
+	used = !!(desc->flags & VRING_DESC_F_USED(1));
+	avail = !!(desc->flags & VRING_DESC_F_AVAIL(1));
+
+	return used == avail && used == vr->used_wrap_counter;
+}
+
 /* The standard layout for the ring is a continuous chunk of memory which
  * looks like this.  We assume num is a power of 2.
  *
diff --git a/drivers/net/virtio/virtqueue.h b/drivers/net/virtio/virtqueue.h
index 7f337cd42..aba6be3f9 100644
--- a/drivers/net/virtio/virtqueue.h
+++ b/drivers/net/virtio/virtqueue.h
@@ -245,6 +245,17 @@ struct virtio_tx_region {
 			   __attribute__((__aligned__(16)));
 };
 
+static inline uint16_t
+update_pq_avail_index(struct virtqueue *vq)
+{
+	if (++vq->vq_avail_idx >= vq->vq_nentries) {
+		vq->vq_avail_idx = 0;
+		vq->vq_ring.avail_wrap_counter ^= 1;
+	}
+
+	return vq->vq_avail_idx;
+}
+
 static inline void
 vring_desc_init_packed(struct vring *vr, int n)
 {
-- 
2.17.1

[PATCH v6 04/11] net/virtio: flush packed receive virtqueues

[Jens Freimann]

Flush used descriptors in packed receive virtqueue. As descriptors
can be chained we need to look at the stored number of used descriptors
to find out the length of the chain.

Signed-off-by: Jens Freimann <jfreimann@redhat.com>
---
 drivers/net/virtio/virtqueue.c | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/drivers/net/virtio/virtqueue.c b/drivers/net/virtio/virtqueue.c
index 56a77cc71..46dd9de52 100644
--- a/drivers/net/virtio/virtqueue.c
+++ b/drivers/net/virtio/virtqueue.c
@@ -58,12 +58,33 @@ virtqueue_detach_unused(struct virtqueue *vq)
 void
 virtqueue_rxvq_flush(struct virtqueue *vq)
 {
+	struct vring_desc_packed *descs = vq->vq_ring.desc_packed;
 	struct virtnet_rx *rxq = &vq->rxq;
 	struct virtio_hw *hw = vq->hw;
 	struct vring_used_elem *uep;
 	struct vq_desc_extra *dxp;
 	uint16_t used_idx, desc_idx;
 	uint16_t nb_used, i;
+	uint16_t size = vq->vq_nentries;
+
+	if (vtpci_packed_queue(vq->hw)) {
+		i = vq->vq_used_cons_idx;
+		if (i >= vq->vq_used_cons_idx) {
+			PMD_INIT_LOG(ERR, "vq_used_cons_idx out of range");
+			return;
+		}
+		while (desc_is_used(&descs[i], &vq->vq_ring) &&
+			i < vq->vq_nentries) {
+			dxp = &vq->vq_descx[i];
+			if (dxp->cookie != NULL)
+				rte_pktmbuf_free(dxp->cookie);
+			vq->vq_free_cnt += dxp->ndescs;
+			i = i + dxp->ndescs;
+			i = i >= size ? i - size : i;
+			dxp->ndescs = 0;
+		}
+		return;
+	}
 
 	nb_used = VIRTQUEUE_NUSED(vq);
 
-- 
2.17.1

[PATCH v6 05/11] net/virtio: dump packed virtqueue data

[Jens Freimann]

Add support to dump packed virtqueue data to the
VIRTQUEUE_DUMP() macro.

Signed-off-by: Jens Freimann <jfreimann@redhat.com>
---
 drivers/net/virtio/virtqueue.h | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/drivers/net/virtio/virtqueue.h b/drivers/net/virtio/virtqueue.h
index aba6be3f9..eb220563f 100644
--- a/drivers/net/virtio/virtqueue.h
+++ b/drivers/net/virtio/virtqueue.h
@@ -376,6 +376,15 @@ virtqueue_notify(struct virtqueue *vq)
 	uint16_t used_idx, nused; \
 	used_idx = (vq)->vq_ring.used->idx; \
 	nused = (uint16_t)(used_idx - (vq)->vq_used_cons_idx); \
+	if (vtpci_packed_queue((vq)->hw)) { \
+	  PMD_INIT_LOG(DEBUG, \
+	  "VQ: - size=%d; free=%d; used_cons_idx=%d; avail_idx=%d", \
+          "VQ: - avail_wrap_counter=%d; used_wrap_counter=%d", \
+	  (vq)->vq_nentries, (vq)->vq_free_cnt, (vq)->vq_used_cons_idx, \
+	  (vq)->vq_avail_idx, (vq)->vq_ring.avail_wrap_counter, \
+	  (vq)->vq_ring.used_wrap_counter); \
+	  break; \
+	} \
 	PMD_INIT_LOG(DEBUG, \
 	  "VQ: - size=%d; free=%d; used=%d; desc_head_idx=%d;" \
 	  " avail.idx=%d; used_cons_idx=%d; used.idx=%d;" \
-- 
2.17.1

[PATCH v6 06/11] net/virtio: implement transmit path for packed queues

[Jens Freimann]

This implements the transmit path for devices with
support for packed virtqueues.

Add the feature bit and enable code to
add buffers to vring and mark descriptors as available.

Signed-off-by: Jens Freiman <jfreimann@redhat.com>
---
 drivers/net/virtio/virtio_ethdev.c |   8 +-
 drivers/net/virtio/virtio_ethdev.h |   2 +
 drivers/net/virtio/virtio_ring.h   |  15 +-
 drivers/net/virtio/virtio_rxtx.c   | 243 +++++++++++++++++++++++++++++
 drivers/net/virtio/virtqueue.h     |  17 +-
 5 files changed, 280 insertions(+), 5 deletions(-)

diff --git a/drivers/net/virtio/virtio_ethdev.c b/drivers/net/virtio/virtio_ethdev.c
index 29f3e1043..5c28af282 100644
--- a/drivers/net/virtio/virtio_ethdev.c
+++ b/drivers/net/virtio/virtio_ethdev.c
@@ -384,6 +384,8 @@ virtio_init_queue(struct rte_eth_dev *dev, uint16_t vtpci_queue_idx)
 	vq->hw = hw;
 	vq->vq_queue_index = vtpci_queue_idx;
 	vq->vq_nentries = vq_size;
+	if (vtpci_packed_queue(hw))
+		vq->vq_ring.avail_wrap_counter = 1;
 
 	/*
 	 * Reserve a memzone for vring elements
@@ -1338,7 +1340,11 @@ set_rxtx_funcs(struct rte_eth_dev *eth_dev)
 		eth_dev->rx_pkt_burst = &virtio_recv_pkts;
 	}
 
-	if (hw->use_inorder_tx) {
+	if (vtpci_packed_queue(hw)) {
+		PMD_INIT_LOG(INFO, "virtio: using virtio 1.1 Tx path on port %u",
+			eth_dev->data->port_id);
+		eth_dev->tx_pkt_burst = virtio_xmit_pkts_packed;
+	} else if (hw->use_inorder_tx) {
 		PMD_INIT_LOG(INFO, "virtio: using inorder Tx path on port %u",
 			eth_dev->data->port_id);
 		eth_dev->tx_pkt_burst = virtio_xmit_pkts_inorder;
diff --git a/drivers/net/virtio/virtio_ethdev.h b/drivers/net/virtio/virtio_ethdev.h
index b726ad108..04161b461 100644
--- a/drivers/net/virtio/virtio_ethdev.h
+++ b/drivers/net/virtio/virtio_ethdev.h
@@ -79,6 +79,8 @@ uint16_t virtio_recv_mergeable_pkts_inorder(void *rx_queue,
 
 uint16_t virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts,
 		uint16_t nb_pkts);
+uint16_t virtio_xmit_pkts_packed(void *tx_queue, struct rte_mbuf **tx_pkts,
+		uint16_t nb_pkts);
 
 uint16_t virtio_xmit_pkts_inorder(void *tx_queue, struct rte_mbuf **tx_pkts,
 		uint16_t nb_pkts);
diff --git a/drivers/net/virtio/virtio_ring.h b/drivers/net/virtio/virtio_ring.h
index b9e63d4d4..dbffd4dcd 100644
--- a/drivers/net/virtio/virtio_ring.h
+++ b/drivers/net/virtio/virtio_ring.h
@@ -108,14 +108,25 @@ set_desc_avail(struct vring *vr, struct vring_desc_packed *desc)
 }
 
 static inline int
-desc_is_used(struct vring_desc_packed *desc, struct vring *vr)
+_desc_is_used(struct vring_desc_packed *desc)
 {
 	uint16_t used, avail;
 
 	used = !!(desc->flags & VRING_DESC_F_USED(1));
 	avail = !!(desc->flags & VRING_DESC_F_AVAIL(1));
 
-	return used == avail && used == vr->used_wrap_counter;
+	return used == avail;
+
+}
+
+static inline int
+desc_is_used(struct vring_desc_packed *desc, struct vring *vr)
+{
+	uint16_t used;
+
+	used = !!(desc->flags & VRING_DESC_F_USED(1));
+
+	return _desc_is_used(desc) && used == vr->used_wrap_counter;
 }
 
 /* The standard layout for the ring is a continuous chunk of memory which
diff --git a/drivers/net/virtio/virtio_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
index eb891433e..ea6300563 100644
--- a/drivers/net/virtio/virtio_rxtx.c
+++ b/drivers/net/virtio/virtio_rxtx.c
@@ -38,6 +38,7 @@
 #define  VIRTIO_DUMP_PACKET(m, len) do { } while (0)
 #endif
 
+
 int
 virtio_dev_rx_queue_done(void *rxq, uint16_t offset)
 {
@@ -165,6 +166,31 @@ virtqueue_dequeue_rx_inorder(struct virtqueue *vq,
 #endif
 
 /* Cleanup from completed transmits. */
+static void
+virtio_xmit_cleanup_packed(struct virtqueue *vq)
+{
+	uint16_t idx;
+	uint16_t size = vq->vq_nentries;
+	struct vring_desc_packed *desc = vq->vq_ring.desc_packed;
+	struct vq_desc_extra *dxp;
+
+	idx = vq->vq_used_cons_idx;
+	while (_desc_is_used(&desc[idx]) &&
+	       vq->vq_free_cnt < size) {
+		dxp = &vq->vq_descx[idx];
+		vq->vq_free_cnt += dxp->ndescs;
+		idx += dxp->ndescs;
+		idx = idx >= size ? idx - size : idx;
+		if (idx == 0) {
+			vq->vq_ring.used_wrap_counter ^= 1;
+		}
+		if (dxp->cookie != NULL) {
+			rte_pktmbuf_free(dxp->cookie);
+			dxp->cookie = NULL;
+		}
+	}
+}
+
 static void
 virtio_xmit_cleanup(struct virtqueue *vq, uint16_t num)
 {
@@ -456,6 +482,129 @@ virtqueue_enqueue_xmit_inorder(struct virtnet_tx *txvq,
 	vq->vq_desc_head_idx = idx & (vq->vq_nentries - 1);
 }
 
+static inline void
+virtqueue_enqueue_xmit_packed(struct virtnet_tx *txvq, struct rte_mbuf *cookie,
+			uint16_t needed, int use_indirect, int can_push,
+			int in_order)
+{
+	struct virtio_tx_region *txr = txvq->virtio_net_hdr_mz->addr;
+	struct vq_desc_extra *dxp;
+	struct virtqueue *vq = txvq->vq;
+	struct vring_desc_packed *start_dp;
+	uint16_t seg_num = cookie->nb_segs;
+	uint16_t head_idx, idx, prev;
+	uint16_t head_id;
+	uint16_t head_size = vq->hw->vtnet_hdr_size;
+	struct virtio_net_hdr *hdr;
+	int wrap_counter = vq->vq_ring.avail_wrap_counter;
+
+	head_idx = vq->vq_desc_head_idx;
+	idx = head_idx;
+	dxp = &vq->vq_descx[idx];
+	dxp->cookie = (void *)cookie;
+	dxp->ndescs = needed;
+
+	start_dp = vq->vq_ring.desc_packed;
+	head_id = start_dp[head_idx].index;
+
+	if (can_push) {
+		/* prepend cannot fail, checked by caller */
+		hdr = (struct virtio_net_hdr *)
+			rte_pktmbuf_prepend(cookie, head_size);
+		/* rte_pktmbuf_prepend() counts the hdr size to the pkt length,
+		 * which is wrong. Below subtract restores correct pkt size.
+		 */
+		cookie->pkt_len -= head_size;
+
+		/* if offload disabled, it is not zeroed below, do it now */
+		if (!vq->hw->has_tx_offload) {
+			ASSIGN_UNLESS_EQUAL(hdr->csum_start, 0);
+			ASSIGN_UNLESS_EQUAL(hdr->csum_offset, 0);
+			ASSIGN_UNLESS_EQUAL(hdr->flags, 0);
+			ASSIGN_UNLESS_EQUAL(hdr->gso_type, 0);
+			ASSIGN_UNLESS_EQUAL(hdr->gso_size, 0);
+			ASSIGN_UNLESS_EQUAL(hdr->hdr_len, 0);
+		}
+	} else if (use_indirect) {
+		/* setup tx ring slot to point to indirect
+		 * descriptor list stored in reserved region.
+		 *
+		 * the first slot in indirect ring is already preset
+		 * to point to the header in reserved region
+		 */
+		start_dp[idx].addr  = txvq->virtio_net_hdr_mem +
+			RTE_PTR_DIFF(&txr[idx].tx_indir, txr);
+		start_dp[idx].len   = (seg_num + 1) * sizeof(struct vring_desc);
+		start_dp[idx].flags = VRING_DESC_F_INDIRECT;
+		hdr = (struct virtio_net_hdr *)&txr[idx].tx_hdr;
+
+		/* loop below will fill in rest of the indirect elements */
+		start_dp = txr[idx].tx_indir_pq;
+		idx = 1;
+	} else {
+		/* setup first tx ring slot to point to header
+		 * stored in reserved region.
+		 */
+		start_dp[idx].addr  = txvq->virtio_net_hdr_mem +
+			RTE_PTR_DIFF(&txr[idx].tx_hdr, txr);
+		start_dp[idx].len   = vq->hw->vtnet_hdr_size;
+		start_dp[idx].flags = VRING_DESC_F_NEXT |
+			VRING_DESC_F_AVAIL(vq->vq_ring.avail_wrap_counter) |
+			VRING_DESC_F_USED(!vq->vq_ring.used_wrap_counter); 
+		hdr = (struct virtio_net_hdr *)&txr[idx].tx_hdr;
+
+		idx++;
+	}
+
+	virtqueue_xmit_offload(hdr, cookie, vq->hw->has_tx_offload);
+
+	do {
+		if (idx >= vq->vq_nentries) {
+			idx -= vq->vq_nentries;
+			vq->vq_ring.avail_wrap_counter ^= 1;
+			vq->vq_ring.used_wrap_counter ^= 1;
+		}
+		start_dp[idx].addr  = VIRTIO_MBUF_DATA_DMA_ADDR(cookie, vq);
+		start_dp[idx].len   = cookie->data_len;
+		start_dp[idx].flags = VRING_DESC_F_NEXT |
+			VRING_DESC_F_AVAIL(vq->vq_ring.avail_wrap_counter) |
+			VRING_DESC_F_USED(!vq->vq_ring.used_wrap_counter); 
+		idx++;
+	} while ((cookie = cookie->next) != NULL);
+
+	if (use_indirect)
+		idx = vq->vq_ring.desc_packed[head_idx].index;
+
+	if (idx >= vq->vq_nentries) {
+		idx -= vq->vq_nentries;
+		vq->vq_ring.avail_wrap_counter ^= 1;
+		vq->vq_ring.used_wrap_counter ^= 1;
+	}
+
+	vq->vq_free_cnt = (uint16_t)(vq->vq_free_cnt - needed);
+
+	if (needed > 1) {
+		prev = (idx - 1 > 0 ? idx - 1 : vq->vq_nentries) - 1;
+		start_dp[prev].index = head_id;
+		start_dp[prev].flags =
+			(VRING_DESC_F_AVAIL(wrap_counter) |
+			 VRING_DESC_F_USED(!wrap_counter));
+	}
+	start_dp[head_idx].flags =
+		(VRING_DESC_F_AVAIL(wrap_counter) |
+		 VRING_DESC_F_USED(!wrap_counter));
+	rte_smp_wmb();
+
+	vq->vq_desc_head_idx = idx;
+	vq->vq_avail_idx = idx;
+
+	if (!in_order) {
+		if (vq->vq_desc_head_idx == VQ_RING_DESC_CHAIN_END)
+			vq->vq_desc_tail_idx = idx;
+	}
+}
+
+
 static inline void
 virtqueue_enqueue_xmit(struct virtnet_tx *txvq, struct rte_mbuf *cookie,
 			uint16_t needed, int use_indirect, int can_push,
@@ -736,6 +885,9 @@ virtio_dev_tx_queue_setup_finish(struct rte_eth_dev *dev,
 	if (hw->use_inorder_tx)
 		vq->vq_ring.desc[vq->vq_nentries - 1].next = 0;
 
+	if (vtpci_packed_queue(hw))
+		vq->vq_ring.avail_wrap_counter = 1;
+
 	VIRTQUEUE_DUMP(vq);
 
 	return 0;
@@ -1346,6 +1498,97 @@ virtio_recv_mergeable_pkts(void *rx_queue,
 	return nb_rx;
 }
 
+uint16_t
+virtio_xmit_pkts_packed(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
+{
+	struct virtnet_tx *txvq = tx_queue;
+	struct virtqueue *vq = txvq->vq;
+	struct virtio_hw *hw = vq->hw;
+	uint16_t hdr_size = hw->vtnet_hdr_size;
+	uint16_t nb_tx = 0;
+	int error;
+
+	if (unlikely(hw->started == 0 && tx_pkts != hw->inject_pkts))
+		return nb_tx;
+
+	if (unlikely(nb_pkts < 1))
+		return nb_pkts;
+
+	PMD_TX_LOG(DEBUG, "%d packets to xmit", nb_pkts);
+
+	virtio_rmb();
+	if (likely(nb_pkts > vq->vq_nentries - vq->vq_free_thresh))
+		virtio_xmit_cleanup_packed(vq);
+
+	for (nb_tx = 0; nb_tx < nb_pkts; nb_tx++) {
+		struct rte_mbuf *txm = tx_pkts[nb_tx];
+		int can_push = 0, use_indirect = 0, slots, need;
+
+		/* Do VLAN tag insertion */
+		if (unlikely(txm->ol_flags & PKT_TX_VLAN_PKT)) {
+			error = rte_vlan_insert(&txm);
+			if (unlikely(error)) {
+				rte_pktmbuf_free(txm);
+				continue;
+			}
+		}
+
+		/* optimize ring usage */
+		if ((vtpci_with_feature(hw, VIRTIO_F_ANY_LAYOUT) ||
+		      vtpci_with_feature(hw, VIRTIO_F_VERSION_1)) &&
+		    rte_mbuf_refcnt_read(txm) == 1 &&
+		    RTE_MBUF_DIRECT(txm) &&
+		    txm->nb_segs == 1 &&
+		    rte_pktmbuf_headroom(txm) >= hdr_size &&
+		    rte_is_aligned(rte_pktmbuf_mtod(txm, char *),
+				   __alignof__(struct virtio_net_hdr_mrg_rxbuf)))
+			can_push = 1;
+		else if (vtpci_with_feature(hw, VIRTIO_RING_F_INDIRECT_DESC) &&
+			 txm->nb_segs < VIRTIO_MAX_TX_INDIRECT)
+			use_indirect = 1;
+
+		/* How many main ring entries are needed to this Tx?
+		 * any_layout => number of segments
+		 * indirect   => 1
+		 * default    => number of segments + 1
+		 */
+		slots = use_indirect ? 1 : (txm->nb_segs + !can_push);
+		need = slots - vq->vq_free_cnt;
+
+		/* Positive value indicates it need free vring descriptors */
+		if (unlikely(need > 0)) {
+			virtio_rmb();
+			need = RTE_MIN(need, (int)nb_pkts);
+
+			virtio_xmit_cleanup_packed(vq);
+			need = slots - vq->vq_free_cnt;
+			if (unlikely(need > 0)) {
+				PMD_TX_LOG(ERR,
+					   "No free tx descriptors to transmit");
+				break;
+			}
+		}
+
+		/* Enqueue Packet buffers */
+		virtqueue_enqueue_xmit_packed(txvq, txm, slots, use_indirect,
+			can_push, 0);
+
+		txvq->stats.bytes += txm->pkt_len;
+		virtio_update_packet_stats(&txvq->stats, txm);
+	}
+
+	txvq->stats.packets += nb_tx;
+
+	if (likely(nb_tx)) {
+		if (unlikely(virtqueue_kick_prepare_packed(vq))) {
+			virtqueue_notify(vq);
+			PMD_TX_LOG(DEBUG, "Notified backend after xmit");
+		}
+	}
+
+	return nb_tx;
+}
+
 uint16_t
 virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 {
diff --git a/drivers/net/virtio/virtqueue.h b/drivers/net/virtio/virtqueue.h
index eb220563f..ffa2d8f92 100644
--- a/drivers/net/virtio/virtqueue.h
+++ b/drivers/net/virtio/virtqueue.h
@@ -241,8 +241,12 @@ struct virtio_net_hdr_mrg_rxbuf {
 #define VIRTIO_MAX_TX_INDIRECT 8
 struct virtio_tx_region {
 	struct virtio_net_hdr_mrg_rxbuf tx_hdr;
-	struct vring_desc tx_indir[VIRTIO_MAX_TX_INDIRECT]
-			   __attribute__((__aligned__(16)));
+	union {
+		struct vring_desc tx_indir[VIRTIO_MAX_TX_INDIRECT]
+			__attribute__((__aligned__(16)));
+		struct vring_desc_packed tx_indir_pq[VIRTIO_MAX_TX_INDIRECT]
+			__attribute__((__aligned__(16)));
+	};
 };
 
 static inline uint16_t
@@ -360,6 +364,15 @@ virtqueue_kick_prepare(struct virtqueue *vq)
 	return !(vq->vq_ring.used->flags & VRING_USED_F_NO_NOTIFY);
 }
 
+static inline int
+virtqueue_kick_prepare_packed(struct virtqueue *vq)
+{
+	uint16_t flags;
+
+	flags = vq->vq_ring.device_event->desc_event_flags & RING_EVENT_FLAGS_DESC;
+	return (flags != RING_EVENT_FLAGS_DISABLE);
+}
+
 static inline void
 virtqueue_notify(struct virtqueue *vq)
 {
-- 
2.17.1

[PATCH v6 07/11] net/virtio: implement receive path for packed queues

[Jens Freimann]

Implement the receive part.

Signed-off-by: Jens Freimann <jfreimann@redhat.com>
---
 drivers/net/virtio/virtio_ethdev.c |  15 +++-
 drivers/net/virtio/virtio_ethdev.h |   2 +
 drivers/net/virtio/virtio_rxtx.c   | 130 +++++++++++++++++++++++++++++
 3 files changed, 144 insertions(+), 3 deletions(-)

diff --git a/drivers/net/virtio/virtio_ethdev.c b/drivers/net/virtio/virtio_ethdev.c
index 5c28af282..f03cf04a9 100644
--- a/drivers/net/virtio/virtio_ethdev.c
+++ b/drivers/net/virtio/virtio_ethdev.c
@@ -384,8 +384,10 @@ virtio_init_queue(struct rte_eth_dev *dev, uint16_t vtpci_queue_idx)
 	vq->hw = hw;
 	vq->vq_queue_index = vtpci_queue_idx;
 	vq->vq_nentries = vq_size;
-	if (vtpci_packed_queue(hw))
+	if (vtpci_packed_queue(hw)) {
 		vq->vq_ring.avail_wrap_counter = 1;
+		vq->vq_ring.used_wrap_counter = 1;
+	}
 
 	/*
 	 * Reserve a memzone for vring elements
@@ -1320,7 +1322,13 @@ set_rxtx_funcs(struct rte_eth_dev *eth_dev)
 {
 	struct virtio_hw *hw = eth_dev->data->dev_private;
 
-	if (hw->use_simple_rx) {
+	/*
+	 * workarount for packed vqs which don't support
+	 * mrg_rxbuf at this point
+	 */
+	if (vtpci_packed_queue(hw)) {
+		eth_dev->rx_pkt_burst = &virtio_recv_pkts_packed;
+	} else if (hw->use_simple_rx) {
 		PMD_INIT_LOG(INFO, "virtio: using simple Rx path on port %u",
 			eth_dev->data->port_id);
 		eth_dev->rx_pkt_burst = virtio_recv_pkts_vec;
@@ -1484,7 +1492,8 @@ virtio_init_device(struct rte_eth_dev *eth_dev, uint64_t req_features)
 
 	/* Setting up rx_header size for the device */
 	if (vtpci_with_feature(hw, VIRTIO_NET_F_MRG_RXBUF) ||
-	    vtpci_with_feature(hw, VIRTIO_F_VERSION_1))
+	    vtpci_with_feature(hw, VIRTIO_F_VERSION_1) ||
+	    vtpci_with_feature(hw, VIRTIO_F_RING_PACKED))
 		hw->vtnet_hdr_size = sizeof(struct virtio_net_hdr_mrg_rxbuf);
 	else
 		hw->vtnet_hdr_size = sizeof(struct virtio_net_hdr);
diff --git a/drivers/net/virtio/virtio_ethdev.h b/drivers/net/virtio/virtio_ethdev.h
index 04161b461..25eaff224 100644
--- a/drivers/net/virtio/virtio_ethdev.h
+++ b/drivers/net/virtio/virtio_ethdev.h
@@ -70,6 +70,8 @@ int virtio_dev_tx_queue_setup_finish(struct rte_eth_dev *dev,
 
 uint16_t virtio_recv_pkts(void *rx_queue, struct rte_mbuf **rx_pkts,
 		uint16_t nb_pkts);
+uint16_t virtio_recv_pkts_packed(void *rx_queue, struct rte_mbuf **rx_pkts,
+		uint16_t nb_pkts);
 
 uint16_t virtio_recv_mergeable_pkts(void *rx_queue, struct rte_mbuf **rx_pkts,
 		uint16_t nb_pkts);
diff --git a/drivers/net/virtio/virtio_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
index ea6300563..90a0e306f 100644
--- a/drivers/net/virtio/virtio_rxtx.c
+++ b/drivers/net/virtio/virtio_rxtx.c
@@ -31,6 +31,7 @@
 #include "virtqueue.h"
 #include "virtio_rxtx.h"
 #include "virtio_rxtx_simple.h"
+#include "virtio_ring.h"
 
 #ifdef RTE_LIBRTE_VIRTIO_DEBUG_DUMP
 #define VIRTIO_DUMP_PACKET(m, len) rte_pktmbuf_dump(stdout, m, len)
@@ -753,6 +754,34 @@ virtio_dev_rx_queue_setup_finish(struct rte_eth_dev *dev, uint16_t queue_idx)
 
 	PMD_INIT_FUNC_TRACE();
 
+	if (vtpci_packed_queue(hw)) {
+		struct vring_desc_packed *desc;
+		struct vq_desc_extra *dxp;
+
+		for (desc_idx = 0; desc_idx < vq->vq_nentries;
+				desc_idx++) {
+			m = rte_mbuf_raw_alloc(rxvq->mpool);
+			if (unlikely(m == NULL))
+				return -ENOMEM;
+
+			dxp = &vq->vq_descx[desc_idx];
+			dxp->cookie = m;
+			dxp->ndescs = 1;
+
+			desc = &vq->vq_ring.desc_packed[desc_idx];
+			desc->addr = VIRTIO_MBUF_ADDR(m, vq) +
+				RTE_PKTMBUF_HEADROOM - hw->vtnet_hdr_size;
+			desc->len = m->buf_len - RTE_PKTMBUF_HEADROOM +
+				hw->vtnet_hdr_size;
+			desc->flags |= VRING_DESC_F_WRITE;
+			rte_smp_wmb();
+			set_desc_avail(&vq->vq_ring, desc);
+		}
+		vq->vq_ring.avail_wrap_counter ^= 1;
+		nbufs = desc_idx;
+		goto out;
+	}
+
 	/* Allocate blank mbufs for the each rx descriptor */
 	nbufs = 0;
 
@@ -816,6 +845,7 @@ virtio_dev_rx_queue_setup_finish(struct rte_eth_dev *dev, uint16_t queue_idx)
 		vq_update_avail_idx(vq);
 	}
 
+out:
 	PMD_INIT_LOG(DEBUG, "Allocated %d bufs", nbufs);
 
 	VIRTQUEUE_DUMP(vq);
@@ -1034,6 +1064,106 @@ virtio_rx_offload(struct rte_mbuf *m, struct virtio_net_hdr *hdr)
 	return 0;
 }
 
+uint16_t
+virtio_recv_pkts_packed(void *rx_queue, struct rte_mbuf **rx_pkts,
+		     uint16_t nb_pkts)
+{
+	struct virtnet_rx *rxvq = rx_queue;
+	struct virtqueue *vq = rxvq->vq;
+	struct virtio_hw *hw = vq->hw;
+	struct rte_mbuf *rxm, *nmb;
+	uint16_t nb_rx;
+	uint32_t len;
+	uint32_t i;
+	uint32_t hdr_size;
+	struct virtio_net_hdr *hdr;
+	struct vring_desc_packed *descs = vq->vq_ring.desc_packed;
+	struct vring_desc_packed *desc;
+	uint16_t used_idx, id;
+	struct vq_desc_extra *dxp;
+
+	nb_rx = 0;
+	if (unlikely(hw->started == 0))
+		return nb_rx;
+
+	hdr_size = hw->vtnet_hdr_size;
+
+	for (i = 0; i < nb_pkts; i++) {
+		rte_smp_rmb();
+		used_idx = vq->vq_used_cons_idx;
+		desc = &descs[used_idx];
+		id = desc->index;
+		if (!desc_is_used(desc, &vq->vq_ring))
+			break;
+
+		nmb = rte_mbuf_raw_alloc(rxvq->mpool);
+		if (unlikely(nmb == NULL)) {
+			struct rte_eth_dev *dev
+				= &rte_eth_devices[rxvq->port_id];
+			dev->data->rx_mbuf_alloc_failed++;
+			break;
+		}
+
+		dxp = &vq->vq_descx[id];
+		len = desc->len;
+		rxm = dxp->cookie;
+		dxp->cookie = nmb;
+		dxp->ndescs = 1;
+
+		desc->addr = VIRTIO_MBUF_ADDR(nmb, vq) +
+			RTE_PKTMBUF_HEADROOM - hw->vtnet_hdr_size;
+		desc->len = nmb->buf_len - RTE_PKTMBUF_HEADROOM +
+			hw->vtnet_hdr_size;
+		desc->flags = VRING_DESC_F_WRITE;
+		rte_smp_wmb();
+
+		PMD_RX_LOG(DEBUG, "packet len:%d", len);
+
+		if (unlikely(len < hdr_size + ETHER_HDR_LEN)) {
+			PMD_RX_LOG(ERR, "Packet drop");
+			rte_pktmbuf_free(rxm);
+			rxvq->stats.errors++;
+			continue;
+		}
+
+		rxm->port = rxvq->port_id;
+		rxm->data_off = RTE_PKTMBUF_HEADROOM;
+		rxm->ol_flags = 0;
+		rxm->vlan_tci = 0;
+
+		rxm->pkt_len = (uint32_t)(len - hdr_size);
+		rxm->data_len = (uint16_t)(len - hdr_size);
+
+		hdr = (struct virtio_net_hdr *)((char *)rxm->buf_addr +
+			RTE_PKTMBUF_HEADROOM - hdr_size);
+
+		if (hw->vlan_strip)
+			rte_vlan_strip(rxm);
+
+		if (hw->has_rx_offload && virtio_rx_offload(rxm, hdr) < 0) {
+			rte_pktmbuf_free(rxm);
+			rxvq->stats.errors++;
+			continue;
+		}
+
+		VIRTIO_DUMP_PACKET(rxm, rxm->data_len);
+
+		rxvq->stats.bytes += rxm->pkt_len;
+		virtio_update_packet_stats(&rxvq->stats, rxm);
+
+		rx_pkts[nb_rx++] = rxm;
+		vq->vq_used_cons_idx += dxp->ndescs;
+		if (vq->vq_used_cons_idx >= vq->vq_nentries) {
+			vq->vq_used_cons_idx -= vq->vq_nentries;
+			vq->vq_ring.used_wrap_counter ^= 1;
+		}
+	}
+
+	rxvq->stats.packets += nb_rx;
+
+	return nb_rx;
+}
+
 #define VIRTIO_MBUF_BURST_SZ 64
 #define DESC_PER_CACHELINE (RTE_CACHE_LINE_SIZE / sizeof(struct vring_desc))
 uint16_t
-- 
2.17.1

[PATCH v6 08/11] net/virtio: add support for mergeable buffers with packed virtqueues

[Jens Freimann]

Implement support for receiving merged buffers in virtio when packed
virtqueues are enabled.

Signed-off-by: Jens Freimann <jfreimann@redhat.com>
---
 drivers/net/virtio/virtio_ethdev.c |   8 +-
 drivers/net/virtio/virtio_rxtx.c   | 117 ++++++++++++++++++++++++++---
 drivers/net/virtio/virtqueue.h     |   1 +
 3 files changed, 113 insertions(+), 13 deletions(-)

diff --git a/drivers/net/virtio/virtio_ethdev.c b/drivers/net/virtio/virtio_ethdev.c
index f03cf04a9..c4ef095ed 100644
--- a/drivers/net/virtio/virtio_ethdev.c
+++ b/drivers/net/virtio/virtio_ethdev.c
@@ -1322,12 +1322,12 @@ set_rxtx_funcs(struct rte_eth_dev *eth_dev)
 {
 	struct virtio_hw *hw = eth_dev->data->dev_private;
 
-	/*
-	 * workarount for packed vqs which don't support
-	 * mrg_rxbuf at this point
-	 */
 	if (vtpci_packed_queue(hw)) {
+		if (vtpci_with_feature(hw, VIRTIO_NET_F_MRG_RXBUF)) {
+			eth_dev->rx_pkt_burst = &virtio_recv_mergeable_pkts;
+		} else {
 		eth_dev->rx_pkt_burst = &virtio_recv_pkts_packed;
+		}
 	} else if (hw->use_simple_rx) {
 		PMD_INIT_LOG(INFO, "virtio: using simple Rx path on port %u",
 			eth_dev->data->port_id);
diff --git a/drivers/net/virtio/virtio_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
index 90a0e306f..35c375cd5 100644
--- a/drivers/net/virtio/virtio_rxtx.c
+++ b/drivers/net/virtio/virtio_rxtx.c
@@ -90,6 +90,80 @@ vq_ring_free_chain(struct virtqueue *vq, uint16_t desc_idx)
 	dp->next = VQ_RING_DESC_CHAIN_END;
 }
 
+static void
+virtio_refill_packed(struct virtqueue *vq, uint16_t used_idx,
+		     struct virtnet_rx *rxvq)
+{
+	struct vq_desc_extra *dxp;
+	struct vring_desc_packed *descs = vq->vq_ring.desc_packed;
+	struct vring_desc_packed *desc;
+	struct rte_mbuf *nmb;
+
+	nmb = rte_mbuf_raw_alloc(rxvq->mpool);
+	if (unlikely(nmb == NULL)) {
+		struct rte_eth_dev *dev
+			= &rte_eth_devices[rxvq->port_id];
+		dev->data->rx_mbuf_alloc_failed++;
+		return;
+	}
+
+	desc = &descs[used_idx];
+
+	dxp = &vq->vq_descx[used_idx];
+
+	dxp->cookie = nmb;
+	dxp->ndescs = 1;
+
+	desc->addr = VIRTIO_MBUF_ADDR(nmb, vq) +
+		RTE_PKTMBUF_HEADROOM - vq->hw->vtnet_hdr_size;
+	desc->len = nmb->buf_len - RTE_PKTMBUF_HEADROOM +
+		vq->hw->vtnet_hdr_size;
+	desc->flags |= VRING_DESC_F_WRITE;
+}
+
+static uint16_t
+virtqueue_dequeue_burst_rx_packed(struct virtqueue *vq,
+				  struct rte_mbuf **rx_pkts,
+				  uint32_t *len,
+				  uint16_t num,
+				  struct virtnet_rx *rx_queue)
+{
+	struct rte_mbuf *cookie;
+	uint16_t used_idx;
+	uint16_t id;
+	struct vring_desc_packed *desc;
+	uint16_t i;
+
+	for (i = 0; i < num; i++) {
+		used_idx = vq->vq_used_cons_idx;
+		desc = &vq->vq_ring.desc_packed[used_idx];
+		if (!desc_is_used(desc, &vq->vq_ring))
+			return i;
+		len[i] = desc->len;
+		id = desc[used_idx].index;
+		cookie = (struct rte_mbuf *)vq->vq_descx[id].cookie;
+
+		if (unlikely(cookie == NULL)) {
+			PMD_DRV_LOG(ERR, "vring descriptor with no mbuf cookie at %u",
+				vq->vq_used_cons_idx);
+			break;
+		}
+		rte_prefetch0(cookie);
+		rte_packet_prefetch(rte_pktmbuf_mtod(cookie, void *));
+		rx_pkts[i] = cookie;
+
+		virtio_refill_packed(vq, used_idx, rx_queue);
+
+		if (++vq->vq_used_cons_idx >= vq->vq_nentries) {
+			vq->vq_used_cons_idx = 0;
+			vq->vq_ring.used_wrap_counter ^= 1;
+		}
+		vq->vq_descx[id].cookie = NULL;
+	}
+
+	return i;
+}
+
 static uint16_t
 virtqueue_dequeue_burst_rx(struct virtqueue *vq, struct rte_mbuf **rx_pkts,
 			   uint32_t *len, uint16_t num)
@@ -1476,12 +1550,16 @@ virtio_recv_mergeable_pkts(void *rx_queue,
 	uint16_t extra_idx;
 	uint32_t seg_res;
 	uint32_t hdr_size;
+	uint32_t rx_num = 0;
 
 	nb_rx = 0;
 	if (unlikely(hw->started == 0))
 		return nb_rx;
 
-	nb_used = VIRTQUEUE_NUSED(vq);
+	if (vtpci_packed_queue(vq->hw))
+		nb_used = VIRTIO_MBUF_BURST_SZ;
+	else
+		nb_used = VIRTQUEUE_NUSED(vq);
 
 	virtio_rmb();
 
@@ -1494,13 +1572,21 @@ virtio_recv_mergeable_pkts(void *rx_queue,
 	seg_res = 0;
 	hdr_size = hw->vtnet_hdr_size;
 
+	vq->vq_used_idx = vq->vq_used_cons_idx;
+
 	while (i < nb_used) {
 		struct virtio_net_hdr_mrg_rxbuf *header;
 
 		if (nb_rx == nb_pkts)
 			break;
 
-		num = virtqueue_dequeue_burst_rx(vq, rcv_pkts, len, 1);
+		if (vtpci_packed_queue(vq->hw))
+			num = virtqueue_dequeue_burst_rx_packed(vq, rcv_pkts,
+				len, 1, (struct virtnet_rx *)rx_queue);
+		else
+			num = virtqueue_dequeue_burst_rx(vq, rcv_pkts, len, 1);
+		if (num == 0)
+			return nb_rx;
 		if (num != 1)
 			continue;
 
@@ -1552,12 +1638,15 @@ virtio_recv_mergeable_pkts(void *rx_queue,
 			 */
 			uint16_t  rcv_cnt =
 				RTE_MIN(seg_res, RTE_DIM(rcv_pkts));
-			if (likely(VIRTQUEUE_NUSED(vq) >= rcv_cnt)) {
-				uint32_t rx_num =
-					virtqueue_dequeue_burst_rx(vq,
-					rcv_pkts, len, rcv_cnt);
-				i += rx_num;
-				rcv_cnt = rx_num;
+			if (vtpci_packed_queue(vq->hw)) {
+				if (likely(vq->vq_free_cnt >= rcv_cnt)) {	
+					rx_num = virtqueue_dequeue_burst_rx_packed(vq,
+						     rcv_pkts, len, rcv_cnt,
+						     (struct virtnet_rx *)rx_queue);
+				}
+			} else if (likely(VIRTQUEUE_NUSED(vq) >= rcv_cnt)) {
+				rx_num = virtqueue_dequeue_burst_rx(vq,
+					      rcv_pkts, len, rcv_cnt);
 			} else {
 				PMD_RX_LOG(ERR,
 					   "No enough segments for packet.");
@@ -1566,6 +1655,8 @@ virtio_recv_mergeable_pkts(void *rx_queue,
 				rxvq->stats.errors++;
 				break;
 			}
+			i += rx_num;
+			rcv_cnt = rx_num;
 
 			extra_idx = 0;
 
@@ -1599,6 +1690,15 @@ virtio_recv_mergeable_pkts(void *rx_queue,
 
 	rxvq->stats.packets += nb_rx;
 
+	if (vtpci_packed_queue(vq->hw)) {
+		if (unlikely(virtqueue_kick_prepare_packed(vq)) &&
+		    likely(nb_rx)) {
+			virtqueue_notify(vq);
+			PMD_RX_LOG(DEBUG, "Notified");
+		}
+		return nb_rx;
+	}
+
 	/* Allocate new mbuf for the used descriptor */
 	while (likely(!virtqueue_full(vq))) {
 		new_mbuf = rte_mbuf_raw_alloc(rxvq->mpool);
@@ -1618,7 +1718,6 @@ virtio_recv_mergeable_pkts(void *rx_queue,
 
 	if (likely(nb_enqueued)) {
 		vq_update_avail_idx(vq);
-
 		if (unlikely(virtqueue_kick_prepare(vq))) {
 			virtqueue_notify(vq);
 			PMD_RX_LOG(DEBUG, "Notified");
diff --git a/drivers/net/virtio/virtqueue.h b/drivers/net/virtio/virtqueue.h
index ffa2d8f92..15df6a050 100644
--- a/drivers/net/virtio/virtqueue.h
+++ b/drivers/net/virtio/virtqueue.h
@@ -171,6 +171,7 @@ struct virtqueue {
 	 * trails vq_ring.used->idx.
 	 */
 	uint16_t vq_used_cons_idx;
+	uint16_t vq_used_idx;
 	uint16_t vq_nentries;  /**< vring desc numbers */
 	uint16_t vq_free_cnt;  /**< num of desc available */
 	uint16_t vq_avail_idx; /**< sync until needed */
-- 
2.17.1

[PATCH v6 09/11] net/virtio: add virtio send command packed queue support

[Jens Freimann]

Use packed virtqueue format when reading and writing descriptors
to/from the ring.

Signed-off-by: Jens Freimann <jfreimann@redhat.com>
---
 drivers/net/virtio/virtio_ethdev.c | 90 ++++++++++++++++++++++++++++++
 1 file changed, 90 insertions(+)

diff --git a/drivers/net/virtio/virtio_ethdev.c b/drivers/net/virtio/virtio_ethdev.c
index c4ef095ed..c1d95141d 100644
--- a/drivers/net/virtio/virtio_ethdev.c
+++ b/drivers/net/virtio/virtio_ethdev.c
@@ -141,6 +141,90 @@ static const struct rte_virtio_xstats_name_off rte_virtio_txq_stat_strings[] = {
 
 struct virtio_hw_internal virtio_hw_internal[RTE_MAX_ETHPORTS];
 
+static struct virtio_pmd_ctrl *
+virtio_pq_send_command(struct virtnet_ctl *cvq, struct virtio_pmd_ctrl *ctrl,
+		       int *dlen, int pkt_num)
+{
+	struct virtqueue *vq = cvq->vq;
+	int head;
+	struct vring_desc_packed *desc = vq->vq_ring.desc_packed;
+	struct virtio_pmd_ctrl *result;
+	int wrap_counter;
+	int sum = 0;
+	int k;
+
+	/*
+	 * Format is enforced in qemu code:
+	 * One TX packet for header;
+	 * At least one TX packet per argument;
+	 * One RX packet for ACK.
+	 */
+	head = vq->vq_avail_idx;
+	wrap_counter = vq->vq_ring.avail_wrap_counter;
+	desc[head].flags = VRING_DESC_F_NEXT;
+	desc[head].addr = cvq->virtio_net_hdr_mem;
+	desc[head].len = sizeof(struct virtio_net_ctrl_hdr);
+	vq->vq_free_cnt--;
+	if (++vq->vq_avail_idx >= vq->vq_nentries) {
+		vq->vq_avail_idx -= vq->vq_nentries;
+		vq->vq_ring.avail_wrap_counter ^= 1;
+	}
+
+	for (k = 0; k < pkt_num; k++) {
+		desc[vq->vq_avail_idx].addr = cvq->virtio_net_hdr_mem
+			+ sizeof(struct virtio_net_ctrl_hdr)
+			+ sizeof(ctrl->status) + sizeof(uint8_t) * sum;
+		desc[vq->vq_avail_idx].len = dlen[k];
+		desc[vq->vq_avail_idx].flags = VRING_DESC_F_NEXT;
+		sum += dlen[k];
+		vq->vq_free_cnt--;
+		_set_desc_avail(&desc[vq->vq_avail_idx],
+				vq->vq_ring.avail_wrap_counter);
+		rte_smp_wmb();
+		vq->vq_free_cnt--;
+		if (++vq->vq_avail_idx >= vq->vq_nentries) {
+			vq->vq_avail_idx -= vq->vq_nentries;
+			vq->vq_ring.avail_wrap_counter ^= 1;
+		}
+	}
+
+
+	desc[vq->vq_avail_idx].addr = cvq->virtio_net_hdr_mem
+		+ sizeof(struct virtio_net_ctrl_hdr);
+	desc[vq->vq_avail_idx].len = sizeof(ctrl->status);
+	desc[vq->vq_avail_idx].flags = VRING_DESC_F_WRITE;
+	_set_desc_avail(&desc[vq->vq_avail_idx],
+			vq->vq_ring.avail_wrap_counter);
+	_set_desc_avail(&desc[head], wrap_counter);
+	rte_smp_wmb();
+
+	vq->vq_free_cnt--;
+	if (++vq->vq_avail_idx >= vq->vq_nentries) {
+		vq->vq_avail_idx -= vq->vq_nentries;
+		vq->vq_ring.avail_wrap_counter ^= 1;
+	}
+
+	virtqueue_notify(vq);
+
+	/* wait for used descriptors in virtqueue */
+	do {
+		rte_rmb();
+		usleep(100);
+	} while (!_desc_is_used(&desc[head]));
+
+	/* now get used descriptors */
+	while(desc_is_used(&desc[vq->vq_used_cons_idx], &vq->vq_ring)) {
+		vq->vq_free_cnt++;
+		if (++vq->vq_used_cons_idx >= vq->vq_nentries) {
+			vq->vq_used_cons_idx -= vq->vq_nentries;
+			vq->vq_ring.used_wrap_counter ^= 1;
+		}
+	}
+
+	result = cvq->virtio_net_hdr_mz->addr;
+	return result;
+}
+
 static int
 virtio_send_command(struct virtnet_ctl *cvq, struct virtio_pmd_ctrl *ctrl,
 		int *dlen, int pkt_num)
@@ -174,6 +258,11 @@ virtio_send_command(struct virtnet_ctl *cvq, struct virtio_pmd_ctrl *ctrl,
 	memcpy(cvq->virtio_net_hdr_mz->addr, ctrl,
 		sizeof(struct virtio_pmd_ctrl));
 
+	if (vtpci_packed_queue(vq->hw)) {
+		result = virtio_pq_send_command(cvq, ctrl, dlen, pkt_num);
+		goto out_unlock;
+	}
+
 	/*
 	 * Format is enforced in qemu code:
 	 * One TX packet for header;
@@ -245,6 +334,7 @@ virtio_send_command(struct virtnet_ctl *cvq, struct virtio_pmd_ctrl *ctrl,
 
 	result = cvq->virtio_net_hdr_mz->addr;
 
+out_unlock:
 	rte_spinlock_unlock(&cvq->lock);
 	return result->status;
 }
-- 
2.17.1

[PATCH v6 10/11] net/virtio-user: add option to use packed queues

[Jens Freimann]

From: Yuanhan Liu <yuanhan.liu@linux.intel.com>

Add option to enable packed queue support for virtio-user
devices.

Signed-off-by: Yuanhan Liu <yuanhan.liu@linux.intel.com>
---
 drivers/net/virtio/virtio_user/virtio_user_dev.c | 10 ++++++++--
 drivers/net/virtio/virtio_user/virtio_user_dev.h |  2 +-
 drivers/net/virtio/virtio_user_ethdev.c          | 14 +++++++++++++-
 3 files changed, 22 insertions(+), 4 deletions(-)

diff --git a/drivers/net/virtio/virtio_user/virtio_user_dev.c b/drivers/net/virtio/virtio_user/virtio_user_dev.c
index 869e96f87..0556bf7af 100644
--- a/drivers/net/virtio/virtio_user/virtio_user_dev.c
+++ b/drivers/net/virtio/virtio_user/virtio_user_dev.c
@@ -391,12 +391,13 @@ virtio_user_dev_setup(struct virtio_user_dev *dev)
 	 1ULL << VIRTIO_NET_F_GUEST_TSO4	|	\
 	 1ULL << VIRTIO_NET_F_GUEST_TSO6	|	\
 	 1ULL << VIRTIO_F_IN_ORDER		|	\
-	 1ULL << VIRTIO_F_VERSION_1)
+	 1ULL << VIRTIO_F_VERSION_1		|	\
+	 1ULL << VIRTIO_F_RING_PACKED)
 
 int
 virtio_user_dev_init(struct virtio_user_dev *dev, char *path, int queues,
 		     int cq, int queue_size, const char *mac, char **ifname,
-		     int mrg_rxbuf, int in_order)
+		     int mrg_rxbuf, int in_order, int packed_vq)
 {
 	pthread_mutex_init(&dev->mutex, NULL);
 	snprintf(dev->path, PATH_MAX, "%s", path);
@@ -451,6 +452,11 @@ virtio_user_dev_init(struct virtio_user_dev *dev, char *path, int queues,
 		dev->unsupported_features |= (1ull << VIRTIO_F_IN_ORDER);
 	}
 
+	if (!packed_vq) {
+		dev->device_features &= ~(1ull << VIRTIO_F_RING_PACKED);
+		dev->unsupported_features |= (1ull << VIRTIO_F_RING_PACKED);
+	}
+
 	if (dev->mac_specified) {
 		dev->device_features |= (1ull << VIRTIO_NET_F_MAC);
 	} else {
diff --git a/drivers/net/virtio/virtio_user/virtio_user_dev.h b/drivers/net/virtio/virtio_user/virtio_user_dev.h
index d6e0e137b..7f46ba1d9 100644
--- a/drivers/net/virtio/virtio_user/virtio_user_dev.h
+++ b/drivers/net/virtio/virtio_user/virtio_user_dev.h
@@ -49,7 +49,7 @@ int virtio_user_start_device(struct virtio_user_dev *dev);
 int virtio_user_stop_device(struct virtio_user_dev *dev);
 int virtio_user_dev_init(struct virtio_user_dev *dev, char *path, int queues,
 			 int cq, int queue_size, const char *mac, char **ifname,
-			 int mrg_rxbuf, int in_order);
+			 int mrg_rxbuf, int in_order, int packed_vq);
 void virtio_user_dev_uninit(struct virtio_user_dev *dev);
 void virtio_user_handle_cq(struct virtio_user_dev *dev, uint16_t queue_idx);
 uint8_t virtio_user_handle_mq(struct virtio_user_dev *dev, uint16_t q_pairs);
diff --git a/drivers/net/virtio/virtio_user_ethdev.c b/drivers/net/virtio/virtio_user_ethdev.c
index 525d16cab..72ac86186 100644
--- a/drivers/net/virtio/virtio_user_ethdev.c
+++ b/drivers/net/virtio/virtio_user_ethdev.c
@@ -364,6 +364,8 @@ static const char *valid_args[] = {
 	VIRTIO_USER_ARG_MRG_RXBUF,
 #define VIRTIO_USER_ARG_IN_ORDER       "in_order"
 	VIRTIO_USER_ARG_IN_ORDER,
+#define VIRTIO_USER_ARG_PACKED_VQ "packed_vq"
+	VIRTIO_USER_ARG_PACKED_VQ,
 	NULL
 };
 
@@ -473,6 +475,7 @@ virtio_user_pmd_probe(struct rte_vdev_device *dev)
 	char *ifname = NULL;
 	char *mac_addr = NULL;
 	int ret = -1;
+	uint64_t packed_vq = 0;
 
 	kvlist = rte_kvargs_parse(rte_vdev_device_args(dev), valid_args);
 	if (!kvlist) {
@@ -556,6 +559,15 @@ virtio_user_pmd_probe(struct rte_vdev_device *dev)
 		cq = 1;
 	}
 
+	if (rte_kvargs_count(kvlist, VIRTIO_USER_ARG_PACKED_VQ) == 1) {
+		if (rte_kvargs_process(kvlist, VIRTIO_USER_ARG_PACKED_VQ,
+				       &get_integer_arg, &packed_vq) < 0) {
+			PMD_INIT_LOG(ERR, "error to parse %s",
+				     VIRTIO_USER_ARG_PACKED_VQ);
+			goto end;
+		}
+	}
+
 	if (queues > 1 && cq == 0) {
 		PMD_INIT_LOG(ERR, "multi-q requires ctrl-q");
 		goto end;
@@ -603,7 +615,7 @@ virtio_user_pmd_probe(struct rte_vdev_device *dev)
 			vu_dev->is_server = false;
 		if (virtio_user_dev_init(hw->virtio_user_dev, path, queues, cq,
 				 queue_size, mac_addr, &ifname, mrg_rxbuf,
-				 in_order) < 0) {
+				 in_order, packed_vq) < 0) {
 			PMD_INIT_LOG(ERR, "virtio_user_dev_init fails");
 			virtio_user_eth_dev_free(eth_dev);
 			goto end;
-- 
2.17.1

[PATCH v6 11/11] net/virtio: enable packed virtqueues by default

[Jens Freimann]

Signed-off-by: Jens Freimann <jfreimann@redhat.com>
---
 drivers/net/virtio/virtio_ethdev.h | 1 +
 1 file changed, 1 insertion(+)

diff --git a/drivers/net/virtio/virtio_ethdev.h b/drivers/net/virtio/virtio_ethdev.h
index 25eaff224..2c8a69cbf 100644
--- a/drivers/net/virtio/virtio_ethdev.h
+++ b/drivers/net/virtio/virtio_ethdev.h
@@ -34,6 +34,7 @@
 	 1u << VIRTIO_RING_F_INDIRECT_DESC |    \
 	 1ULL << VIRTIO_F_VERSION_1       |	\
 	 1ULL << VIRTIO_F_IN_ORDER        |	\
+	 1ULL << VIRTIO_F_RING_PACKED	  |	\
 	 1ULL << VIRTIO_F_IOMMU_PLATFORM)
 
 #define VIRTIO_PMD_SUPPORTED_GUEST_FEATURES	\
-- 
2.17.1

Re: [PATCH v6 06/11] net/virtio: implement transmit path for packed queues

[Tiwei Bie]

On Fri, Sep 21, 2018 at 12:33:03PM +0200, Jens Freimann wrote:
[...]
>  
>  static inline int
> -desc_is_used(struct vring_desc_packed *desc, struct vring *vr)
> +_desc_is_used(struct vring_desc_packed *desc)
>  {
>  	uint16_t used, avail;
>  
>  	used = !!(desc->flags & VRING_DESC_F_USED(1));
>  	avail = !!(desc->flags & VRING_DESC_F_AVAIL(1));
>  
> -	return used == avail && used == vr->used_wrap_counter;
> +	return used == avail;
> +
> +}
> +
> +static inline int
> +desc_is_used(struct vring_desc_packed *desc, struct vring *vr)
> +{
> +	uint16_t used;
> +
> +	used = !!(desc->flags & VRING_DESC_F_USED(1));
> +
> +	return _desc_is_used(desc) && used == vr->used_wrap_counter;
>  }
>  
>  /* The standard layout for the ring is a continuous chunk of memory which
> diff --git a/drivers/net/virtio/virtio_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
> index eb891433e..ea6300563 100644
> --- a/drivers/net/virtio/virtio_rxtx.c
> +++ b/drivers/net/virtio/virtio_rxtx.c
> @@ -38,6 +38,7 @@
>  #define  VIRTIO_DUMP_PACKET(m, len) do { } while (0)
>  #endif
>  
> +
>  int
>  virtio_dev_rx_queue_done(void *rxq, uint16_t offset)
>  {
> @@ -165,6 +166,31 @@ virtqueue_dequeue_rx_inorder(struct virtqueue *vq,
>  #endif
>  
>  /* Cleanup from completed transmits. */
> +static void
> +virtio_xmit_cleanup_packed(struct virtqueue *vq)
> +{
> +	uint16_t idx;
> +	uint16_t size = vq->vq_nentries;
> +	struct vring_desc_packed *desc = vq->vq_ring.desc_packed;
> +	struct vq_desc_extra *dxp;
> +
> +	idx = vq->vq_used_cons_idx;
> +	while (_desc_is_used(&desc[idx]) &&

We can't just compare the AVAIL bit and USED bit to
check whether a desc is used.

> +	       vq->vq_free_cnt < size) {
> +		dxp = &vq->vq_descx[idx];

The code is still assuming the descs will be written
back by device in order. The vq->vq_descx[] needs to
be managed e.g. as a list to support the out-of-order
processing. IOW, we can't assume vq->vq_descx[idx]
is corresponding to desc[idx] when device may write
back the descs out of order.

> +		vq->vq_free_cnt += dxp->ndescs;
> +		idx += dxp->ndescs;
> +		idx = idx >= size ? idx - size : idx;
> +		if (idx == 0) {
> +			vq->vq_ring.used_wrap_counter ^= 1;
> +		}
> +		if (dxp->cookie != NULL) {
> +			rte_pktmbuf_free(dxp->cookie);
> +			dxp->cookie = NULL;
> +		}
> +	}
> +}
[...]

Re: [PATCH v6 00/11] implement packed virtqueues

[Tiwei Bie]

On Fri, Sep 21, 2018 at 12:32:57PM +0200, Jens Freimann wrote:
> This is a basic implementation of packed virtqueues as specified in the
> Virtio 1.1 draft. A compiled version of the current draft is available
> at https://github.com/oasis-tcs/virtio-docs.git (or as .pdf at
> https://github.com/oasis-tcs/virtio-docs/blob/master/virtio-v1.1-packed-wd10.pdf
> 
> A packed virtqueue is different from a split virtqueue in that it
> consists of only a single descriptor ring that replaces available and
> used ring, index and descriptor buffer.
> 
> Each descriptor is readable and writable and has a flags field. These flags
> will mark if a descriptor is available or used.  To detect new available descriptors
> even after the ring has wrapped, device and driver each have a
> single-bit wrap counter that is flipped from 0 to 1 and vice versa every time
> the last descriptor in the ring is used/made available.
> 
> The idea behind this is to 1. improve performance by avoiding cache misses
> and 2. be easier for devices to implement.
> 
> Regarding performance: with these patches I get 21.13 Mpps on my system
> as compared to 18.8 Mpps with the virtio 1.0 code. Packet size was 64

Did you enable multiple-queue and use multiple cores on
vhost side? If not, I guess the above performance gain
is the gain in vhost side instead of virtio side.

If you use more cores on vhost side or virtio side, will
you see any performance changes?

Did you do any performance test with the kernel vhost-net
backend (with zero-copy enabled and disabled)? I think we
also need some performance data for these two cases. And
it can help us to make sure that it works with the kernel
backends.

And for the "virtio-PMD + vhost-PMD" test cases, I think
we need below performance data:

#1. The maximum 1 core performance of virtio PMD when using split ring.
#2. The maximum 1 core performance of virtio PMD when using packed ring.
#3. The maximum 1 core performance of vhost PMD when using split ring.
#4. The maximum 1 core performance of vhost PMD when using packed ring.

And then we can have a clear understanding of the
performance gain in DPDK with packed ring.

And FYI, the maximum 1 core performance of virtio PMD
can be got in below steps:

1. Launch vhost-PMD with multiple queues, and use multiple
   CPU cores for forwarding.
2. Launch virtio-PMD with multiple queues and use 1 CPU
   core for forwarding.
3. Repeat above two steps with adding more CPU cores
   for forwarding in vhost-PMD side until we can't see
   performance increase anymore.

Besides, I just did a quick glance at the Tx implementation,
it still assumes the descs will be written back in order
by device. You can find more details from my comments on
that patch.

Thanks



> bytes, 0.05% acceptable loss.  Test setup is described as in
> http://dpdk.org/doc/guides/howto/pvp_reference_benchmark.html
> 
> Packet generator:
> MoonGen
> Intel(R) Xeon(R) CPU E5-2665 0 @ 2.40GHz
> Intel X710 NIC
> RHEL 7.4
> 
> Device under test:
> Intel(R) Xeon(R) CPU E5-2667 v4 @ 3.20GHz
> Intel X710 NIC
> RHEL 7.4
> 
> VM on DuT: RHEL7.4
> 
> I plan to do more performance test with bigger frame sizes.
> 
> changes from v5->v6:
> * fix VIRTQUEUE_DUMP macro
> * rework mergeable rx buffer path, support out of order (not sure if I
>   need a .next field to support chains) 
> * move wmb in virtio_receive_pkts_packed() (Gavin)
> * rename to virtio_init_split/_packed (Maxime)
> * add support for ctrl virtqueues (Tiwei, thx Max for fixing)
> * rework tx path to support update_packet_stats and
>   virtqueue_xmit_offload, TODO: merge with split-ring code to
>   avoid a lot of duplicate code
> * remove unnecessary check for avoiding to call VIRTQUEUE_DUMP (Maxime)
> 
> changes from v4->v5:
> * fix VIRTQUEUE_DUMP macro
> * fix wrap counter logic in transmit and receive functions  
> 
> changes from v3->v4:
> * added helpers to increment index and set available/used flags
> * driver keeps track of number of descriptors used
> * change logic in set_rxtx_funcs()
> * add patch for ctrl virtqueue with support for packed virtqueues
> * rename virtio-1.1.h to virtio-packed.h
> * fix wrong sizeof() in "vhost: vring address setup for packed queues"
> * fix coding style of function definition in "net/virtio: add packed
>   virtqueue helpers"
> * fix padding in vring_size()
> * move patches to enable packed virtqueues end of series
> * v4 has two open problems: I'm sending it out anyway for feedback/help:
>  * when VIRTIO_NET_F_MRG_RXBUF enabled only 128 packets are send in
>    guest, i.e. when ring is full for the first time. I suspect a bug in
>    setting the avail/used flags
> 
> changes from v2->v3:
> * implement event suppression
> * add code do dump packed virtqueues
> * don't use assert in vhost code
> * rename virtio-user parameter to packed-vq
> * support rxvf flush
> 
> changes from v1->v2:
> * don't use VIRTQ_DESC_F_NEXT in used descriptors (Jason)
> * no rte_panice() in guest triggerable code (Maxime)
> * use unlikely when checking for vq (Maxime)
> * rename everything from _1_1 to _packed  (Yuanhan)
> * add two more patches to implement mergeable receive buffers
> 
> *** BLURB HERE ***
> 
> Jens Freimann (10):
>   net/virtio: vring init for packed queues
>   net/virtio: add packed virtqueue defines
>   net/virtio: add packed virtqueue helpers
>   net/virtio: flush packed receive virtqueues
>   net/virtio: dump packed virtqueue data
>   net/virtio: implement transmit path for packed queues
>   net/virtio: implement receive path for packed queues
>   net/virtio: add support for mergeable buffers with packed virtqueues
>   net/virtio: add virtio send command packed queue support
>   net/virtio: enable packed virtqueues by default
> 
> Yuanhan Liu (1):
>   net/virtio-user: add option to use packed queues
> 
>  drivers/net/virtio/virtio_ethdev.c            | 135 ++++-
>  drivers/net/virtio/virtio_ethdev.h            |   5 +
>  drivers/net/virtio/virtio_pci.h               |   8 +
>  drivers/net/virtio/virtio_ring.h              |  96 +++-
>  drivers/net/virtio/virtio_rxtx.c              | 490 +++++++++++++++++-
>  .../net/virtio/virtio_user/virtio_user_dev.c  |  10 +-
>  .../net/virtio/virtio_user/virtio_user_dev.h  |   2 +-
>  drivers/net/virtio/virtio_user_ethdev.c       |  14 +-
>  drivers/net/virtio/virtqueue.c                |  21 +
>  drivers/net/virtio/virtqueue.h                |  50 +-
>  10 files changed, 796 insertions(+), 35 deletions(-)
> 
> -- 
> 2.17.1
> 

Re: [PATCH v6 09/11] net/virtio: add virtio send command packed queue support

[Tiwei Bie]

On Fri, Sep 21, 2018 at 12:33:06PM +0200, Jens Freimann wrote:
> Use packed virtqueue format when reading and writing descriptors
> to/from the ring.
> 
> Signed-off-by: Jens Freimann <jfreimann@redhat.com>
> ---
>  drivers/net/virtio/virtio_ethdev.c | 90 ++++++++++++++++++++++++++++++
>  1 file changed, 90 insertions(+)
> 
> diff --git a/drivers/net/virtio/virtio_ethdev.c b/drivers/net/virtio/virtio_ethdev.c
> index c4ef095ed..c1d95141d 100644
> --- a/drivers/net/virtio/virtio_ethdev.c
> +++ b/drivers/net/virtio/virtio_ethdev.c

The support for virtio-user is missing. And virtio-user will
break when packed ring is enabled. FYI, below is the relvelant
code in virtio-user:

https://github.com/DPDK/dpdk/blob/55d6bb67c9b3/drivers/net/virtio/virtio_user/virtio_user_dev.c#L584-L606


> @@ -141,6 +141,90 @@ static const struct rte_virtio_xstats_name_off rte_virtio_txq_stat_strings[] = {
>  
>  struct virtio_hw_internal virtio_hw_internal[RTE_MAX_ETHPORTS];
>  
> +static struct virtio_pmd_ctrl *
> +virtio_pq_send_command(struct virtnet_ctl *cvq, struct virtio_pmd_ctrl *ctrl,
> +		       int *dlen, int pkt_num)
> +{
> +	struct virtqueue *vq = cvq->vq;
> +	int head;
> +	struct vring_desc_packed *desc = vq->vq_ring.desc_packed;
> +	struct virtio_pmd_ctrl *result;
> +	int wrap_counter;
> +	int sum = 0;
> +	int k;
> +
> +	/*
> +	 * Format is enforced in qemu code:
> +	 * One TX packet for header;
> +	 * At least one TX packet per argument;
> +	 * One RX packet for ACK.
> +	 */
> +	head = vq->vq_avail_idx;
> +	wrap_counter = vq->vq_ring.avail_wrap_counter;
> +	desc[head].flags = VRING_DESC_F_NEXT;
> +	desc[head].addr = cvq->virtio_net_hdr_mem;
> +	desc[head].len = sizeof(struct virtio_net_ctrl_hdr);
> +	vq->vq_free_cnt--;
> +	if (++vq->vq_avail_idx >= vq->vq_nentries) {
> +		vq->vq_avail_idx -= vq->vq_nentries;
> +		vq->vq_ring.avail_wrap_counter ^= 1;
> +	}
> +
> +	for (k = 0; k < pkt_num; k++) {
> +		desc[vq->vq_avail_idx].addr = cvq->virtio_net_hdr_mem
> +			+ sizeof(struct virtio_net_ctrl_hdr)
> +			+ sizeof(ctrl->status) + sizeof(uint8_t) * sum;
> +		desc[vq->vq_avail_idx].len = dlen[k];
> +		desc[vq->vq_avail_idx].flags = VRING_DESC_F_NEXT;
> +		sum += dlen[k];
> +		vq->vq_free_cnt--;
> +		_set_desc_avail(&desc[vq->vq_avail_idx],
> +				vq->vq_ring.avail_wrap_counter);
> +		rte_smp_wmb();
> +		vq->vq_free_cnt--;
> +		if (++vq->vq_avail_idx >= vq->vq_nentries) {
> +			vq->vq_avail_idx -= vq->vq_nentries;
> +			vq->vq_ring.avail_wrap_counter ^= 1;
> +		}
> +	}
> +
> +
> +	desc[vq->vq_avail_idx].addr = cvq->virtio_net_hdr_mem
> +		+ sizeof(struct virtio_net_ctrl_hdr);
> +	desc[vq->vq_avail_idx].len = sizeof(ctrl->status);
> +	desc[vq->vq_avail_idx].flags = VRING_DESC_F_WRITE;
> +	_set_desc_avail(&desc[vq->vq_avail_idx],
> +			vq->vq_ring.avail_wrap_counter);
> +	_set_desc_avail(&desc[head], wrap_counter);
> +	rte_smp_wmb();
> +
> +	vq->vq_free_cnt--;
> +	if (++vq->vq_avail_idx >= vq->vq_nentries) {
> +		vq->vq_avail_idx -= vq->vq_nentries;
> +		vq->vq_ring.avail_wrap_counter ^= 1;
> +	}
> +
> +	virtqueue_notify(vq);
> +
> +	/* wait for used descriptors in virtqueue */
> +	do {
> +		rte_rmb();
> +		usleep(100);
> +	} while (!_desc_is_used(&desc[head]));
> +
> +	/* now get used descriptors */
> +	while(desc_is_used(&desc[vq->vq_used_cons_idx], &vq->vq_ring)) {
> +		vq->vq_free_cnt++;
> +		if (++vq->vq_used_cons_idx >= vq->vq_nentries) {
> +			vq->vq_used_cons_idx -= vq->vq_nentries;
> +			vq->vq_ring.used_wrap_counter ^= 1;
> +		}
> +	}
> +
> +	result = cvq->virtio_net_hdr_mz->addr;
> +	return result;
> +}
> +
>  static int
>  virtio_send_command(struct virtnet_ctl *cvq, struct virtio_pmd_ctrl *ctrl,
>  		int *dlen, int pkt_num)
> @@ -174,6 +258,11 @@ virtio_send_command(struct virtnet_ctl *cvq, struct virtio_pmd_ctrl *ctrl,
>  	memcpy(cvq->virtio_net_hdr_mz->addr, ctrl,
>  		sizeof(struct virtio_pmd_ctrl));
>  
> +	if (vtpci_packed_queue(vq->hw)) {
> +		result = virtio_pq_send_command(cvq, ctrl, dlen, pkt_num);
> +		goto out_unlock;
> +	}
> +
>  	/*
>  	 * Format is enforced in qemu code:
>  	 * One TX packet for header;
> @@ -245,6 +334,7 @@ virtio_send_command(struct virtnet_ctl *cvq, struct virtio_pmd_ctrl *ctrl,
>  
>  	result = cvq->virtio_net_hdr_mz->addr;
>  
> +out_unlock:
>  	rte_spinlock_unlock(&cvq->lock);
>  	return result->status;
>  }
> -- 
> 2.17.1
> 

Re: [PATCH v6 06/11] net/virtio: implement transmit path for packed queues

[Jens Freimann]

On Fri, Sep 21, 2018 at 08:26:58PM +0800, Tiwei Bie wrote:
>On Fri, Sep 21, 2018 at 12:33:03PM +0200, Jens Freimann wrote:
>[...]
>>
>>  static inline int
>> -desc_is_used(struct vring_desc_packed *desc, struct vring *vr)
>> +_desc_is_used(struct vring_desc_packed *desc)
>>  {
>>  	uint16_t used, avail;
>>
>>  	used = !!(desc->flags & VRING_DESC_F_USED(1));
>>  	avail = !!(desc->flags & VRING_DESC_F_AVAIL(1));
>>
>> -	return used == avail && used == vr->used_wrap_counter;
>> +	return used == avail;
>> +
>> +}
>> +
>> +static inline int
>> +desc_is_used(struct vring_desc_packed *desc, struct vring *vr)
>> +{
>> +	uint16_t used;
>> +
>> +	used = !!(desc->flags & VRING_DESC_F_USED(1));
>> +
>> +	return _desc_is_used(desc) && used == vr->used_wrap_counter;
>>  }
>>
>>  /* The standard layout for the ring is a continuous chunk of memory which
>> diff --git a/drivers/net/virtio/virtio_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
>> index eb891433e..ea6300563 100644
>> --- a/drivers/net/virtio/virtio_rxtx.c
>> +++ b/drivers/net/virtio/virtio_rxtx.c
>> @@ -38,6 +38,7 @@
>>  #define  VIRTIO_DUMP_PACKET(m, len) do { } while (0)
>>  #endif
>>
>> +
>>  int
>>  virtio_dev_rx_queue_done(void *rxq, uint16_t offset)
>>  {
>> @@ -165,6 +166,31 @@ virtqueue_dequeue_rx_inorder(struct virtqueue *vq,
>>  #endif
>>
>>  /* Cleanup from completed transmits. */
>> +static void
>> +virtio_xmit_cleanup_packed(struct virtqueue *vq)
>> +{
>> +	uint16_t idx;
>> +	uint16_t size = vq->vq_nentries;
>> +	struct vring_desc_packed *desc = vq->vq_ring.desc_packed;
>> +	struct vq_desc_extra *dxp;
>> +
>> +	idx = vq->vq_used_cons_idx;
>> +	while (_desc_is_used(&desc[idx]) &&
>
>We can't just compare the AVAIL bit and USED bit to
>check whether a desc is used.

We can't compare with the current wrap counter value as well
because it won't match the flags in descriptors. So check against
used_wrap_counter ^= 1 then?
>
>> +	       vq->vq_free_cnt < size) {
>> +		dxp = &vq->vq_descx[idx];
>
>The code is still assuming the descs will be written
>back by device in order. The vq->vq_descx[] needs to
>be managed e.g. as a list to support the out-of-order
>processing. IOW, we can't assume vq->vq_descx[idx]
>is corresponding to desc[idx] when device may write
>back the descs out of order.

I changed it to not assume this in other spots but missed this one.  I
will check more carefully and add code to make vq_descx entries a list.

Thanks for the review!

regards,
Jens 

Re: [PATCH v6 06/11] net/virtio: implement transmit path for packed queues

[Tiwei Bie]

On Fri, Sep 21, 2018 at 02:37:32PM +0200, Jens Freimann wrote:
> On Fri, Sep 21, 2018 at 08:26:58PM +0800, Tiwei Bie wrote:
> > On Fri, Sep 21, 2018 at 12:33:03PM +0200, Jens Freimann wrote:
> > [...]
> > > 
> > >  static inline int
> > > -desc_is_used(struct vring_desc_packed *desc, struct vring *vr)
> > > +_desc_is_used(struct vring_desc_packed *desc)
> > >  {
> > >  	uint16_t used, avail;
> > > 
> > >  	used = !!(desc->flags & VRING_DESC_F_USED(1));
> > >  	avail = !!(desc->flags & VRING_DESC_F_AVAIL(1));
> > > 
> > > -	return used == avail && used == vr->used_wrap_counter;
> > > +	return used == avail;
> > > +
> > > +}
> > > +
> > > +static inline int
> > > +desc_is_used(struct vring_desc_packed *desc, struct vring *vr)
> > > +{
> > > +	uint16_t used;
> > > +
> > > +	used = !!(desc->flags & VRING_DESC_F_USED(1));
> > > +
> > > +	return _desc_is_used(desc) && used == vr->used_wrap_counter;
> > >  }
> > > 
> > >  /* The standard layout for the ring is a continuous chunk of memory which
> > > diff --git a/drivers/net/virtio/virtio_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
> > > index eb891433e..ea6300563 100644
> > > --- a/drivers/net/virtio/virtio_rxtx.c
> > > +++ b/drivers/net/virtio/virtio_rxtx.c
> > > @@ -38,6 +38,7 @@
> > >  #define  VIRTIO_DUMP_PACKET(m, len) do { } while (0)
> > >  #endif
> > > 
> > > +
> > >  int
> > >  virtio_dev_rx_queue_done(void *rxq, uint16_t offset)
> > >  {
> > > @@ -165,6 +166,31 @@ virtqueue_dequeue_rx_inorder(struct virtqueue *vq,
> > >  #endif
> > > 
> > >  /* Cleanup from completed transmits. */
> > > +static void
> > > +virtio_xmit_cleanup_packed(struct virtqueue *vq)
> > > +{
> > > +	uint16_t idx;
> > > +	uint16_t size = vq->vq_nentries;
> > > +	struct vring_desc_packed *desc = vq->vq_ring.desc_packed;
> > > +	struct vq_desc_extra *dxp;
> > > +
> > > +	idx = vq->vq_used_cons_idx;
> > > +	while (_desc_is_used(&desc[idx]) &&
> > 
> > We can't just compare the AVAIL bit and USED bit to
> > check whether a desc is used.
> 
> We can't compare with the current wrap counter value as well
> because it won't match the flags in descriptors. So check against
> used_wrap_counter ^= 1 then?

I haven't looked into this series yet, so I'm not sure what's
the best way to get the wrap-counter we need here. But, yes,
you need some way to get the wrap-counter we should use here.

> > 
> > > +	       vq->vq_free_cnt < size) {
> > > +		dxp = &vq->vq_descx[idx];
> > 
> > The code is still assuming the descs will be written
> > back by device in order. The vq->vq_descx[] needs to
> > be managed e.g. as a list to support the out-of-order
> > processing. IOW, we can't assume vq->vq_descx[idx]
> > is corresponding to desc[idx] when device may write
> > back the descs out of order.
> 
> I changed it to not assume this in other spots but missed this one.  I
> will check more carefully and add code to make vq_descx entries a list.

After making it support the out-of-order, we may want to do
some performance test for the Tx path only. Because I suspect
we may not be able to get the expected performance improvements
in packed ring due to this when device is faster than driver.

Thanks

Re: [PATCH v6 00/11] implement packed virtqueues

[Jens Freimann]

On Fri, Sep 21, 2018 at 08:32:22PM +0800, Tiwei Bie wrote:
>On Fri, Sep 21, 2018 at 12:32:57PM +0200, Jens Freimann wrote:
>> This is a basic implementation of packed virtqueues as specified in the
>> Virtio 1.1 draft. A compiled version of the current draft is available
>> at https://github.com/oasis-tcs/virtio-docs.git (or as .pdf at
>> https://github.com/oasis-tcs/virtio-docs/blob/master/virtio-v1.1-packed-wd10.pdf
>>
>> A packed virtqueue is different from a split virtqueue in that it
>> consists of only a single descriptor ring that replaces available and
>> used ring, index and descriptor buffer.
>>
>> Each descriptor is readable and writable and has a flags field. These flags
>> will mark if a descriptor is available or used.  To detect new available descriptors
>> even after the ring has wrapped, device and driver each have a
>> single-bit wrap counter that is flipped from 0 to 1 and vice versa every time
>> the last descriptor in the ring is used/made available.
>>
>> The idea behind this is to 1. improve performance by avoiding cache misses
>> and 2. be easier for devices to implement.
>>
>> Regarding performance: with these patches I get 21.13 Mpps on my system
>> as compared to 18.8 Mpps with the virtio 1.0 code. Packet size was 64
>
>Did you enable multiple-queue and use multiple cores on
>vhost side? If not, I guess the above performance gain
>is the gain in vhost side instead of virtio side.

I tested several variations back then and they all looked very good.
But code change a lot meanwhile and I need to do more benchmarking
in any case.

>
>If you use more cores on vhost side or virtio side, will
>you see any performance changes?
>
>Did you do any performance test with the kernel vhost-net
>backend (with zero-copy enabled and disabled)? I think we
>also need some performance data for these two cases. And
>it can help us to make sure that it works with the kernel
>backends.

I tested against vhost-kernel but only to test functionality not
to benchmark. 
>
>And for the "virtio-PMD + vhost-PMD" test cases, I think
>we need below performance data:
>
>#1. The maximum 1 core performance of virtio PMD when using split ring.
>#2. The maximum 1 core performance of virtio PMD when using packed ring.
>#3. The maximum 1 core performance of vhost PMD when using split ring.
>#4. The maximum 1 core performance of vhost PMD when using packed ring.
>
>And then we can have a clear understanding of the
>performance gain in DPDK with packed ring.
>
>And FYI, the maximum 1 core performance of virtio PMD
>can be got in below steps:
>
>1. Launch vhost-PMD with multiple queues, and use multiple
>   CPU cores for forwarding.
>2. Launch virtio-PMD with multiple queues and use 1 CPU
>   core for forwarding.
>3. Repeat above two steps with adding more CPU cores
>   for forwarding in vhost-PMD side until we can't see
>   performance increase anymore.

 Thanks for the suggestions, I'll come back with more
numbers.

>
>Besides, I just did a quick glance at the Tx implementation,
>it still assumes the descs will be written back in order
>by device. You can find more details from my comments on
>that patch.

Saw it and noted. I had hoped to be able to avoid the list but
I see no way around it now. 

Thanks for your review Tiwei!

regards,
Jens