Re: [PATCH net-next v1 07/11] xsk: Deliver provider chunks without copying

[Stanislav Fomichev <sdf.kernel@gmail.com>]

On 11/10, Björn Töpel wrote:
> A provider socket still receives through the copy path, which takes
> FILL entries without the provider's ownership checks: a chunk that a
> page pool holds could be handed out twice. No driver opts in yet.
> 
> Put packets whose buffers all belong to the socket's provider on the
> RX ring as they are, and copy others into chunks the provider claims.
> Only the queue NAPI may produce to such a socket, so generic XDP and
> CPUMAP redirects to it are dropped.
> 
> A device may place payload at chunk offset 0, as fbnic does, so only
> the first buffer of a packet keeps the UMEM headroom. Such buffers may
> be sent back as they are, so only descriptors with XDP_TX_METADATA
> need room for TX metadata in front.
> 
> Signed-off-by: Björn Töpel <bjorn@kernel.org>
> ---
>  Documentation/networking/af_xdp.rst          |   6 +-
>  Documentation/networking/xsk-tx-metadata.rst |   3 +-
>  include/net/page_pool/memory_provider.h      |   2 +
>  net/core/page_pool.c                         |  34 +++-
>  net/xdp/xsk.c                                |  82 ++++++++-
>  net/xdp/xsk.h                                |   5 +
>  net/xdp/xsk_pp.c                             | 165 +++++++++++++++++++
>  net/xdp/xsk_queue.h                          |  18 +-
>  8 files changed, 304 insertions(+), 11 deletions(-)
> 
> diff --git a/Documentation/networking/af_xdp.rst b/Documentation/networking/af_xdp.rst
> index cc3f0d16b28f..e73a0005e169 100644
> --- a/Documentation/networking/af_xdp.rst
> +++ b/Documentation/networking/af_xdp.rst
> @@ -428,8 +428,10 @@ area and that your largest packet size can be 2K.
>  There is also an option to set the headroom of each single buffer in
>  the UMEM. If you set this to N bytes, it means that the packet will
>  start N bytes into the buffer leaving the first N bytes for the
> -application to use. The final option is the flags field, but it will
> -be dealt with in separate sections for each UMEM flag.
> +application to use. In a multi-buffer packet, only the first buffer is
> +guaranteed this headroom; the others may start at offset 0 of their
> +chunk. The final option is the flags field, but it will be dealt with
> +in separate sections for each UMEM flag.
>  
>  SO_BINDTODEVICE setsockopt
>  --------------------------
> diff --git a/Documentation/networking/xsk-tx-metadata.rst b/Documentation/networking/xsk-tx-metadata.rst
> index f240930ca1df..8bb74a569c26 100644
> --- a/Documentation/networking/xsk-tx-metadata.rst
> +++ b/Documentation/networking/xsk-tx-metadata.rst
> @@ -59,7 +59,8 @@ The flags field enables the particular offload:
>  Besides the flags above, in order to trigger the offloads, the first
>  packet's ``struct xdp_desc`` descriptor should set ``XDP_TX_METADATA``
>  bit in the ``options`` field. Also note that in a multi-buffer packet
> -only the first chunk should carry the metadata.
> +only the first chunk should carry the metadata. Only a descriptor with
> +``XDP_TX_METADATA`` needs ``tx_metadata_len`` bytes in front of it.
>  
>  Software TX Checksum
>  ====================
> diff --git a/include/net/page_pool/memory_provider.h b/include/net/page_pool/memory_provider.h
> index daec73c9dcee..849db08513b8 100644
> --- a/include/net/page_pool/memory_provider.h
> +++ b/include/net/page_pool/memory_provider.h
> @@ -49,6 +49,8 @@ struct memory_provider_ops {
>  bool net_mp_niov_set_dma_addr(struct net_iov *niov, dma_addr_t addr);
>  void net_mp_niov_set_page_pool(struct page_pool *pool, struct net_iov *niov);
>  void net_mp_niov_clear_page_pool(struct net_iov *niov);
> +void net_mp_release_page_pool_bulk(struct page_pool *pool,
> +				   netmem_ref *netmems, u32 count);
>  
>  int netif_mp_open_rxq(struct net_device *dev, unsigned int rxq_idx,
>  		      const struct pp_memory_provider_params *p,
> diff --git a/net/core/page_pool.c b/net/core/page_pool.c
> index e897e79d31f1..56c299f2280d 100644
> --- a/net/core/page_pool.c
> +++ b/net/core/page_pool.c
> @@ -1401,7 +1401,7 @@ bool net_mp_niov_set_dma_addr(struct net_iov *niov, dma_addr_t addr)
>  }
>  
>  /* Associate a niov with a page pool. Should follow with a matching
> - * net_mp_niov_clear_page_pool()
> + * net_mp_niov_clear_page_pool() or net_mp_release_page_pool_bulk()
>   */
>  void net_mp_niov_set_page_pool(struct page_pool *pool, struct net_iov *niov)
>  {
> @@ -1422,3 +1422,35 @@ void net_mp_niov_clear_page_pool(struct net_iov *niov)
>  
>  	page_pool_clear_pp_info(netmem);
>  }
> +
> +/**
> + * net_mp_release_page_pool_bulk() - take net_iovs back from a page pool
> + * @pool: page pool that holds one reference to each net_iov
> + * @netmems: net_iovs of the memory provider of @pool
> + * @count: number of entries in @netmems
> + *
> + * A memory provider takes net_iovs out of a page pool without
> + * ->release_netmem(), for example to give them to userspace. @pool may be
> + * freed on return.
> + */
> +void net_mp_release_page_pool_bulk(struct page_pool *pool,
> +				   netmem_ref *netmems, u32 count)
> +{
> +	bool trace = trace_page_pool_state_release_enabled();
> +	atomic_t *release_cnt = &pool->pages_state_release_cnt;
> +	int released;
> +	u32 i;
> +
> +	for (i = 0; i < count; i++) {
> +		DEBUG_NET_WARN_ON_ONCE(netmem_get_pp(netmems[i]) != pool);
> +		page_pool_clear_pp_info(netmems[i]);
> +		if (!trace)
> +			continue;
> +
> +		released = atomic_inc_return_relaxed(release_cnt);
> +		trace_page_pool_state_release(pool, netmems[i], released);
> +	}
> +
> +	if (!trace)
> +		atomic_add(count, release_cnt);
> +}
> diff --git a/net/xdp/xsk.c b/net/xdp/xsk.c
> index 113b0b08356d..b419c673e3f1 100644
> --- a/net/xdp/xsk.c
> +++ b/net/xdp/xsk.c
> @@ -386,6 +386,54 @@ static int __xsk_rcv(struct xdp_sock *xs, struct xdp_buff *xdp, u32 len)
>  	return -ENOMEM;
>  }
>  
> +static int xsk_pp_rcv_copy(struct xdp_sock *xs, struct xdp_buff *xdp, u32 len)
> +{
> +	u32 frame_size = __xsk_pool_get_rx_frame_size(xs->pool);
> +	u32 headroom = xsk_pool_get_headroom(xs->pool);
> +	void *copy_from = xsk_copy_xdp_start(xdp);
> +	u32 from_len, meta_len, rem, num_desc, i;
> +	u64 addrs[MAX_SKB_FRAGS + 1];
> +	skb_frag_t *frag = NULL;
> +	u32 to_len, copied;
> +	void *to;
> +
> +	from_len = xdp->data_end - copy_from;
> +	meta_len = xdp->data - copy_from;
> +	rem = len + meta_len;
> +
> +	num_desc = (len - 1) / frame_size + 1;
> +	if (num_desc > ARRAY_SIZE(addrs)) {
> +		xs->rx_dropped++;
> +		return -ENOSPC;
> +	}
> +	if (xskq_prod_nb_free(xs->rx, num_desc) < num_desc) {
> +		xs->rx_queue_full++;
> +		return -ENOBUFS;
> +	}
> +	if (!xsk_pp_claim_copy(xs->pool->pp, addrs, num_desc)) {
> +		xs->rx_dropped++;
> +		return -ENOMEM;
> +	}
> +
> +	if (xdp_buff_has_frags(xdp))
> +		frag = &xdp_get_shared_info_from_buff(xdp)->frags[0];
> +
> +	for (i = 0; rem && i < num_desc; i++) {
> +		to = xp_raw_get_data(xs->pool, addrs[i] + headroom) - meta_len;
> +		to_len = frame_size + meta_len;
> +		copied = xsk_copy_xdp(to, &copy_from, to_len, &from_len, &frag,
> +				      rem);
> +		rem -= copied;
> +
> +		__xskq_prod_reserve_desc(xs->rx, addrs[i] + headroom,
> +					 copied - meta_len,
> +					 rem ? XDP_PKT_CONTD : 0);
> +		meta_len = 0;
> +	}
> +
> +	return 0;
> +}
> +
>  static bool xsk_tx_writeable(struct xdp_sock *xs)
>  {
>  	if (xskq_cons_present_entries(xs->tx) > xs->tx->nentries / 2)
> @@ -453,8 +501,14 @@ static int xsk_rcv_check(struct xdp_sock *xs, struct xdp_buff *xdp, u32 len)
>  
>  static void xsk_flush(struct xdp_sock *xs)
>  {
> +	struct xsk_pp *pp = xs->pool->pp;
> +
> +	if (pp)
> +		xsk_pp_flush(pp);
>  	xskq_prod_submit(xs->rx);
> -	__xskq_cons_release(xs->pool->fq);
> +	/* A provider releases FILL entries under its lock. */
> +	if (!pp)
> +		__xskq_cons_release(xs->pool->fq);
>  	sock_def_readable(&xs->sk);

optional nit:

if (pp) {
	pp_flush
	prod_submit
} else {
	prod_submit
	cons_release
}

sock_def_readable

(hard to follow these scattered if(pp) checks :-( )

>  }
>  
> @@ -464,6 +518,11 @@ int xsk_generic_rcv(struct xdp_sock *xs, struct xdp_buff *xdp)
>  	int err;
>  
>  	err = xsk_rcv_check(xs, xdp, len);
> +	/* Only the NAPI of the queue feeds a memory provider socket. */
> +	if (!err && xs->pool->pp) {
> +		xs->rx_dropped++;
> +		err = -EOPNOTSUPP;
> +	}

This is confusing? Why even call xsk_rcv_check for pp case?

>  	if (!err) {
>  		spin_lock_bh(&xs->pool->rx_lock);
>  		err = __xsk_rcv(xs, xdp, len);
> @@ -488,7 +547,26 @@ static int xsk_rcv(struct xdp_sock *xs, struct xdp_buff *xdp)
>  		return xsk_rcv_zc(xs, xdp, len);
>  	}
>  
> -	err = __xsk_rcv(xs, xdp, len);
> +	if (!xs->pool->pp) {
> +		err = __xsk_rcv(xs, xdp, len);
> +	} else {

[..]

> +		/* A CPUMAP program runs XDP on another CPU, not in NAPI. */
> +		if (unlikely(!xdp_rxq_info_is_reg(xdp->rxq))) {
> +			xs->rx_dropped++;
> +			return -EINVAL;
> +		}
> +
> +		err = xsk_pp_rcv_zc(xs->pool->pp, xs, xdp);
> +		if (err != -EXDEV)
> +			return err;
> +
> +		/* Copy only packets of the queue, from its NAPI */
> +		if (unlikely(!xsk_pp_queue_buf(xs->pool->pp, xdp))) {
> +			xs->rx_dropped++;
> +			return -EINVAL;
> +		}
> +		err = xsk_pp_rcv_copy(xs, xdp, len);

Move this into __xsk_rcv_pp ?