[dpdk-dev] [PATCH 0/4] RFC virtio performance enhancement and cleanups

[dpdk-dev] [PATCH 0/4] RFC virtio performance enhancement and cleanups

[Stephen Hemminger]

These are compile tested only, haven't debugged or checked out the corner
case. Submitted for discussion and future planning.

Stephen Hemminger (4):
  virtio: clean up space checks on xmit
  virtio: don't use unlikely for normal tx stuff
  virtio: use indirect ring elements
  virtio: use any layout on transmit

 drivers/net/virtio/virtio_ethdev.c |  11 ++-
 drivers/net/virtio/virtio_ethdev.h |   4 +-
 drivers/net/virtio/virtio_rxtx.c   | 151 ++++++++++++++++++++++++-------------
 drivers/net/virtio/virtqueue.h     |   8 ++
 4 files changed, 115 insertions(+), 59 deletions(-)

-- 
2.1.4

[dpdk-dev] [PATCH 1/4] virtio: clean up space checks on xmit

[Stephen Hemminger]

The space check for transmit ring only needs a single conditional.
I.e only need to recheck for space if there was no space in first check.

This can help performance and simplifies loop.

Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/net/virtio/virtio_rxtx.c | 66 ++++++++++++++++------------------------
 1 file changed, 27 insertions(+), 39 deletions(-)

diff --git a/drivers/net/virtio/virtio_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
index c5b53bb..5b50ed0 100644
--- a/drivers/net/virtio/virtio_rxtx.c
+++ b/drivers/net/virtio/virtio_rxtx.c
@@ -745,7 +745,6 @@ uint16_t
 virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 {
 	struct virtqueue *txvq = tx_queue;
-	struct rte_mbuf *txm;
 	uint16_t nb_used, nb_tx;
 	int error;
 
@@ -759,57 +758,46 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 	if (likely(nb_used > txvq->vq_nentries - txvq->vq_free_thresh))
 		virtio_xmit_cleanup(txvq, nb_used);
 
-	nb_tx = 0;
+	for (nb_tx = 0; nb_tx < nb_pkts; nb_tx++) {
+		struct rte_mbuf *txm = tx_pkts[nb_tx];
+		int need = txm->nb_segs - txvq->vq_free_cnt + 1;
 
-	while (nb_tx < nb_pkts) {
-		/* Need one more descriptor for virtio header. */
-		int need = tx_pkts[nb_tx]->nb_segs - txvq->vq_free_cnt + 1;
-
-		/*Positive value indicates it need free vring descriptors */
+		/* Positive value indicates it need free vring descriptors */
 		if (unlikely(need > 0)) {
 			nb_used = VIRTQUEUE_NUSED(txvq);
 			virtio_rmb();
 			need = RTE_MIN(need, (int)nb_used);
 
 			virtio_xmit_cleanup(txvq, need);
-			need = (int)tx_pkts[nb_tx]->nb_segs -
-				txvq->vq_free_cnt + 1;
-		}
-
-		/*
-		 * Zero or negative value indicates it has enough free
-		 * descriptors to use for transmitting.
-		 */
-		if (likely(need <= 0)) {
-			txm = tx_pkts[nb_tx];
-
-			/* 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);
-					++nb_tx;
-					continue;
-				}
+			need = txm->nb_segs - txvq->vq_free_cnt + 1;
+			if (unlikely(need > 0)) {
+				PMD_TX_LOG(ERR,
+					   "No free tx descriptors to transmit");
+				break;
 			}
+		}
 
-			/* Enqueue Packet buffers */
-			error = virtqueue_enqueue_xmit(txvq, txm);
+		/* Do VLAN tag insertion */
+		if (unlikely(txm->ol_flags & PKT_TX_VLAN_PKT)) {
+			error = rte_vlan_insert(&txm);
 			if (unlikely(error)) {
-				if (error == ENOSPC)
-					PMD_TX_LOG(ERR, "virtqueue_enqueue Free count = 0");
-				else if (error == EMSGSIZE)
-					PMD_TX_LOG(ERR, "virtqueue_enqueue Free count < 1");
-				else
-					PMD_TX_LOG(ERR, "virtqueue_enqueue error: %d", error);
-				break;
+				rte_pktmbuf_free(txm);
+				continue;
 			}
-			nb_tx++;
-			txvq->bytes += txm->pkt_len;
-		} else {
-			PMD_TX_LOG(ERR, "No free tx descriptors to transmit");
+		}
+
+		/* Enqueue Packet buffers */
+		error = virtqueue_enqueue_xmit(txvq, txm);
+		if (unlikely(error)) {
+			if (error == ENOSPC)
+				PMD_TX_LOG(ERR, "virtqueue_enqueue Free count = 0");
+			else if (error == EMSGSIZE)
+				PMD_TX_LOG(ERR, "virtqueue_enqueue Free count < 1");
+			else
+				PMD_TX_LOG(ERR, "virtqueue_enqueue error: %d", error);
 			break;
 		}
+		txvq->bytes += txm->pkt_len;
 	}
 
 	txvq->packets += nb_tx;
-- 
2.1.4

[dpdk-dev] [PATCH 2/4] virtio: don't use unlikely for normal tx stuff

[Stephen Hemminger]

Don't use unlikely() for VLAN or ring getting full.
GCC will not optimize code in unlikely paths and since these can
happen with normal code that can hurt performance.

Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/net/virtio/virtio_rxtx.c | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/drivers/net/virtio/virtio_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
index 5b50ed0..dbe6665 100644
--- a/drivers/net/virtio/virtio_rxtx.c
+++ b/drivers/net/virtio/virtio_rxtx.c
@@ -763,7 +763,7 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 		int need = txm->nb_segs - txvq->vq_free_cnt + 1;
 
 		/* Positive value indicates it need free vring descriptors */
-		if (unlikely(need > 0)) {
+		if (need > 0) {
 			nb_used = VIRTQUEUE_NUSED(txvq);
 			virtio_rmb();
 			need = RTE_MIN(need, (int)nb_used);
@@ -778,7 +778,7 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 		}
 
 		/* Do VLAN tag insertion */
-		if (unlikely(txm->ol_flags & PKT_TX_VLAN_PKT)) {
+		if (txm->ol_flags & PKT_TX_VLAN_PKT) {
 			error = rte_vlan_insert(&txm);
 			if (unlikely(error)) {
 				rte_pktmbuf_free(txm);
@@ -798,10 +798,9 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 			break;
 		}
 		txvq->bytes += txm->pkt_len;
+		++txvq->packets;
 	}
 
-	txvq->packets += nb_tx;
-
 	if (likely(nb_tx)) {
 		vq_update_avail_idx(txvq);
 
-- 
2.1.4

[dpdk-dev] [PATCH 3/4] virtio: use indirect ring elements

[Stephen Hemminger]

The virtio ring in QEMU/KVM is usually limited to 256 entries
and the normal way that virtio driver was queuing mbufs required
nsegs + 1 ring elements. By using the indirect ring element feature
if available, each packet will take only one ring slot even for
multi-segment packets.

Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/net/virtio/virtio_ethdev.c | 11 +++++---
 drivers/net/virtio/virtio_ethdev.h |  3 ++-
 drivers/net/virtio/virtio_rxtx.c   | 51 ++++++++++++++++++++++++++++++--------
 drivers/net/virtio/virtqueue.h     |  8 ++++++
 4 files changed, 57 insertions(+), 16 deletions(-)

diff --git a/drivers/net/virtio/virtio_ethdev.c b/drivers/net/virtio/virtio_ethdev.c
index 465d3cd..bcfb87b 100644
--- a/drivers/net/virtio/virtio_ethdev.c
+++ b/drivers/net/virtio/virtio_ethdev.c
@@ -359,12 +359,15 @@ int virtio_dev_queue_setup(struct rte_eth_dev *dev,
 	if (queue_type == VTNET_TQ) {
 		/*
 		 * For each xmit packet, allocate a virtio_net_hdr
+		 * and indirect ring elements
 		 */
 		snprintf(vq_name, sizeof(vq_name), "port%d_tvq%d_hdrzone",
-			dev->data->port_id, queue_idx);
-		vq->virtio_net_hdr_mz = rte_memzone_reserve_aligned(vq_name,
-			vq_size * hw->vtnet_hdr_size,
-			socket_id, 0, RTE_CACHE_LINE_SIZE);
+			 dev->data->port_id, queue_idx);
+
+		vq->virtio_net_hdr_mz =
+			rte_memzone_reserve_aligned(vq_name,
+						    vq_size * sizeof(struct virtio_tx_region),
+						    socket_id, 0, RTE_CACHE_LINE_SIZE);
 		if (vq->virtio_net_hdr_mz == NULL) {
 			if (rte_errno == EEXIST)
 				vq->virtio_net_hdr_mz =
diff --git a/drivers/net/virtio/virtio_ethdev.h b/drivers/net/virtio/virtio_ethdev.h
index 9026d42..07a9265 100644
--- a/drivers/net/virtio/virtio_ethdev.h
+++ b/drivers/net/virtio/virtio_ethdev.h
@@ -64,7 +64,8 @@
 	 1u << VIRTIO_NET_F_CTRL_VQ	  |	\
 	 1u << VIRTIO_NET_F_CTRL_RX	  |	\
 	 1u << VIRTIO_NET_F_CTRL_VLAN	  |	\
-	 1u << VIRTIO_NET_F_MRG_RXBUF)
+	 1u << VIRTIO_NET_F_MRG_RXBUF     |	\
+	 1u << VIRTIO_RING_F_INDIRECT_DESC)
 
 /*
  * CQ function prototype
diff --git a/drivers/net/virtio/virtio_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
index dbe6665..8979695 100644
--- a/drivers/net/virtio/virtio_rxtx.c
+++ b/drivers/net/virtio/virtio_rxtx.c
@@ -199,14 +199,15 @@ virtqueue_enqueue_recv_refill(struct virtqueue *vq, struct rte_mbuf *cookie)
 }
 
 static int
-virtqueue_enqueue_xmit(struct virtqueue *txvq, struct rte_mbuf *cookie)
+virtqueue_enqueue_xmit(struct virtqueue *txvq, struct rte_mbuf *cookie,
+		       int use_indirect)
 {
 	struct vq_desc_extra *dxp;
 	struct vring_desc *start_dp;
 	uint16_t seg_num = cookie->nb_segs;
-	uint16_t needed = 1 + seg_num;
+	uint16_t needed = use_indirect ? 1 : 1 + seg_num;
 	uint16_t head_idx, idx;
-	uint16_t head_size = txvq->hw->vtnet_hdr_size;
+	unsigned long offs;
 
 	if (unlikely(txvq->vq_free_cnt == 0))
 		return -ENOSPC;
@@ -220,11 +221,26 @@ virtqueue_enqueue_xmit(struct virtqueue *txvq, struct rte_mbuf *cookie)
 	dxp = &txvq->vq_descx[idx];
 	dxp->cookie = (void *)cookie;
 	dxp->ndescs = needed;
-
 	start_dp = txvq->vq_ring.desc;
-	start_dp[idx].addr =
-		txvq->virtio_net_hdr_mem + idx * head_size;
-	start_dp[idx].len = (uint32_t)head_size;
+
+	if (use_indirect) {
+		offs = offsetof(struct virtio_tx_region, tx_indir)
+			+ idx * sizeof(struct virtio_tx_region);
+
+		start_dp[idx].addr = txvq->virtio_net_hdr_mem + offs;
+		start_dp[idx].len = sizeof(struct vring_desc);
+		start_dp[idx].flags = VRING_DESC_F_INDIRECT;
+
+		start_dp = (struct vring_desc *)
+			((char *)txvq->virtio_net_hdr_mz->addr + offs);
+		idx = 0;
+	}
+
+	offs = offsetof(struct virtio_tx_region, tx_hdr)
+		+ idx * sizeof(struct virtio_tx_region);
+
+	start_dp[idx].addr = txvq->virtio_net_hdr_mem + offs;
+	start_dp[idx].len = txvq->hw->vtnet_hdr_size;
 	start_dp[idx].flags = VRING_DESC_F_NEXT;
 
 	for (; ((seg_num > 0) && (cookie != NULL)); seg_num--) {
@@ -236,7 +252,12 @@ virtqueue_enqueue_xmit(struct virtqueue *txvq, struct rte_mbuf *cookie)
 	}
 
 	start_dp[idx].flags &= ~VRING_DESC_F_NEXT;
-	idx = start_dp[idx].next;
+
+	if (use_indirect)
+		idx = txvq->vq_ring.desc[head_idx].next;
+	else
+		idx = start_dp[idx].next;
+
 	txvq->vq_desc_head_idx = idx;
 	if (txvq->vq_desc_head_idx == VQ_RING_DESC_CHAIN_END)
 		txvq->vq_desc_tail_idx = idx;
@@ -760,7 +781,15 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 
 	for (nb_tx = 0; nb_tx < nb_pkts; nb_tx++) {
 		struct rte_mbuf *txm = tx_pkts[nb_tx];
-		int need = txm->nb_segs - txvq->vq_free_cnt + 1;
+		int use_indirect, slots, need;
+
+		use_indirect = vtpci_with_feature(txvq->hw,
+						  VIRTIO_RING_F_INDIRECT_DESC)
+			&& (txm->nb_segs < VIRTIO_MAX_TX_INDIRECT);
+
+		/* How many ring entries are needed to this Tx? */
+		slots = use_indirect ? 1 : 1 + txm->nb_segs;
+		need = slots - txvq->vq_free_cnt;
 
 		/* Positive value indicates it need free vring descriptors */
 		if (need > 0) {
@@ -769,7 +798,7 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 			need = RTE_MIN(need, (int)nb_used);
 
 			virtio_xmit_cleanup(txvq, need);
-			need = txm->nb_segs - txvq->vq_free_cnt + 1;
+			need = slots - txvq->vq_free_cnt;
 			if (unlikely(need > 0)) {
 				PMD_TX_LOG(ERR,
 					   "No free tx descriptors to transmit");
@@ -787,7 +816,7 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 		}
 
 		/* Enqueue Packet buffers */
-		error = virtqueue_enqueue_xmit(txvq, txm);
+		error = virtqueue_enqueue_xmit(txvq, txm, use_indirect);
 		if (unlikely(error)) {
 			if (error == ENOSPC)
 				PMD_TX_LOG(ERR, "virtqueue_enqueue Free count = 0");
diff --git a/drivers/net/virtio/virtqueue.h b/drivers/net/virtio/virtqueue.h
index 7789411..a9410b4 100644
--- a/drivers/net/virtio/virtqueue.h
+++ b/drivers/net/virtio/virtqueue.h
@@ -237,6 +237,14 @@ struct virtio_net_hdr_mrg_rxbuf {
 	uint16_t num_buffers; /**< Number of merged rx buffers */
 };
 
+/* Region reserved to allow for transmit header and indirect ring */
+#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)));
+};
+
 /**
  * Tell the backend not to interrupt us.
  */
-- 
2.1.4

Re: [dpdk-dev] [PATCH 3/4] virtio: use indirect ring elements

[Ouyang, Changchun]

> -----Original Message-----
> From: Stephen Hemminger [mailto:stephen@networkplumber.org]
> Sent: Saturday, September 5, 2015 4:58 AM
> To: Xie, Huawei; Ouyang, Changchun
> Cc: dev@dpdk.org; Stephen Hemminger
> Subject: [PATCH 3/4] virtio: use indirect ring elements
> 
> The virtio ring in QEMU/KVM is usually limited to 256 entries and the normal
> way that virtio driver was queuing mbufs required nsegs + 1 ring elements.
> By using the indirect ring element feature if available, each packet will take
> only one ring slot even for multi-segment packets.
> 
> Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> ---
>  drivers/net/virtio/virtio_ethdev.c | 11 +++++---
> drivers/net/virtio/virtio_ethdev.h |  3 ++-
>  drivers/net/virtio/virtio_rxtx.c   | 51 ++++++++++++++++++++++++++++++-
> -------
>  drivers/net/virtio/virtqueue.h     |  8 ++++++
>  4 files changed, 57 insertions(+), 16 deletions(-)
> 
> diff --git a/drivers/net/virtio/virtio_ethdev.c
> b/drivers/net/virtio/virtio_ethdev.c
> index 465d3cd..bcfb87b 100644
> --- a/drivers/net/virtio/virtio_ethdev.c
> +++ b/drivers/net/virtio/virtio_ethdev.c
> @@ -359,12 +359,15 @@ int virtio_dev_queue_setup(struct rte_eth_dev
> *dev,
>  	if (queue_type == VTNET_TQ) {

Do we also need implement indirect ring elements for RX path?

>  		/*
>  		 * For each xmit packet, allocate a virtio_net_hdr
> +		 * and indirect ring elements
>  		 */
>  		snprintf(vq_name, sizeof(vq_name),
> "port%d_tvq%d_hdrzone",
> -			dev->data->port_id, queue_idx);
> -		vq->virtio_net_hdr_mz =
> rte_memzone_reserve_aligned(vq_name,
> -			vq_size * hw->vtnet_hdr_size,
> -			socket_id, 0, RTE_CACHE_LINE_SIZE);
> +			 dev->data->port_id, queue_idx);
> +
> +		vq->virtio_net_hdr_mz =
> +			rte_memzone_reserve_aligned(vq_name,
> +						    vq_size * sizeof(struct
> virtio_tx_region),
> +						    socket_id, 0,
> RTE_CACHE_LINE_SIZE);
>  		if (vq->virtio_net_hdr_mz == NULL) {
>  			if (rte_errno == EEXIST)
>  				vq->virtio_net_hdr_mz =
> diff --git a/drivers/net/virtio/virtio_ethdev.h
> b/drivers/net/virtio/virtio_ethdev.h
> index 9026d42..07a9265 100644
> --- a/drivers/net/virtio/virtio_ethdev.h
> +++ b/drivers/net/virtio/virtio_ethdev.h
> @@ -64,7 +64,8 @@
>  	 1u << VIRTIO_NET_F_CTRL_VQ	  |	\
>  	 1u << VIRTIO_NET_F_CTRL_RX	  |	\
>  	 1u << VIRTIO_NET_F_CTRL_VLAN	  |	\
> -	 1u << VIRTIO_NET_F_MRG_RXBUF)
> +	 1u << VIRTIO_NET_F_MRG_RXBUF     |	\
> +	 1u << VIRTIO_RING_F_INDIRECT_DESC)
> 
>  /*
>   * CQ function prototype
> diff --git a/drivers/net/virtio/virtio_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
> index dbe6665..8979695 100644
> --- a/drivers/net/virtio/virtio_rxtx.c
> +++ b/drivers/net/virtio/virtio_rxtx.c
> @@ -199,14 +199,15 @@ virtqueue_enqueue_recv_refill(struct virtqueue
> *vq, struct rte_mbuf *cookie)  }
> 
>  static int
> -virtqueue_enqueue_xmit(struct virtqueue *txvq, struct rte_mbuf *cookie)
> +virtqueue_enqueue_xmit(struct virtqueue *txvq, struct rte_mbuf *cookie,
> +		       int use_indirect)
>  {
>  	struct vq_desc_extra *dxp;
>  	struct vring_desc *start_dp;
>  	uint16_t seg_num = cookie->nb_segs;
> -	uint16_t needed = 1 + seg_num;
> +	uint16_t needed = use_indirect ? 1 : 1 + seg_num;

Do we need check if seg_num > VIRTIO_MAX_TX_INDIRECT?
That mean one slot is not enough for the whole big packet even it is indirect ring.

>  	uint16_t head_idx, idx;
> -	uint16_t head_size = txvq->hw->vtnet_hdr_size;
> +	unsigned long offs;
> 
>  	if (unlikely(txvq->vq_free_cnt == 0))
>  		return -ENOSPC;
> @@ -220,11 +221,26 @@ virtqueue_enqueue_xmit(struct virtqueue *txvq,
> struct rte_mbuf *cookie)
>  	dxp = &txvq->vq_descx[idx];
>  	dxp->cookie = (void *)cookie;
>  	dxp->ndescs = needed;
> -
>  	start_dp = txvq->vq_ring.desc;
> -	start_dp[idx].addr =
> -		txvq->virtio_net_hdr_mem + idx * head_size;
> -	start_dp[idx].len = (uint32_t)head_size;
> +
> +	if (use_indirect) {
> +		offs = offsetof(struct virtio_tx_region, tx_indir)
> +			+ idx * sizeof(struct virtio_tx_region);
> +
> +		start_dp[idx].addr = txvq->virtio_net_hdr_mem + offs;
> +		start_dp[idx].len = sizeof(struct vring_desc);
> +		start_dp[idx].flags = VRING_DESC_F_INDIRECT;
> +
> +		start_dp = (struct vring_desc *)
> +			((char *)txvq->virtio_net_hdr_mz->addr + offs);
> +		idx = 0;
> +	}
> +
> +	offs = offsetof(struct virtio_tx_region, tx_hdr)
> +		+ idx * sizeof(struct virtio_tx_region);
> +
> +	start_dp[idx].addr = txvq->virtio_net_hdr_mem + offs;
> +	start_dp[idx].len = txvq->hw->vtnet_hdr_size;
>  	start_dp[idx].flags = VRING_DESC_F_NEXT;
> 
>  	for (; ((seg_num > 0) && (cookie != NULL)); seg_num--) { @@ -236,7
> +252,12 @@ virtqueue_enqueue_xmit(struct virtqueue *txvq, struct
> rte_mbuf *cookie)
>  	}
> 
>  	start_dp[idx].flags &= ~VRING_DESC_F_NEXT;
> -	idx = start_dp[idx].next;
> +
> +	if (use_indirect)
> +		idx = txvq->vq_ring.desc[head_idx].next;
> +	else
> +		idx = start_dp[idx].next;
> +
>  	txvq->vq_desc_head_idx = idx;
>  	if (txvq->vq_desc_head_idx == VQ_RING_DESC_CHAIN_END)
>  		txvq->vq_desc_tail_idx = idx;
> @@ -760,7 +781,15 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf
> **tx_pkts, uint16_t nb_pkts)
> 
>  	for (nb_tx = 0; nb_tx < nb_pkts; nb_tx++) {
>  		struct rte_mbuf *txm = tx_pkts[nb_tx];
> -		int need = txm->nb_segs - txvq->vq_free_cnt + 1;
> +		int use_indirect, slots, need;
> +
> +		use_indirect = vtpci_with_feature(txvq->hw,
> +
> VIRTIO_RING_F_INDIRECT_DESC)
> +			&& (txm->nb_segs < VIRTIO_MAX_TX_INDIRECT);
> +
> +		/* How many ring entries are needed to this Tx? */
> +		slots = use_indirect ? 1 : 1 + txm->nb_segs;
> +		need = slots - txvq->vq_free_cnt;
> 
>  		/* Positive value indicates it need free vring descriptors */
>  		if (need > 0) {
> @@ -769,7 +798,7 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf
> **tx_pkts, uint16_t nb_pkts)
>  			need = RTE_MIN(need, (int)nb_used);
> 
>  			virtio_xmit_cleanup(txvq, need);
> -			need = txm->nb_segs - txvq->vq_free_cnt + 1;
> +			need = slots - txvq->vq_free_cnt;
>  			if (unlikely(need > 0)) {
>  				PMD_TX_LOG(ERR,
>  					   "No free tx descriptors to transmit");
> @@ -787,7 +816,7 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf
> **tx_pkts, uint16_t nb_pkts)
>  		}
> 
>  		/* Enqueue Packet buffers */
> -		error = virtqueue_enqueue_xmit(txvq, txm);
> +		error = virtqueue_enqueue_xmit(txvq, txm, use_indirect);
>  		if (unlikely(error)) {
>  			if (error == ENOSPC)
>  				PMD_TX_LOG(ERR, "virtqueue_enqueue
> Free count = 0"); diff --git a/drivers/net/virtio/virtqueue.h
> b/drivers/net/virtio/virtqueue.h index 7789411..a9410b4 100644
> --- a/drivers/net/virtio/virtqueue.h
> +++ b/drivers/net/virtio/virtqueue.h
> @@ -237,6 +237,14 @@ struct virtio_net_hdr_mrg_rxbuf {
>  	uint16_t num_buffers; /**< Number of merged rx buffers */  };
> 
> +/* Region reserved to allow for transmit header and indirect ring */
> +#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))); };
> +
>  /**
>   * Tell the backend not to interrupt us.
>   */
> --
> 2.1.4

Re: [dpdk-dev] [PATCH 3/4] virtio: use indirect ring elements

[Stephen Hemminger]

On Sun, 6 Sep 2015 08:36:10 +0000
"Ouyang, Changchun" <changchun.ouyang@intel.com> wrote:

> > -----Original Message-----
> > From: Stephen Hemminger [mailto:stephen@networkplumber.org]
> > Sent: Saturday, September 5, 2015 4:58 AM
> > To: Xie, Huawei; Ouyang, Changchun
> > Cc: dev@dpdk.org; Stephen Hemminger
> > Subject: [PATCH 3/4] virtio: use indirect ring elements
> > 
> > The virtio ring in QEMU/KVM is usually limited to 256 entries and the normal
> > way that virtio driver was queuing mbufs required nsegs + 1 ring elements.
> > By using the indirect ring element feature if available, each packet will take
> > only one ring slot even for multi-segment packets.
> > 
> > Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> > ---
> >  drivers/net/virtio/virtio_ethdev.c | 11 +++++---
> > drivers/net/virtio/virtio_ethdev.h |  3 ++-
> >  drivers/net/virtio/virtio_rxtx.c   | 51 ++++++++++++++++++++++++++++++-
> > -------
> >  drivers/net/virtio/virtqueue.h     |  8 ++++++
> >  4 files changed, 57 insertions(+), 16 deletions(-)
> > 
> > diff --git a/drivers/net/virtio/virtio_ethdev.c
> > b/drivers/net/virtio/virtio_ethdev.c
> > index 465d3cd..bcfb87b 100644
> > --- a/drivers/net/virtio/virtio_ethdev.c
> > +++ b/drivers/net/virtio/virtio_ethdev.c
> > @@ -359,12 +359,15 @@ int virtio_dev_queue_setup(struct rte_eth_dev
> > *dev,
> >  	if (queue_type == VTNET_TQ) {  
> 
> Do we also need implement indirect ring elements for RX path?

No. Look at Linux driver, indirect elements are never passed to RX driver.

Re: [dpdk-dev] [PATCH 3/4] virtio: use indirect ring elements

[Ouyang, Changchun]

> -----Original Message-----
> From: Stephen Hemminger [mailto:stephen@networkplumber.org]
> Sent: Saturday, September 5, 2015 4:58 AM
> To: Xie, Huawei; Ouyang, Changchun
> Cc: dev@dpdk.org; Stephen Hemminger
> Subject: [PATCH 3/4] virtio: use indirect ring elements
> 
> The virtio ring in QEMU/KVM is usually limited to 256 entries and the normal
> way that virtio driver was queuing mbufs required nsegs + 1 ring elements.
> By using the indirect ring element feature if available, each packet will take
> only one ring slot even for multi-segment packets.
> 
> Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> ---
>  drivers/net/virtio/virtio_ethdev.c | 11 +++++---
> drivers/net/virtio/virtio_ethdev.h |  3 ++-
>  drivers/net/virtio/virtio_rxtx.c   | 51 ++++++++++++++++++++++++++++++-
> -------
>  drivers/net/virtio/virtqueue.h     |  8 ++++++
>  4 files changed, 57 insertions(+), 16 deletions(-)
> 
> diff --git a/drivers/net/virtio/virtio_ethdev.c
> b/drivers/net/virtio/virtio_ethdev.c
> index 465d3cd..bcfb87b 100644
> --- a/drivers/net/virtio/virtio_ethdev.c
> +++ b/drivers/net/virtio/virtio_ethdev.c
> @@ -359,12 +359,15 @@ int virtio_dev_queue_setup(struct rte_eth_dev
> *dev,
>  	if (queue_type == VTNET_TQ) {
>  		/*
>  		 * For each xmit packet, allocate a virtio_net_hdr
> +		 * and indirect ring elements
>  		 */
>  		snprintf(vq_name, sizeof(vq_name),
> "port%d_tvq%d_hdrzone",
> -			dev->data->port_id, queue_idx);
> -		vq->virtio_net_hdr_mz =
> rte_memzone_reserve_aligned(vq_name,
> -			vq_size * hw->vtnet_hdr_size,
> -			socket_id, 0, RTE_CACHE_LINE_SIZE);
> +			 dev->data->port_id, queue_idx);
> +
> +		vq->virtio_net_hdr_mz =
> +			rte_memzone_reserve_aligned(vq_name,
> +						    vq_size * sizeof(struct
> virtio_tx_region),
> +						    socket_id, 0,
> RTE_CACHE_LINE_SIZE);
>  		if (vq->virtio_net_hdr_mz == NULL) {
>  			if (rte_errno == EEXIST)
>  				vq->virtio_net_hdr_mz =
> diff --git a/drivers/net/virtio/virtio_ethdev.h
> b/drivers/net/virtio/virtio_ethdev.h
> index 9026d42..07a9265 100644
> --- a/drivers/net/virtio/virtio_ethdev.h
> +++ b/drivers/net/virtio/virtio_ethdev.h
> @@ -64,7 +64,8 @@
>  	 1u << VIRTIO_NET_F_CTRL_VQ	  |	\
>  	 1u << VIRTIO_NET_F_CTRL_RX	  |	\
>  	 1u << VIRTIO_NET_F_CTRL_VLAN	  |	\
> -	 1u << VIRTIO_NET_F_MRG_RXBUF)
> +	 1u << VIRTIO_NET_F_MRG_RXBUF     |	\
> +	 1u << VIRTIO_RING_F_INDIRECT_DESC)
> 
>  /*
>   * CQ function prototype
> diff --git a/drivers/net/virtio/virtio_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
> index dbe6665..8979695 100644
> --- a/drivers/net/virtio/virtio_rxtx.c
> +++ b/drivers/net/virtio/virtio_rxtx.c
> @@ -199,14 +199,15 @@ virtqueue_enqueue_recv_refill(struct virtqueue
> *vq, struct rte_mbuf *cookie)  }
> 
>  static int
> -virtqueue_enqueue_xmit(struct virtqueue *txvq, struct rte_mbuf *cookie)
> +virtqueue_enqueue_xmit(struct virtqueue *txvq, struct rte_mbuf *cookie,
> +		       int use_indirect)
>  {
>  	struct vq_desc_extra *dxp;
>  	struct vring_desc *start_dp;
>  	uint16_t seg_num = cookie->nb_segs;
> -	uint16_t needed = 1 + seg_num;
> +	uint16_t needed = use_indirect ? 1 : 1 + seg_num;
>  	uint16_t head_idx, idx;
> -	uint16_t head_size = txvq->hw->vtnet_hdr_size;
> +	unsigned long offs;
> 
>  	if (unlikely(txvq->vq_free_cnt == 0))
>  		return -ENOSPC;
> @@ -220,11 +221,26 @@ virtqueue_enqueue_xmit(struct virtqueue *txvq,
> struct rte_mbuf *cookie)
>  	dxp = &txvq->vq_descx[idx];
>  	dxp->cookie = (void *)cookie;
>  	dxp->ndescs = needed;
> -
>  	start_dp = txvq->vq_ring.desc;
> -	start_dp[idx].addr =
> -		txvq->virtio_net_hdr_mem + idx * head_size;
> -	start_dp[idx].len = (uint32_t)head_size;
> +
> +	if (use_indirect) {
> +		offs = offsetof(struct virtio_tx_region, tx_indir)
> +			+ idx * sizeof(struct virtio_tx_region);
> +
> +		start_dp[idx].addr = txvq->virtio_net_hdr_mem + offs;
> +		start_dp[idx].len = sizeof(struct vring_desc);

Should the length be N * sizeof(struct vring_desc)?

> +		start_dp[idx].flags = VRING_DESC_F_INDIRECT;
> +
> +		start_dp = (struct vring_desc *)
> +			((char *)txvq->virtio_net_hdr_mz->addr + offs);
> +		idx = 0;
> +	}
> +
> +	offs = offsetof(struct virtio_tx_region, tx_hdr)
> +		+ idx * sizeof(struct virtio_tx_region);
> +
> +	start_dp[idx].addr = txvq->virtio_net_hdr_mem + offs;
> +	start_dp[idx].len = txvq->hw->vtnet_hdr_size;
>  	start_dp[idx].flags = VRING_DESC_F_NEXT;
> 
>  	for (; ((seg_num > 0) && (cookie != NULL)); seg_num--) { @@ -236,7
> +252,12 @@ virtqueue_enqueue_xmit(struct virtqueue *txvq, struct
> rte_mbuf *cookie)
>  	}
> 
>  	start_dp[idx].flags &= ~VRING_DESC_F_NEXT;
> -	idx = start_dp[idx].next;
> +
> +	if (use_indirect)
> +		idx = txvq->vq_ring.desc[head_idx].next;
> +	else
> +		idx = start_dp[idx].next;
> +
>  	txvq->vq_desc_head_idx = idx;
>  	if (txvq->vq_desc_head_idx == VQ_RING_DESC_CHAIN_END)
>  		txvq->vq_desc_tail_idx = idx;
> @@ -760,7 +781,15 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf
> **tx_pkts, uint16_t nb_pkts)
> 
>  	for (nb_tx = 0; nb_tx < nb_pkts; nb_tx++) {
>  		struct rte_mbuf *txm = tx_pkts[nb_tx];
> -		int need = txm->nb_segs - txvq->vq_free_cnt + 1;
> +		int use_indirect, slots, need;
> +
> +		use_indirect = vtpci_with_feature(txvq->hw,
> +
> VIRTIO_RING_F_INDIRECT_DESC)
> +			&& (txm->nb_segs < VIRTIO_MAX_TX_INDIRECT);
> +
> +		/* How many ring entries are needed to this Tx? */
> +		slots = use_indirect ? 1 : 1 + txm->nb_segs;
> +		need = slots - txvq->vq_free_cnt;
> 
>  		/* Positive value indicates it need free vring descriptors */
>  		if (need > 0) {
> @@ -769,7 +798,7 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf
> **tx_pkts, uint16_t nb_pkts)
>  			need = RTE_MIN(need, (int)nb_used);
> 
>  			virtio_xmit_cleanup(txvq, need);
> -			need = txm->nb_segs - txvq->vq_free_cnt + 1;
> +			need = slots - txvq->vq_free_cnt;
>  			if (unlikely(need > 0)) {
>  				PMD_TX_LOG(ERR,
>  					   "No free tx descriptors to transmit");
> @@ -787,7 +816,7 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf
> **tx_pkts, uint16_t nb_pkts)
>  		}
> 
>  		/* Enqueue Packet buffers */
> -		error = virtqueue_enqueue_xmit(txvq, txm);
> +		error = virtqueue_enqueue_xmit(txvq, txm, use_indirect);
>  		if (unlikely(error)) {
>  			if (error == ENOSPC)
>  				PMD_TX_LOG(ERR, "virtqueue_enqueue
> Free count = 0"); diff --git a/drivers/net/virtio/virtqueue.h
> b/drivers/net/virtio/virtqueue.h index 7789411..a9410b4 100644
> --- a/drivers/net/virtio/virtqueue.h
> +++ b/drivers/net/virtio/virtqueue.h
> @@ -237,6 +237,14 @@ struct virtio_net_hdr_mrg_rxbuf {
>  	uint16_t num_buffers; /**< Number of merged rx buffers */  };
> 
> +/* Region reserved to allow for transmit header and indirect ring */
> +#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))); };
> +
>  /**
>   * Tell the backend not to interrupt us.
>   */
> --
> 2.1.4

Re: [dpdk-dev] [PATCH 3/4] virtio: use indirect ring elements

[Stephen Hemminger]

On Sun, 6 Sep 2015 08:40:44 +0000
"Ouyang, Changchun" <changchun.ouyang@intel.com> wrote:

> > @@ -220,11 +221,26 @@ virtqueue_enqueue_xmit(struct virtqueue *txvq,
> > struct rte_mbuf *cookie)
> >  	dxp = &txvq->vq_descx[idx];
> >  	dxp->cookie = (void *)cookie;
> >  	dxp->ndescs = needed;
> > -
> >  	start_dp = txvq->vq_ring.desc;
> > -	start_dp[idx].addr =
> > -		txvq->virtio_net_hdr_mem + idx * head_size;
> > -	start_dp[idx].len = (uint32_t)head_size;
> > +
> > +	if (use_indirect) {
> > +		offs = offsetof(struct virtio_tx_region, tx_indir)
> > +			+ idx * sizeof(struct virtio_tx_region);
> > +
> > +		start_dp[idx].addr = txvq->virtio_net_hdr_mem + offs;
> > +		start_dp[idx].len = sizeof(struct vring_desc);  
> 
> Should the length be N * sizeof(struct vring_desc)?

Yes.

Re: [dpdk-dev] [PATCH 3/4] virtio: use indirect ring elements

[Thomas Monjalon]

General comment after reading this page:
	http://dpdk.org/dev/patchwork/patch/6905/
Please remove useless context when replying to make answers shorter
and easier to read.
Thanks

[dpdk-dev] [PATCH 4/4] virtio: use any layout on transmit

[Stephen Hemminger]

Virtio supports a feature that allows sender to put transmit
header prepended to data.  It requires that the mbuf be writeable, correct
alignment, and the feature has been negotiatied.  If all this works out,
then it will be the optimum way to transmit a single segment packet.

Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 drivers/net/virtio/virtio_ethdev.h |  3 +-
 drivers/net/virtio/virtio_rxtx.c   | 67 ++++++++++++++++++++++++++------------
 2 files changed, 49 insertions(+), 21 deletions(-)

diff --git a/drivers/net/virtio/virtio_ethdev.h b/drivers/net/virtio/virtio_ethdev.h
index 07a9265..f260fbb 100644
--- a/drivers/net/virtio/virtio_ethdev.h
+++ b/drivers/net/virtio/virtio_ethdev.h
@@ -65,7 +65,8 @@
 	 1u << VIRTIO_NET_F_CTRL_RX	  |	\
 	 1u << VIRTIO_NET_F_CTRL_VLAN	  |	\
 	 1u << VIRTIO_NET_F_MRG_RXBUF     |	\
-	 1u << VIRTIO_RING_F_INDIRECT_DESC)
+	 1u << VIRTIO_RING_F_INDIRECT_DESC|	\
+	 1u << VIRTIO_F_ANY_LAYOUT)
 
 /*
  * CQ function prototype
diff --git a/drivers/net/virtio/virtio_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
index 8979695..5ec9b29 100644
--- a/drivers/net/virtio/virtio_rxtx.c
+++ b/drivers/net/virtio/virtio_rxtx.c
@@ -200,13 +200,14 @@ virtqueue_enqueue_recv_refill(struct virtqueue *vq, struct rte_mbuf *cookie)
 
 static int
 virtqueue_enqueue_xmit(struct virtqueue *txvq, struct rte_mbuf *cookie,
-		       int use_indirect)
+		       int use_indirect, int can_push)
 {
 	struct vq_desc_extra *dxp;
 	struct vring_desc *start_dp;
 	uint16_t seg_num = cookie->nb_segs;
-	uint16_t needed = use_indirect ? 1 : 1 + seg_num;
+	uint16_t needed = use_indirect ? 1 : !can_push + seg_num;
 	uint16_t head_idx, idx;
+	uint16_t head_size = txvq->hw->vtnet_hdr_size;
 	unsigned long offs;
 
 	if (unlikely(txvq->vq_free_cnt == 0))
@@ -236,27 +237,31 @@ virtqueue_enqueue_xmit(struct virtqueue *txvq, struct rte_mbuf *cookie,
 		idx = 0;
 	}
 
-	offs = offsetof(struct virtio_tx_region, tx_hdr)
-		+ idx * sizeof(struct virtio_tx_region);
+	if (can_push) {
+		/* put on zero'd transmit header (no offloads) */
+		void *hdr = rte_pktmbuf_prepend(cookie, head_size);
 
-	start_dp[idx].addr = txvq->virtio_net_hdr_mem + offs;
-	start_dp[idx].len = txvq->hw->vtnet_hdr_size;
-	start_dp[idx].flags = VRING_DESC_F_NEXT;
+		memset(hdr, 0, head_size);
+	} else {
+		offs = offsetof(struct virtio_tx_region, tx_hdr)
+			+ idx * sizeof(struct virtio_tx_region);
 
-	for (; ((seg_num > 0) && (cookie != NULL)); seg_num--) {
+		start_dp[idx].addr = txvq->virtio_net_hdr_mem + offs;
+		start_dp[idx].len = head_size;
+		start_dp[idx].flags = VRING_DESC_F_NEXT;
 		idx = start_dp[idx].next;
+	}
+
+	for (; ((seg_num > 0) && (cookie != NULL)); seg_num--) {
 		start_dp[idx].addr  = RTE_MBUF_DATA_DMA_ADDR(cookie);
 		start_dp[idx].len   = cookie->data_len;
-		start_dp[idx].flags = VRING_DESC_F_NEXT;
 		cookie = cookie->next;
+		start_dp[idx].flags = cookie ? VRING_DESC_F_NEXT : 0;
+		idx = start_dp[idx].next;
 	}
 
-	start_dp[idx].flags &= ~VRING_DESC_F_NEXT;
-
 	if (use_indirect)
 		idx = txvq->vq_ring.desc[head_idx].next;
-	else
-		idx = start_dp[idx].next;
 
 	txvq->vq_desc_head_idx = idx;
 	if (txvq->vq_desc_head_idx == VQ_RING_DESC_CHAIN_END)
@@ -762,6 +767,26 @@ virtio_recv_mergeable_pkts(void *rx_queue,
 	return nb_rx;
 }
 
+/* Evaluate whether the virtio header can just be put in place in the mbuf */
+static int virtio_xmit_push_ok(const struct virtqueue *txvq,
+			       const struct rte_mbuf *m)
+{
+	if (rte_mbuf_refcnt_read(m) != 1)
+		return 0;	/* no mbuf is shared */
+
+	if (rte_pktmbuf_headroom(m) < txvq->hw->vtnet_hdr_size)
+		return 0;	/* no space in headroom */
+
+	if (!rte_is_aligned(rte_pktmbuf_mtod(m, char *),
+			    sizeof(struct virtio_net_hdr_mrg_rxbuf)))
+		return 0;	/* not alligned */
+
+	if (m->nb_segs > 1)
+		return 0;	/* better off using indirect */
+
+	return vtpci_with_feature(txvq->hw, VIRTIO_F_ANY_LAYOUT);
+}
+
 uint16_t
 virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 {
@@ -781,14 +806,16 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 
 	for (nb_tx = 0; nb_tx < nb_pkts; nb_tx++) {
 		struct rte_mbuf *txm = tx_pkts[nb_tx];
-		int use_indirect, slots, need;
-
-		use_indirect = vtpci_with_feature(txvq->hw,
-						  VIRTIO_RING_F_INDIRECT_DESC)
-			&& (txm->nb_segs < VIRTIO_MAX_TX_INDIRECT);
+		int use_indirect = 0, slots, need;
+		int can_push = virtio_xmit_push_ok(txvq, txm);
+		
+		if (!can_push &&
+		    txm->nb_segs < VIRTIO_MAX_TX_INDIRECT &&
+		    vtpci_with_feature(txvq->hw, VIRTIO_RING_F_INDIRECT_DESC))
+			use_indirect = 1;
 
 		/* How many ring entries are needed to this Tx? */
-		slots = use_indirect ? 1 : 1 + txm->nb_segs;
+		slots = use_indirect ? 1 : !can_push + txm->nb_segs;
 		need = slots - txvq->vq_free_cnt;
 
 		/* Positive value indicates it need free vring descriptors */
@@ -816,7 +843,7 @@ virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 		}
 
 		/* Enqueue Packet buffers */
-		error = virtqueue_enqueue_xmit(txvq, txm, use_indirect);
+		error = virtqueue_enqueue_xmit(txvq, txm, use_indirect, can_push);
 		if (unlikely(error)) {
 			if (error == ENOSPC)
 				PMD_TX_LOG(ERR, "virtqueue_enqueue Free count = 0");
-- 
2.1.4

Re: [dpdk-dev] [PATCH 0/4] RFC virtio performance enhancement and cleanups

[Thomas Monjalon]

Hi Stephen,

2015-09-04 13:58, Stephen Hemminger:
> These are compile tested only, haven't debugged or checked out the corner
> case. Submitted for discussion and future planning.
> 
> Stephen Hemminger (4):
>   virtio: clean up space checks on xmit
>   virtio: don't use unlikely for normal tx stuff
>   virtio: use indirect ring elements
>   virtio: use any layout on transmit

Thanks for the nice optimizations.
Do you have some performance numbers to share?