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

[Jens Freimann <jfreimann@redhat.com>]

On Thu, Oct 25, 2018 at 05:31:04PM +0800, Tiwei Bie wrote:
>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)

As maxime noted a made a mistake when squashing in fixes to this
patch. I sent a new version and updated my branch on github. 
>
>> +{
>> +	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.

Also fixed in the new version of this patch. 

I'll also fix your other findings. Thanks for the review!


regards,
Jens 

>> +			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
>>