Re: [RFC] net: xsk: add a simple buffer reuse queue

["Björn Töpel" <bjorn.topel@intel.com>]

On 2018-08-29 21:19, Jakub Kicinski wrote:
> XSK UMEM is strongly single producer single consumer so reuse of
> frames is challenging.  Add a simple "stash" of FILL packets to
> reuse for drivers to optionally make use of.  This is useful
> when driver has to free (ndo_stop) or resize a ring with an active
> AF_XDP ZC socket.
>

I'll take a stab using this in i40e. I have a couple of
comments/thoughts on this RFC, but let me get back when I have an actual
patch in place. :-)


Thanks!
Björn


> Signed-off-by: Jakub Kicinski <jakub.kicinski@netronome.com>
> ---
>   include/net/xdp_sock.h | 44 +++++++++++++++++++++++++++++++++
>   net/xdp/xdp_umem.c     |  2 ++
>   net/xdp/xsk_queue.c    | 56 ++++++++++++++++++++++++++++++++++++++++++
>   net/xdp/xsk_queue.h    |  3 +++
>   4 files changed, 105 insertions(+)
> 
> diff --git a/include/net/xdp_sock.h b/include/net/xdp_sock.h
> index 6871e4755975..108c1c100de4 100644
> --- a/include/net/xdp_sock.h
> +++ b/include/net/xdp_sock.h
> @@ -14,6 +14,7 @@
>   #include <net/sock.h>
>   
>   struct net_device;
> +struct xdp_umem_fq_reuse;
>   struct xsk_queue;
>   
>   struct xdp_umem_props {
> @@ -41,6 +42,7 @@ struct xdp_umem {
>   	struct page **pgs;
>   	u32 npgs;
>   	struct net_device *dev;
> +	struct xdp_umem_fq_reuse *fq_reuse;
>   	u16 queue_id;
>   	bool zc;
>   	spinlock_t xsk_list_lock;
> @@ -110,4 +112,46 @@ static inline dma_addr_t xdp_umem_get_dma(struct xdp_umem *umem, u64 addr)
>   	return umem->pages[addr >> PAGE_SHIFT].dma + (addr & (PAGE_SIZE - 1));
>   }
>   
> +struct xdp_umem_fq_reuse {
> +	u32 nentries;
> +	u32 length;
> +	u64 handles[];
> +};
> +
> +/* Following functions are not thread-safe in any way */
> +struct xdp_umem_fq_reuse *xsk_reuseq_prepare(u32 nentries);
> +struct xdp_umem_fq_reuse *xsk_reuseq_swap(struct xdp_umem *umem,
> +					  struct xdp_umem_fq_reuse *newq);
> +void xsk_reuseq_free(struct xdp_umem_fq_reuse *rq);
> +
> +/* Reuse-queue aware version of FILL queue helpers */
> +static inline u64 *xsk_umem_peek_addr_rq(struct xdp_umem *umem, u64 *addr)
> +{
> +	struct xdp_umem_fq_reuse *rq = umem->fq_reuse;
> +
> +	if (!rq->length) {
> +		return xsk_umem_peek_addr(umem, addr);
> +	} else {
> +		*addr = rq->handles[rq->length - 1];
> +		return addr;
> +	}
> +}
> +
> +static inline void xsk_umem_discard_addr_rq(struct xdp_umem *umem)
> +{
> +	struct xdp_umem_fq_reuse *rq = umem->fq_reuse;
> +
> +	if (!rq->length)
> +		xsk_umem_discard_addr(umem);
> +	else
> +		rq->length--;
> +}
> +
> +static inline void xsk_umem_fq_reuse(struct xdp_umem *umem, u64 addr)
> +{
> +	struct xdp_umem_fq_reuse *rq = umem->fq_reuse;
> +
> +	rq->handles[rq->length++] = addr;
> +}
> +
>   #endif /* _LINUX_XDP_SOCK_H */
> diff --git a/net/xdp/xdp_umem.c b/net/xdp/xdp_umem.c
> index e762310c9bee..40303e24c954 100644
> --- a/net/xdp/xdp_umem.c
> +++ b/net/xdp/xdp_umem.c
> @@ -170,6 +170,8 @@ static void xdp_umem_release(struct xdp_umem *umem)
>   		umem->cq = NULL;
>   	}
>   
> +	xsk_reuseq_destroy(umem);
> +
>   	xdp_umem_unpin_pages(umem);
>   
>   	task = get_pid_task(umem->pid, PIDTYPE_PID);
> diff --git a/net/xdp/xsk_queue.c b/net/xdp/xsk_queue.c
> index 6c32e92e98fc..f9ee40a13a9a 100644
> --- a/net/xdp/xsk_queue.c
> +++ b/net/xdp/xsk_queue.c
> @@ -3,7 +3,9 @@
>    * Copyright(c) 2018 Intel Corporation.
>    */
>   
> +#include <linux/log2.h>
>   #include <linux/slab.h>
> +#include <linux/overflow.h>
>   
>   #include "xsk_queue.h"
>   
> @@ -61,3 +63,57 @@ void xskq_destroy(struct xsk_queue *q)
>   	page_frag_free(q->ring);
>   	kfree(q);
>   }
> +
> +struct xdp_umem_fq_reuse *xsk_reuseq_prepare(u32 nentries)
> +{
> +	struct xdp_umem_fq_reuse *newq;
> +
> +	/* Check for overflow */
> +	if (nentries > (u32)roundup_pow_of_two(nentries))
> +		return NULL;
> +	nentries = roundup_pow_of_two(nentries);
> +
> +	newq = kvmalloc(struct_size(newq, handles, nentries), GFP_KERNEL);
> +	if (!newq)
> +		return NULL;
> +	memset(newq, 0, offsetof(typeof(*newq), handles));
> +
> +	newq->nentries = nentries;
> +	return newq;
> +}
> +EXPORT_SYMBOL_GPL(xsk_reuseq_prepare);
> +
> +struct xdp_umem_fq_reuse *xsk_reuseq_swap(struct xdp_umem *umem,
> +					  struct xdp_umem_fq_reuse *newq)
> +{
> +	struct xdp_umem_fq_reuse *oldq = umem->fq_reuse;
> +
> +	if (!oldq) {
> +		umem->fq_reuse = newq;
> +		return NULL;
> +	}
> +
> +	if (newq->nentries < oldq->length)
> +		return newq;
> +
> +
> +	memcpy(newq->handles, oldq->handles,
> +	       array_size(oldq->length, sizeof(u64)));
> +	newq->length = oldq->length;
> +
> +	umem->fq_reuse = newq;
> +	return oldq;
> +}
> +EXPORT_SYMBOL_GPL(xsk_reuseq_swap);
> +
> +void xsk_reuseq_free(struct xdp_umem_fq_reuse *rq)
> +{
> +	kvfree(rq);
> +}
> +EXPORT_SYMBOL_GPL(xsk_reuseq_free);
> +
> +void xsk_reuseq_destroy(struct xdp_umem *umem)
> +{
> +	xsk_reuseq_free(umem->fq_reuse);
> +	umem->fq_reuse = NULL;
> +}
> diff --git a/net/xdp/xsk_queue.h b/net/xdp/xsk_queue.h
> index 8a64b150be54..7a480e3eb35d 100644
> --- a/net/xdp/xsk_queue.h
> +++ b/net/xdp/xsk_queue.h
> @@ -257,4 +257,7 @@ void xskq_set_umem(struct xsk_queue *q, struct xdp_umem_props *umem_props);
>   struct xsk_queue *xskq_create(u32 nentries, bool umem_queue);
>   void xskq_destroy(struct xsk_queue *q_ops);
>   
> +/* Executed by the core when the entire UMEM gets freed */
> +void xsk_reuseq_destroy(struct xdp_umem *umem);
> +
>   #endif /* _LINUX_XSK_QUEUE_H */
>