Re: [PATCH v9 5/8] net/virtio: implement transmit path for packed queues

[Tiwei Bie <tiwei.bie@intel.com>]

On Wed, Oct 24, 2018 at 04:32:33PM +0200, Jens Freimann wrote:
> This implements the transmit path for devices with
> support for packed virtqueues.
> 
> Signed-off-by: Jens Freiman <jfreimann@redhat.com>
> ---
>  drivers/net/virtio/virtio_ethdev.c |  32 +++-
>  drivers/net/virtio/virtio_ethdev.h |   2 +
>  drivers/net/virtio/virtio_rxtx.c   | 284 +++++++++++++++++++++++++++++
>  drivers/net/virtio/virtqueue.h     |  18 +-
>  4 files changed, 324 insertions(+), 12 deletions(-)
> 
> diff --git a/drivers/net/virtio/virtio_ethdev.c b/drivers/net/virtio/virtio_ethdev.c
> index 6130aef16..d2118e6a9 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
> @@ -490,16 +492,22 @@ virtio_init_queue(struct rte_eth_dev *dev, uint16_t vtpci_queue_idx)
>  		memset(txr, 0, vq_size * sizeof(*txr));
>  		for (i = 0; i < vq_size; i++) {
>  			struct vring_desc *start_dp = txr[i].tx_indir;
> -
> -			vring_desc_init_split(start_dp, RTE_DIM(txr[i].tx_indir));
> -
> +			struct vring_desc_packed*start_dp_packed = txr[i].tx_indir_pq;

A space is missing before *start_dp_packed

> +	
>  			/* first indirect descriptor is always the tx header */
> -			start_dp->addr = txvq->virtio_net_hdr_mem
> -				+ i * sizeof(*txr)
> -				+ offsetof(struct virtio_tx_region, tx_hdr);
> -
> -			start_dp->len = hw->vtnet_hdr_size;
> -			start_dp->flags = VRING_DESC_F_NEXT;
> +			if (vtpci_packed_queue(hw)) {
> +				start_dp_packed->addr = txvq->virtio_net_hdr_mem
> +					+ i * sizeof(*txr)
> +					+ offsetof(struct virtio_tx_region, tx_hdr);
> +				start_dp_packed->len = hw->vtnet_hdr_size;
> +			} else {
> +				vring_desc_init_split(start_dp, RTE_DIM(txr[i].tx_indir));
> +				start_dp->addr = txvq->virtio_net_hdr_mem
> +					+ i * sizeof(*txr)
> +					+ offsetof(struct virtio_tx_region, tx_hdr);
> +				start_dp->len = hw->vtnet_hdr_size;
> +				start_dp->flags = VRING_DESC_F_NEXT;
> +			}
>  		}
>  	}
>  
> @@ -1338,7 +1346,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) {

Please make the structure more clear, e.g. something like:

	if (vtpci_packed_queue(hw)) {
		// packed ring
	} else {
		// split ring
	}


>  		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 e0f80e5a4..05d355180 100644
> --- a/drivers/net/virtio/virtio_ethdev.h
> +++ b/drivers/net/virtio/virtio_ethdev.h
> @@ -82,6 +82,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_rxtx.c b/drivers/net/virtio/virtio_rxtx.c
> index eb891433e..837457243 100644
> --- a/drivers/net/virtio/virtio_rxtx.c
> +++ b/drivers/net/virtio/virtio_rxtx.c
> @@ -88,6 +88,42 @@ vq_ring_free_chain(struct virtqueue *vq, uint16_t desc_idx)
>  	dp->next = VQ_RING_DESC_CHAIN_END;
>  }
>  
> +void
> +vq_ring_free_chain_packed(struct virtqueue *vq, uint16_t desc_idx, used_idx)

Does above code pass the build test? (Type is missing for used_idx)

> +{
> +	struct vring_desc_packed *dp;
> +	struct vq_desc_extra *dxp = NULL, *dxp_tail = NULL;

Do we really need to initialize them?

> +	uint16_t desc_idx_last = desc_idx;
> +	int i = 0;
> +
> +	dp  = &vq->vq_ring.desc_packed[used_idx];
> +	dxp = &vq->vq_descx[desc_idx];
> +	vq->vq_free_cnt = (uint16_t)(vq->vq_free_cnt + dxp->ndescs);
> +	if ((dp->flags & VRING_DESC_F_INDIRECT) == 0) {
> +		while (dp->flags & VRING_DESC_F_NEXT && i < dxp->ndescs) {

NEXT can't be used here.

> +			desc_idx_last = dxp->next;
> +			dp = &vq->vq_ring.desc_packed[dxp->next];

dxp->next is used to link the dxps, it's not supposed to
be desc ring's index.

> +			dxp = &vq->vq_descx[dxp->next];
> +			i++;
> +		}
> +	}
> +
> +	/*
> +	 * We must append the existing free chain, if any, to the end of
> +	 * newly freed chain. If the virtqueue was completely used, then
> +	 * head would be VQ_RING_DESC_CHAIN_END (ASSERTed above).
> +	 */
> +	if (vq->vq_desc_tail_idx == VQ_RING_DESC_CHAIN_END) {
> +		vq->vq_desc_head_idx = desc_idx;
> +	} else {
> +		dxp_tail = &vq->vq_descx[vq->vq_desc_tail_idx];
> +		dxp_tail->next = desc_idx;
> +	}
> +
> +	vq->vq_desc_tail_idx = desc_idx_last;
> +	dxp->next = VQ_RING_DESC_CHAIN_END;
> +}
> +
>  static uint16_t
>  virtqueue_dequeue_burst_rx(struct virtqueue *vq, struct rte_mbuf **rx_pkts,
>  			   uint32_t *len, uint16_t num)
> @@ -165,6 +201,33 @@ virtqueue_dequeue_rx_inorder(struct virtqueue *vq,
>  #endif
>  
>  /* Cleanup from completed transmits. */
> +static void
> +virtio_xmit_cleanup_packed(struct virtqueue *vq, int num)
> +{
> +	uint16_t used_idx, id;
> +	uint16_t size = vq->vq_nentries;
> +	struct vring_desc_packed *desc = vq->vq_ring.desc_packed;
> +	struct vq_desc_extra *dxp;
> +
> +	used_idx = vq->vq_used_cons_idx;
> +	while (num && desc_is_used(&desc[used_idx], &vq->vq_ring)) {
> +		num --;
> +		used_idx = vq->vq_used_cons_idx;
> +		id = desc[used_idx].index;
> +		dxp = &vq->vq_descx[id];
> +		if (++vq->vq_used_cons_idx >= size) {

We need to skip dxp->ndescs descriptors here.

> +			vq->vq_used_cons_idx -= size;
> +			vq->vq_ring.used_wrap_counter ^= 1;
> +		}
> +		vq_ring_free_chain_packed(vq, id, used_idx);
> +		if (dxp->cookie != NULL) {
> +			rte_pktmbuf_free(dxp->cookie);
> +			dxp->cookie = NULL;
> +		}
> +		used_idx = vq->vq_used_cons_idx;
> +	}
> +}
> +
>  static void
>  virtio_xmit_cleanup(struct virtqueue *vq, uint16_t num)
>  {
> @@ -456,6 +519,134 @@ 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, *head_dxp;
> +	struct virtqueue *vq = txvq->vq;
> +	struct vring_desc_packed *start_dp, *head_dp;
> +	uint16_t seg_num = cookie->nb_segs;
> +	uint16_t idx, head_id, head_flags;
> +	uint16_t head_size = vq->hw->vtnet_hdr_size;
> +	struct virtio_net_hdr *hdr;
> +	uint16_t prev;
> +
> +	head_id = vq->vq_desc_head_idx;
> +	idx = head_id;
> +	prev = idx;
> +	start_dp = vq->vq_ring.desc_packed;
> +	dxp = &vq->vq_descx[idx];
> +	dxp->ndescs = needed;
> +
> +	head_dp = &vq->vq_ring.desc_packed[head_id];
> +	head_dxp = &vq->vq_descx[head_id];

I don't think desc and descx can always have the same index.

> +	head_dxp->cookie = (void *) cookie;
> +	head_flags = cookie->next ? VRING_DESC_F_NEXT: 0;
> +	head_flags |= VRING_DESC_F_AVAIL(vq->vq_ring.avail_wrap_counter) |
> +		     VRING_DESC_F_USED(!vq->vq_ring.avail_wrap_counter);
> +
[...]
>  
> +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;

We can't "& RING_EVENT_FLAGS_DESC" here.

> +	return (flags != RING_EVENT_FLAGS_DISABLE);
> +}
> +
>  static inline void
>  virtqueue_notify(struct virtqueue *vq)
>  {
> -- 
> 2.17.1
>