Message-ID: <ccaff5f6-2b57-ca77-886d-142cac22774c@gmail.com>
Date:   Tue, 1 Jan 2019 22:44:52 +0900
From:   Toshiaki Makita <toshiaki.makita1@...il.com>
To:     William Tu <u9012063@...il.com>
Cc:     bjorn.topel@...il.com, magnus.karlsson@...il.com, ast@...nel.org,
        daniel@...earbox.net, netdev@...r.kernel.org,
        makita.toshiaki@....ntt.co.jp, yihung.wei@...il.com,
        magnus.karlsson@...el.com
Subject: Re: [PATCH bpf-next RFCv3 2/6] veth: support AF_XDP TX copy-mode.

Hi, William. Nice work.
I have some feedback and questions.

On 18/12/27 (木) 5:27:49, William Tu wrote:
> The patch adds support for AF_XDP async xmit.  Users can use
> AF_XDP on both sides of the veth and get better performance, with
> the cost of ksoftirqd doing the xmit.  The veth_xsk_async_xmit
> simply kicks the napi function, veth_poll, to receive the packets
> that are on the umem transmit ring at the _peer_ side.
> 
> Tested using two namespaces, one runs xdpsock and the other runs
> xdp_rxq_info.  A simple script comparing the performance with/without
> AF_XDP shows improvement from 724Kpps to 1.1Mpps.
> 
>    ip netns add at_ns0
>    ip link add p0 type veth peer name p1
>    ip link set p0 netns at_ns0
>    ip link set dev p1 up
>    ip netns exec at_ns0 ip link set dev p0 up
> 
>    # receiver
>    ip netns exec at_ns0 xdp_rxq_info --dev p0 --action XDP_DROP
> 
>    # sender
>    xdpsock -i p1 -t -N -z
>    or
>    xdpsock -i p1 -t -S
> 
> Signed-off-by: William Tu <u9012063@...il.com>
> ---
>   drivers/net/veth.c | 200 ++++++++++++++++++++++++++++++++++++++++++++++++++++-
>   1 file changed, 199 insertions(+), 1 deletion(-)
> 
> diff --git a/drivers/net/veth.c b/drivers/net/veth.c
> index f412ea1cef18..10cf9ded59f1 100644
> --- a/drivers/net/veth.c
> +++ b/drivers/net/veth.c
> @@ -25,6 +25,10 @@
>   #include <linux/ptr_ring.h>
>   #include <linux/bpf_trace.h>
>   #include <linux/net_tstamp.h>
> +#include <net/xdp_sock.h>
> +#include <linux/mm.h>
> +#include <linux/slab.h>
> +#include <net/page_pool.h>
>   
>   #define DRV_NAME	"veth"
>   #define DRV_VERSION	"1.0"
> @@ -53,6 +57,8 @@ struct veth_rq {
>   	bool			rx_notify_masked;
>   	struct ptr_ring		xdp_ring;
>   	struct xdp_rxq_info	xdp_rxq;
> +	struct xdp_umem *xsk_umem;
> +	u16 qid;

veth_rq has 31 bytes hole after rx_notify_masked, so they should be 
placed there.

>   };
>   
>   struct veth_priv {
> @@ -737,11 +743,95 @@ static int veth_xdp_rcv(struct veth_rq *rq, int budget, unsigned int *xdp_xmit)
>   	return done;
>   }
>   
> +static int veth_xsk_poll(struct napi_struct *napi, int budget)
> +{
> +	struct veth_priv *priv, *peer_priv;
> +	struct net_device *dev, *peer_dev;
> +	struct veth_rq *peer_rq;
> +	struct veth_rq *rq =
> +		container_of(napi, struct veth_rq, xdp_napi);
> +	int done = 0;
> +
> +	dev = rq->dev;
> +	priv = netdev_priv(dev);
> +	peer_dev = priv->peer;
> +	peer_priv = netdev_priv(peer_dev);
> +	peer_rq = &peer_priv->rq[rq->qid];
> +
> +	while (peer_rq->xsk_umem && budget--) {
> +		unsigned int inner_xdp_xmit = 0;
> +		unsigned int metasize = 0;
> +		struct xdp_frame *xdpf;
> +		bool dropped = false;
> +		struct sk_buff *skb;
> +		struct page *page;
> +		void *vaddr;
> +		void *addr;
> +		u32 len;
> +
> +		if (!xsk_umem_consume_tx_virtual(peer_rq->xsk_umem, &vaddr, &len))
> +			break;

How do you prevent races around xsk_umem?
It seems you are checking xsk_umem in above while() condition, but there 
is no guarantee that xsk_umem is not NULL here, since umem can be 
disabled under us?

> +
> +		page = dev_alloc_page();
> +		if (!page) {
> +			xsk_umem_complete_tx(peer_rq->xsk_umem, 1);
> +			xsk_umem_consume_tx_done(peer_rq->xsk_umem);
> +			return -ENOMEM;
> +		}
> +
> +		addr = page_to_virt(page);
> +		xdpf = addr;
> +		memset(xdpf, 0, sizeof(*xdpf));
> +
> +		addr += sizeof(*xdpf);
> +		memcpy(addr, vaddr, len);
> +
> +		xdpf->data = addr + metasize;
> +		xdpf->len = len;
> +		xdpf->headroom = 0;
> +		xdpf->metasize = metasize;
> +		xdpf->mem.type = MEM_TYPE_PAGE_SHARED;
> +
> +		/* put into rq */
> +		skb = veth_xdp_rcv_one(rq, xdpf, &inner_xdp_xmit);
> +		if (!skb) {
> +			/* Peer side has XDP program attached */
> +			if (inner_xdp_xmit & VETH_XDP_TX) {
> +				/* Not supported */
> +				pr_warn("veth: peer XDP_TX not supported\n");

As this can be triggered by users we need ratelimit at least.

But since this is envisioned to be used in OVS, XDP_TX would be a very 
important feature to me. I expect XDP programs in containers to process 
packets and send back to OVS.

> +				xdp_return_frame(xdpf);
> +				dropped = true;
> +				goto skip_tx;
> +			} else if (inner_xdp_xmit & VETH_XDP_REDIR) {
> +				xdp_do_flush_map();
> +			} else {
> +				dropped = true;
> +			}
> +		} else {
> +			napi_gro_receive(&rq->xdp_napi, skb);
> +		}
> +skip_tx:
> +		xsk_umem_complete_tx(peer_rq->xsk_umem, 1);
> +		xsk_umem_consume_tx_done(peer_rq->xsk_umem);
> +
> +		/* update rq stats */
> +		u64_stats_update_begin(&rq->stats.syncp);
> +		rq->stats.xdp_packets++;
> +		rq->stats.xdp_bytes += len;
> +		if (dropped)
> +			rq->stats.xdp_drops++;
> +		u64_stats_update_end(&rq->stats.syncp);
> +		done++;
> +	}
> +	return done;
> +}
> +
>   static int veth_poll(struct napi_struct *napi, int budget)
>   {
>   	struct veth_rq *rq =
>   		container_of(napi, struct veth_rq, xdp_napi);
>   	unsigned int xdp_xmit = 0;
> +	int tx_done;
>   	int done;
>   
>   	xdp_set_return_frame_no_direct();
> @@ -756,13 +846,17 @@ static int veth_poll(struct napi_struct *napi, int budget)
>   		}
>   	}
>   
> +	tx_done = veth_xsk_poll(napi, budget);
> +	if (tx_done > 0)
> +		done += tx_done;
> +

This receives packets more than budget.

>   	if (xdp_xmit & VETH_XDP_TX)
>   		veth_xdp_flush(rq->dev);
>   	if (xdp_xmit & VETH_XDP_REDIR)
>   		xdp_do_flush_map();
>   	xdp_clear_return_frame_no_direct();
>   
> -	return done;
> +	return done > budget ? budget : done;
>   }
>   
>   static int veth_napi_add(struct net_device *dev)
> @@ -776,6 +870,7 @@ static int veth_napi_add(struct net_device *dev)
>   		err = ptr_ring_init(&rq->xdp_ring, VETH_RING_SIZE, GFP_KERNEL);
>   		if (err)
>   			goto err_xdp_ring;
> +		rq->qid = i;
>   	}
>   
>   	for (i = 0; i < dev->real_num_rx_queues; i++) {
> @@ -812,6 +907,7 @@ static void veth_napi_del(struct net_device *dev)
>   		netif_napi_del(&rq->xdp_napi);
>   		rq->rx_notify_masked = false;
>   		ptr_ring_cleanup(&rq->xdp_ring, veth_ptr_free);
> +		rq->qid = -1;
>   	}
>   }
>   
> @@ -836,6 +932,7 @@ static int veth_enable_xdp(struct net_device *dev)
>   
>   			/* Save original mem info as it can be overwritten */
>   			rq->xdp_mem = rq->xdp_rxq.mem;
> +			rq->qid = i;
>   		}
>   
>   		err = veth_napi_add(dev);
> @@ -1115,6 +1212,84 @@ static u32 veth_xdp_query(struct net_device *dev)
>   	return 0;
>   }
>   
> +int veth_xsk_umem_query(struct net_device *dev, struct xdp_umem **umem,
> +			u16 qid)
> +{
> +	struct xdp_umem *queried_umem;
> +
> +	queried_umem = xdp_get_umem_from_qid(dev, qid);
> +
> +	if (!queried_umem)
> +		return -EINVAL;
> +
> +	*umem = queried_umem;
> +	return 0;
> +}
> +
> +static int veth_xsk_umem_enable(struct net_device *dev,
> +				struct xdp_umem *umem,
> +				u16 qid)
> +{
> +	struct veth_priv *priv = netdev_priv(dev);
> +	struct xdp_umem_fq_reuse *reuseq;
> +	int err = 0;
> +
> +	if (qid >= dev->real_num_rx_queues)
> +		return -EINVAL;
> +
> +	reuseq = xsk_reuseq_prepare(priv->rq[0].xdp_ring.size);
> +	if (!reuseq)
> +		return -ENOMEM;
> +
> +	xsk_reuseq_free(xsk_reuseq_swap(umem, reuseq));
> +
> +	priv->rq[qid].xsk_umem = umem;
> +
> +	return err;
> +}
> +
> +static int veth_xsk_umem_disable(struct net_device *dev,
> +				 u16 qid)
> +{
> +	struct veth_priv *priv = netdev_priv(dev);
> +	struct xdp_umem *umem;
> +
> +	umem = xdp_get_umem_from_qid(dev, qid);
> +	if (!umem)
> +		return -EINVAL;
> +
> +	priv->rq[qid].xsk_umem = NULL;
> +	return 0;
> +}
> +
> +int veth_xsk_umem_setup(struct net_device *dev, struct xdp_umem *umem,
> +			u16 qid)
> +{
> +	return umem ? veth_xsk_umem_enable(dev, umem, qid) :
> +		      veth_xsk_umem_disable(dev, qid);
> +}
> +
> +int veth_xsk_async_xmit(struct net_device *dev, u32 qid)
> +{
> +	struct veth_priv *priv, *peer_priv;
> +	struct net_device *peer_dev;
> +	struct veth_rq *peer_rq;
> +
> +	priv = netdev_priv(dev);
> +	peer_dev = priv->peer;
> +	peer_priv = netdev_priv(peer_dev);
> +	peer_rq = &peer_priv->rq[qid];
> +
> +	if (qid >= dev->real_num_rx_queues)
> +		return -ENXIO;
> +
> +	/* Schedule the peer side NAPI to receive */
> +	if (!napi_if_scheduled_mark_missed(&peer_rq->xdp_napi))
> +		napi_schedule(&peer_rq->xdp_napi);
> +
> +	return 0;
> +}
> +
>   static int veth_xdp(struct net_device *dev, struct netdev_bpf *xdp)
>   {
>   	switch (xdp->command) {
> @@ -1123,6 +1298,28 @@ static int veth_xdp(struct net_device *dev, struct netdev_bpf *xdp)
>   	case XDP_QUERY_PROG:
>   		xdp->prog_id = veth_xdp_query(dev);
>   		return 0;
> +	case XDP_QUERY_XSK_UMEM:
> +		return veth_xsk_umem_query(dev, &xdp->xsk.umem,
> +					   xdp->xsk.queue_id);
> +	case XDP_SETUP_XSK_UMEM: {
> +		struct veth_priv *priv;
> +		int err;
> +
> +		/* Enable NAPI on both sides, by enabling
> +		 * their XDP.
> +		 */
> +		err = veth_enable_xdp(dev);

Looks like there is no need to enable XDP on this side. You only use 
peer's NAPI, right?

> +		if (err)
> +			return err;
> +
> +		priv = netdev_priv(dev);
> +		err = veth_enable_xdp(priv->peer);

Enabling NAPI here and never disable it?
Also, what happens if peer disables XDP later by detaching an XDP program?
Probably you need something like refcounting.


BTW I'm not 100% sure if current way of accessing peer rq in NAPI 
handler is safe although I did not find an obvious problem.

Looking into physical NIC drivers, ndo_async_xmit() is expected to kick 
TX softirq? This should be something like below process in veth.

Say you attach an xsk to veth A, and B is the peer of A.

1. async_xmit kicks A's NAPI
2. A's NAPI drains xsk tx_ring and pushes frames into B's xdp_ring, then 
kicks B's NAPI
3. B's NAPI drains xdp_ring and process xdp_frames in the same way as 
normal xdp_frames.

What you are currently doing seems to be skipping 2 and making B's NAPI 
directly drain tx_ring of xsk bound to A.

I don't have particular opinion about which is better. Probably your 
approach is more efficient performance-wise? If later you find some race 
with your approach, please consider more physical-NIC-like approach I 
described above.

> +		if (err)
> +			return err;
> +
> +		return veth_xsk_umem_setup(dev, xdp->xsk.umem,
> +					   xdp->xsk.queue_id);
> +	}
>   	default:
>   		return -EINVAL;
>   	}
> @@ -1145,6 +1342,7 @@ static const struct net_device_ops veth_netdev_ops = {
>   	.ndo_set_rx_headroom	= veth_set_rx_headroom,
>   	.ndo_bpf		= veth_xdp,
>   	.ndo_xdp_xmit		= veth_xdp_xmit,
> +	.ndo_xsk_async_xmit	= veth_xsk_async_xmit,
>   };
>   
>   #define VETH_FEATURES (NETIF_F_SG | NETIF_F_FRAGLIST | NETIF_F_HW_CSUM | \
> 

Powered by blists - more mailing lists