[PATCH net-next v1 00/11] xsk: Zero copy through page-pool memory providers

[PATCH net-next v1 00/11] xsk: Zero copy through page-pool memory providers

[Björn Töpel]

Hi!

AF_XDP zero copy needs a second RX allocator in every driver. This
series makes UMEM a page-pool memory provider instead: a page_pool
driver opts in, and keeps its refill, recycling and DMA sync code.
fbnic is the first user. Patches 1-7 are core, 8-11 fbnic.

Since the RFC [RFC], net_iov is a first class netmem, the core owns
AF_XDP setup, and a refill contract replaces the NAPI kicks. The fixes
are out.

Changes since the RFC:

 - A readable net_iov has an address. netmem_address() and
   skb_frag_address() find it through struct net_iov_area, which now
   holds niov_shift. xdp_frag_address() is gone. (Mina, Stanislav)
 - PP_FLAG_ALLOW_UNREADABLE_NETMEM keeps its meaning. A pool takes a
   readable provider only with the new PP_FLAG_MEM_PROVIDER. (Mina)
 - The core installs the provider at bind, through the queue API, as
   for devmem and io_uring. A driver sets
   netdev_queue_mgmt_ops.xsk_page_pool, and checks chunk size,
   headroom and multi-buffer mode in XDP_VALIDATE_XSK_POOL. MTU and
   XDP program changes of a bound device get the same check.
   (Stanislav)
 - No NAPI kicks. Drivers report the end of each refill with
   page_pool_refill_done(), and the core publishes the queue's NAPI
   for provider and socket wakeups, so no ndo_xsk_wakeup. (Mina)
 - Page pools declare the RX buffers they supply, header or payload.
   A per-queue generation tells new pools from old ones in the restart
   check.
 - MP_CAP_RETAIN says whether the stack may keep provider buffers.
   The stack copies the others on XDP_PASS, frame conversion and
   non-XSKMAP redirects. Frame copies are linear, so multi-buffer
   frames to devmap and cpumap are dropped.
 - xdp_buff does not grow. The head netmem and shared info live in
   driver memory, reached through a union with txq.
 - The buffer pool allocates xdp_buff_xsk only for copy mode and
   driver zero copy: 128 bytes less per chunk. (Stanislav)
 - With XDP_USE_NEED_WAKEUP the FILL flag stays set. Setting it only
   when starved races with userspace.
 - fbnic is split in four: XDP redirect, netmem heads and XDP_TX,
   unsplittable buffers, zero copy.
 - The RFC's fixes are out. The fbnic one is upstream, and the skb
   head sizing is Clement's fix [1]. Other fixes found on the way go
   to net separately. (Stanislav, Maciej)

Unchanged: aligned chunks only (4 KiB on fbnic), and generic XDP and
CPUMAP cannot feed a provider socket.

fbnic, one queue, 66-byte packets: XSK RX 21.2 Mpps (17.0 with epoll),
XSK TX 28.9 Mpps, XDP_DROP costs 1.9%, 30 instructions per packet,
against net-next. Tested on fbnic hardware and the fbnic QEMU model.

Still open from the RFC: how should userspace learn the chunk
constraints before bind?

netkit queue leases and bnxt come later.


Björn

[RFC] https://lore.kernel.org/netdev/20261002190018.696925-1-bjorn@kernel.org/
[1] https://lore.kernel.org/bpf/20260629072300.1664622-1-clecigne@google.com/

Björn Töpel (11):
  net: Get skb fragment addresses through netmem
  net: Add memory provider capabilities
  xdp: Copy buffers with restricted lifetime
  xsk: Allocate UMEM chunks through page_pool
  xsk: Install UMEM page pool providers in the core
  page_pool: Coordinate refill with memory providers
  xsk: Deliver provider chunks without copying
  eth: fbnic: Support XDP redirect
  eth: fbnic: Use netmem for RX headers and XDP_TX
  eth: fbnic: Support RX buffers it cannot split
  eth: fbnic: Support AF_XDP page pool zero copy

 Documentation/networking/af_xdp.rst           |  64 +-
 Documentation/networking/netmem.rst           |  15 +
 Documentation/networking/xsk-tx-metadata.rst  |   3 +-
 .../net/ethernet/meta/fbnic/fbnic_netdev.c    |  22 +-
 drivers/net/ethernet/meta/fbnic/fbnic_txrx.c  | 601 ++++++++++++---
 drivers/net/ethernet/meta/fbnic/fbnic_txrx.h  |  28 +
 drivers/net/netkit.c                          |   3 +-
 include/linux/netdevice.h                     |   8 +
 include/linux/skbuff.h                        |  22 +-
 include/net/netdev_queues.h                   |  12 +
 include/net/netdev_rx_queue.h                 |   8 +
 include/net/netmem.h                          |  41 +-
 include/net/page_pool/helpers.h               |  38 +-
 include/net/page_pool/memory_provider.h       |  21 +
 include/net/page_pool/types.h                 |  33 +-
 include/net/xdp.h                             |  94 ++-
 include/net/xdp_sock.h                        |   4 +
 include/net/xdp_sock_drv.h                    |  45 ++
 include/net/xsk_buff_pool.h                   |   3 +
 io_uring/zcrx.c                               |   2 +
 net/core/dev.c                                |  30 +-
 net/core/dev.h                                |   8 +
 net/core/devmem.c                             |   9 +-
 net/core/devmem.h                             |   4 +-
 net/core/filter.c                             |  20 +-
 net/core/netdev_config.c                      |   2 +
 net/core/netdev_rx_queue.c                    | 143 +++-
 net/core/page_pool.c                          | 100 ++-
 net/core/page_pool_user.c                     |  25 +-
 net/core/xdp.c                                |  78 +-
 net/ethtool/rings.c                           |   7 +-
 net/xdp/Kconfig                               |   1 +
 net/xdp/Makefile                              |   2 +-
 net/xdp/xsk.c                                 |  86 ++-
 net/xdp/xsk.h                                 |  15 +
 net/xdp/xsk_buff_pool.c                       |  72 +-
 net/xdp/xsk_pp.c                              | 697 ++++++++++++++++++
 net/xdp/xsk_queue.h                           |  18 +-
 38 files changed, 2159 insertions(+), 225 deletions(-)
 create mode 100644 net/xdp/xsk_pp.c


base-commit: 2d916a8c307c31b433318509bf0311d52d509f09
-- 
2.56.0

[PATCH net-next v1 01/11] net: Get skb fragment addresses through netmem

[Björn Töpel]

skb_frag_address() goes through struct page, so a net_iov never has an
address. Only unreadable memory should lack one: XDP and BPF read
fragments through this helper, and AF_XDP UMEM will back readable
net_iovs. netmem_address() also returns NULL, not a bogus pointer, for
an unmapped highmem page.

Generic code needs the size of a net_iov to find its address, so keep
it in struct net_iov_area instead of the devmem binding.

Suggested-by: Mina Almasry <almasrymina@google.com>
Suggested-by: Stanislav Fomichev <sdf@fomichev.me>
Signed-off-by: Björn Töpel <bjorn@kernel.org>
---
 include/linux/skbuff.h | 21 +++++++++------------
 include/net/netmem.h   |  6 ++++--
 io_uring/zcrx.c        |  1 +
 net/core/devmem.c      |  8 ++++----
 net/core/devmem.h      |  4 +---
 5 files changed, 19 insertions(+), 21 deletions(-)

diff --git a/include/linux/skbuff.h b/include/linux/skbuff.h
index 73365c5de6f9..8b0039d49905 100644
--- a/include/linux/skbuff.h
+++ b/include/linux/skbuff.h
@@ -3759,33 +3759,30 @@ int skb_cow_data_for_xdp(struct page_pool *pool, struct sk_buff **pskb,
  * skb_frag_address - gets the address of the data contained in a paged fragment
  * @frag: the paged fragment buffer
  *
- * Returns: the address of the data within @frag. The page must already
- * be mapped.
+ * Returns: the address of the data within @frag, or %NULL if the CPU cannot
+ * read it.
  */
 static inline void *skb_frag_address(const skb_frag_t *frag)
 {
-	if (!skb_frag_page(frag))
+	void *ptr = netmem_address(skb_frag_netmem(frag));
+
+	if (!ptr)
 		return NULL;
 
-	return page_address(skb_frag_page(frag)) + skb_frag_off(frag);
+	return ptr + skb_frag_off(frag);
 }
 
 /**
  * skb_frag_address_safe - gets the address of the data contained in a paged fragment
  * @frag: the paged fragment buffer
  *
- * Returns: the address of the data within @frag. Checks that the page
- * is mapped and returns %NULL otherwise.
+ * Returns: the address of the data within @frag. Checks that the memory
+ * is mapped and readable and returns %NULL otherwise.
  */
 static inline void *skb_frag_address_safe(const skb_frag_t *frag)
 {
-	struct page *page = skb_frag_page(frag);
-	void *ptr;
+	void *ptr = netmem_address(skb_frag_netmem(frag));
 
-	if (!page)
-		return NULL;
-
-	ptr = page_address(page);
 	if (unlikely(!ptr))
 		return NULL;
 
diff --git a/include/net/netmem.h b/include/net/netmem.h
index fb693bc1702a..f196b014184c 100644
--- a/include/net/netmem.h
+++ b/include/net/netmem.h
@@ -72,8 +72,8 @@ enum net_iov_type {
 
 /* A memory descriptor representing abstract networking I/O vectors.
  *
- * net_iovs are allocated and used by networking code, and the size of
- * the chunk is PAGE_SIZE.
+ * net_iovs are allocated and used by networking code. Their area sets the
+ * size of a chunk, see net_iov_area.niov_shift.
  *
  * Examples include imported dmabuf memory and imported io_uring memory. See
  * net_iov_type for all the supported types. While current net_iov types are
@@ -104,6 +104,8 @@ struct net_iov_area {
 	/* Array of net_iovs for this area. */
 	struct net_iov *niovs;
 	size_t num_niovs;
+	/* Each net_iov covers 1 << niov_shift bytes. */
+	unsigned int niov_shift;
 };
 
 static inline struct net_iov_area *net_iov_owner(const struct net_iov *niov)
diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c
index 6d7abb1636e7..555f27e6c1ba 100644
--- a/io_uring/zcrx.c
+++ b/io_uring/zcrx.c
@@ -542,6 +542,7 @@ static int __zcrx_create_area(struct io_zcrx_ifq *ifq,
 	ifq->niov_shift = buf_size_shift;
 	nr_iovs = area->mem.size >> ifq->niov_shift;
 	area->nia.num_niovs = nr_iovs;
+	area->nia.niov_shift = buf_size_shift;
 
 	ret = -ENOMEM;
 	area->nia.niovs = kvmalloc_objs(area->nia.niovs[0], nr_iovs,
diff --git a/net/core/devmem.c b/net/core/devmem.c
index 5c7d9640e9d1..b9146b1f294a 100644
--- a/net/core/devmem.c
+++ b/net/core/devmem.c
@@ -287,8 +287,8 @@ int net_devmem_bind_dmabuf_to_queue(struct net_device *dev, u32 rxq_idx,
 	u32 xa_idx;
 	int err;
 
-	if (binding->niov_shift != PAGE_SHIFT)
-		mp_params.rx_page_size = 1U << binding->niov_shift;
+	if (binding->area.niov_shift != PAGE_SHIFT)
+		mp_params.rx_page_size = 1U << binding->area.niov_shift;
 
 	rxq = __netif_get_rx_queue(dev, rxq_idx);
 	err = xa_alloc(&binding->bound_rxqs, &xa_idx, rxq, xa_limit_32b,
@@ -342,7 +342,7 @@ net_devmem_bind_dmabuf(struct net_device *dev, void *vdev,
 	binding->vdev = vdev;
 	write_pnet(&binding->net, sock_net(nl_sk));
 	binding->uid = sk_uid(nl_sk);
-	binding->niov_shift = niov_shift;
+	binding->area.niov_shift = niov_shift;
 	xa_init_flags(&binding->bound_rxqs, XA_FLAGS_ALLOC);
 
 	err = percpu_ref_init(&binding->ref,
@@ -605,7 +605,7 @@ int mp_dmabuf_devmem_init(struct page_pool *pool)
 	pool->dma_sync = false;
 	pool->dma_sync_for_cpu = false;
 
-	if (pool->p.order != binding->niov_shift - PAGE_SHIFT)
+	if (pool->p.order != binding->area.niov_shift - PAGE_SHIFT)
 		return -E2BIG;
 
 	net_devmem_dmabuf_binding_get(binding);
diff --git a/net/core/devmem.h b/net/core/devmem.h
index c4f476947dfa..448521990ad7 100644
--- a/net/core/devmem.h
+++ b/net/core/devmem.h
@@ -83,8 +83,6 @@ struct net_devmem_dmabuf_binding {
 	 */
 	struct net_iov **tx_vec;
 
-	unsigned int niov_shift;
-
 	struct work_struct unbind_w;
 };
 
@@ -127,7 +125,7 @@ static inline unsigned long net_iov_virtual_addr(const struct net_iov *niov)
 	struct net_devmem_dmabuf_binding *binding =
 		net_devmem_iov_binding(niov);
 
-	return (unsigned long)net_iov_idx(niov) << binding->niov_shift;
+	return (unsigned long)net_iov_idx(niov) << binding->area.niov_shift;
 }
 
 static inline bool
-- 
2.56.0

[PATCH net-next v1 02/11] net: Add memory provider capabilities

[Björn Töpel]

The core treats every memory provider as unreadable: installing one
needs header split and no XDP program, and any page pool that allows
unreadable netmem takes it. An AF_XDP UMEM provider is readable, needs
XDP, and hands out whole buffers.

Describe providers by MP_CAP_READABLE and MP_CAP_FRAG; devmem and
io_uring keep their behavior with MP_CAP_FRAG. A page pool takes a
readable provider only with the new PP_FLAG_MEM_PROVIDER:
PP_FLAG_ALLOW_UNREADABLE_NETMEM says that a driver handles unreadable
netmem, not that it handles any provider.

Signed-off-by: Björn Töpel <bjorn@kernel.org>
---
 Documentation/networking/netmem.rst     |  5 ++++
 drivers/net/netkit.c                    |  3 +-
 include/net/page_pool/helpers.h         | 14 +++++++++-
 include/net/page_pool/memory_provider.h | 14 ++++++++++
 include/net/page_pool/types.h           | 15 ++++++++--
 io_uring/zcrx.c                         |  1 +
 net/core/dev.c                          | 10 ++++---
 net/core/devmem.c                       |  1 +
 net/core/netdev_rx_queue.c              | 37 ++++++++++++++++++++-----
 net/core/page_pool.c                    | 34 +++++++++++++++++++++--
 net/ethtool/rings.c                     |  7 +++--
 net/xdp/xsk_buff_pool.c                 |  3 +-
 12 files changed, 122 insertions(+), 22 deletions(-)

diff --git a/Documentation/networking/netmem.rst b/Documentation/networking/netmem.rst
index 6871dcdd700c..39cb75339704 100644
--- a/Documentation/networking/netmem.rst
+++ b/Documentation/networking/netmem.rst
@@ -103,6 +103,11 @@ Driver RX Requirements
      which knows when dma-syncing is (or is not) appropriate.
    - PP_FLAG_ALLOW_UNREADABLE_NETMEM. The driver must specify this flag iff
      tcp-data-split is enabled.
+   - PP_FLAG_MEM_PROVIDER. The page pool takes memory from the memory
+     provider of its RX queue, readable or not. Page pool creation fails if
+     the provider is unreadable and PP_FLAG_ALLOW_UNREADABLE_NETMEM is not
+     set, so do not set it on header page pools of such a queue. Without it,
+     PP_FLAG_ALLOW_UNREADABLE_NETMEM selects unreadable providers only.
 
 5. The driver must not assume the netmem is readable and/or backed by pages.
    The netmem returned by the page_pool may be unreadable, in which case
diff --git a/drivers/net/netkit.c b/drivers/net/netkit.c
index 82e608e733d9..df672ed5ffcc 100644
--- a/drivers/net/netkit.c
+++ b/drivers/net/netkit.c
@@ -12,6 +12,7 @@
 #include <net/netdev_lock.h>
 #include <net/netdev_queues.h>
 #include <net/netdev_rx_queue.h>
+#include <net/page_pool/memory_provider.h>
 #include <net/xdp_sock_drv.h>
 #include <net/netkit.h>
 #include <net/dst.h>
@@ -289,7 +290,7 @@ static int netkit_xsk(struct net_device *dev, struct netdev_bpf *xdp)
 		ret = -EOPNOTSUPP;
 		goto out;
 	}
-	if (!create || !dev_get_min_mp_channel_count(phys))
+	if (!create || netif_all_mps_have_cap(phys, MP_CAP_READABLE))
 		ret = phys->netdev_ops->ndo_bpf(phys, &xdp_lower);
 out:
 	netdev_unlock(phys);
diff --git a/include/net/page_pool/helpers.h b/include/net/page_pool/helpers.h
index 28635ce9454e..96e6b62cdebb 100644
--- a/include/net/page_pool/helpers.h
+++ b/include/net/page_pool/helpers.h
@@ -62,6 +62,7 @@
 
 #include <linux/dma-mapping.h>
 
+#include <net/page_pool/memory_provider.h>
 #include <net/page_pool/types.h>
 #include <net/net_debug.h>
 #include <net/netmem.h>
@@ -519,7 +520,18 @@ static inline void page_pool_nid_changed(struct page_pool *pool, int new_nid)
  */
 static inline bool page_pool_is_unreadable(struct page_pool *pool)
 {
-	return !!pool->mp_ops;
+	return pool->mp_ops && !(pool->mp_ops->caps & MP_CAP_READABLE);
+}
+
+/**
+ * page_pool_supports_frag() - may page pool buffers be split
+ * @pool: queried page pool
+ *
+ * Return: false if the memory provider of @pool hands out whole buffers only
+ */
+static inline bool page_pool_supports_frag(const struct page_pool *pool)
+{
+	return !pool->mp_ops || pool->mp_ops->caps & MP_CAP_FRAG;
 }
 
 #endif /* _NET_PAGE_POOL_HELPERS_H */
diff --git a/include/net/page_pool/memory_provider.h b/include/net/page_pool/memory_provider.h
index bceb5f7740e2..4417e54a0f81 100644
--- a/include/net/page_pool/memory_provider.h
+++ b/include/net/page_pool/memory_provider.h
@@ -19,6 +19,18 @@ struct sk_buff;
  * Per-provider custom logic must be delegated to memory_provider_ops rather
  * than handled directly in page_pool core code.
  */
+
+/**
+ * enum mp_caps - memory provider capabilities
+ * @MP_CAP_READABLE: The CPU can access provider memory. It may back header
+ *	page pools, and XDP programs may run on it.
+ * @MP_CAP_FRAG: Page pool fragments may split provider buffers.
+ */
+enum mp_caps {
+	MP_CAP_READABLE		= BIT(0),
+	MP_CAP_FRAG		= BIT(1),
+};
+
 struct memory_provider_ops {
 	netmem_ref (*alloc_netmems)(struct page_pool *pool, gfp_t gfp);
 	bool (*release_netmem)(struct page_pool *pool, netmem_ref netmem);
@@ -27,6 +39,7 @@ struct memory_provider_ops {
 	int (*nl_fill)(void *mp_priv, struct sk_buff *rsp,
 		       struct netdev_rx_queue *rxq);
 	void (*uninstall)(void *mp_priv, struct netdev_rx_queue *rxq);
+	u32 caps;
 };
 
 bool net_mp_niov_set_dma_addr(struct net_iov *niov, dma_addr_t addr);
@@ -38,6 +51,7 @@ int netif_mp_open_rxq(struct net_device *dev, unsigned int rxq_idx,
 		      struct netlink_ext_ack *extack);
 int netif_mp_close_rxq(struct net_device *dev, unsigned int rxq_idx,
 		       const struct pp_memory_provider_params *old_p);
+bool netif_all_mps_have_cap(struct net_device *dev, u32 cap);
 
 /**
   * net_mp_netmem_place_in_cache() - give a netmem to a page pool
diff --git a/include/net/page_pool/types.h b/include/net/page_pool/types.h
index 6d543076a0c0..0d48c05f1462 100644
--- a/include/net/page_pool/types.h
+++ b/include/net/page_pool/types.h
@@ -31,8 +31,19 @@
  */
 #define PP_FLAG_ALLOW_UNREADABLE_NETMEM BIT(3)
 
+/* Take memory from the memory provider of the RX queue, if one is installed.
+ * Set page_pool_params.slow.netdev and queue_idx, and create the pool under
+ * the instance lock of the netdev.
+ * An unreadable provider also needs PP_FLAG_ALLOW_UNREADABLE_NETMEM, or page
+ * pool creation fails. Without this flag, PP_FLAG_ALLOW_UNREADABLE_NETMEM
+ * selects unreadable providers only.
+ */
+#define PP_FLAG_MEM_PROVIDER	BIT(4)
+
 #define PP_FLAG_ALL		(PP_FLAG_DMA_MAP | PP_FLAG_DMA_SYNC_DEV | \
-				 PP_FLAG_SYSTEM_POOL | PP_FLAG_ALLOW_UNREADABLE_NETMEM)
+				 PP_FLAG_SYSTEM_POOL | \
+				 PP_FLAG_ALLOW_UNREADABLE_NETMEM | \
+				 PP_FLAG_MEM_PROVIDER)
 
 /* Index limit to stay within PP_DMA_INDEX_BITS for DMA indices */
 #define PP_DMA_INDEX_LIMIT XA_LIMIT(1, BIT(PP_DMA_INDEX_BITS) - 1)
@@ -82,7 +93,7 @@ struct pp_alloc_cache {
  * @netdev:	netdev this pool will serve (leave as NULL if none or multiple)
  * @queue_idx:	queue idx this page_pool is being created for.
  * @flags:	PP_FLAG_DMA_MAP, PP_FLAG_DMA_SYNC_DEV, PP_FLAG_SYSTEM_POOL,
- *		PP_FLAG_ALLOW_UNREADABLE_NETMEM.
+ *		PP_FLAG_ALLOW_UNREADABLE_NETMEM, PP_FLAG_MEM_PROVIDER.
  */
 struct page_pool_params {
 	struct_group_tagged(page_pool_params_fast, fast,
diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c
index 555f27e6c1ba..3b5bf97aa203 100644
--- a/io_uring/zcrx.c
+++ b/io_uring/zcrx.c
@@ -1478,6 +1478,7 @@ static const struct memory_provider_ops io_uring_pp_zc_ops = {
 	.destroy		= io_pp_zc_destroy,
 	.nl_fill		= io_pp_nl_fill,
 	.uninstall		= io_pp_uninstall,
+	.caps			= MP_CAP_FRAG,
 };
 
 static unsigned zcrx_parse_rq(netmem_ref *netmem_array, unsigned nr,
diff --git a/net/core/dev.c b/net/core/dev.c
index 3b60b15e2175..663fe3e32ea0 100644
--- a/net/core/dev.c
+++ b/net/core/dev.c
@@ -10546,8 +10546,9 @@ int netif_xdp_propagate(struct net_device *dev, struct netdev_bpf *bpf)
 		return -EBUSY;
 	}
 
-	if (dev_get_min_mp_channel_count(dev)) {
-		NL_SET_ERR_MSG(bpf->extack, "unable to propagate XDP to device using memory provider");
+	if (!netif_all_mps_have_cap(dev, MP_CAP_READABLE)) {
+		NL_SET_ERR_MSG(bpf->extack,
+			       "unable to propagate XDP to device using unreadable memory provider");
 		return -EBUSY;
 	}
 
@@ -10622,8 +10623,9 @@ static int dev_xdp_install(struct net_device *dev, enum bpf_xdp_mode mode,
 		return -EBUSY;
 	}
 
-	if (dev_get_min_mp_channel_count(dev)) {
-		NL_SET_ERR_MSG(extack, "unable to install XDP to device using memory provider");
+	if (!netif_all_mps_have_cap(dev, MP_CAP_READABLE)) {
+		NL_SET_ERR_MSG(extack,
+			       "unable to install XDP to device using unreadable memory provider");
 		return -EBUSY;
 	}
 
diff --git a/net/core/devmem.c b/net/core/devmem.c
index b9146b1f294a..8f190b574beb 100644
--- a/net/core/devmem.c
+++ b/net/core/devmem.c
@@ -698,4 +698,5 @@ static const struct memory_provider_ops dmabuf_devmem_ops = {
 	.release_netmem		= mp_dmabuf_devmem_release_page,
 	.nl_fill		= mp_dmabuf_devmem_nl_fill,
 	.uninstall		= mp_dmabuf_devmem_uninstall,
+	.caps			= MP_CAP_FRAG,
 };
diff --git a/net/core/netdev_rx_queue.c b/net/core/netdev_rx_queue.c
index a7261fedd5e7..9e14a747f9b5 100644
--- a/net/core/netdev_rx_queue.c
+++ b/net/core/netdev_rx_queue.c
@@ -82,9 +82,13 @@ __netif_get_rx_queue_lease(struct net_device **dev, unsigned int *rxq_idx,
 /* See also page_pool_is_unreadable() */
 bool netif_rxq_has_unreadable_mp(struct net_device *dev, unsigned int rxq_idx)
 {
-	if (rxq_idx < dev->real_num_rx_queues)
-		return __netif_get_rx_queue(dev, rxq_idx)->mp_params.mp_ops;
-	return false;
+	const struct memory_provider_ops *ops;
+
+	if (rxq_idx >= dev->real_num_rx_queues)
+		return false;
+
+	ops = __netif_get_rx_queue(dev, rxq_idx)->mp_params.mp_ops;
+	return ops && !(ops->caps & MP_CAP_READABLE);
 }
 EXPORT_SYMBOL(netif_rxq_has_unreadable_mp);
 
@@ -95,6 +99,22 @@ bool netif_rxq_has_mp(struct net_device *dev, unsigned int rxq_idx)
 	return false;
 }
 
+bool netif_all_mps_have_cap(struct net_device *dev, u32 cap)
+{
+	const struct memory_provider_ops *ops;
+	unsigned int i;
+
+	netdev_assert_locked_ops_compat(dev);
+
+	for (i = 0; i < dev->real_num_rx_queues; i++) {
+		ops = dev->_rx[i].mp_params.mp_ops;
+		if (ops && (ops->caps & cap) != cap)
+			return false;
+	}
+
+	return true;
+}
+
 static int netdev_rx_queue_reconfig(struct net_device *dev,
 				    unsigned int rxq_idx,
 				    struct netdev_queue_config *qcfg_old,
@@ -189,6 +209,7 @@ static int __netif_mp_open_rxq(struct net_device *dev, unsigned int rxq_idx,
 			       struct netlink_ext_ack *extack)
 {
 	const struct netdev_queue_mgmt_ops *qops = dev->queue_mgmt_ops;
+	bool unreadable = !(p->mp_ops->caps & MP_CAP_READABLE);
 	struct netdev_queue_config qcfg[2];
 	struct netdev_rx_queue *rxq;
 	int ret;
@@ -196,16 +217,18 @@ static int __netif_mp_open_rxq(struct net_device *dev, unsigned int rxq_idx,
 	if (!qops)
 		return -EOPNOTSUPP;
 
-	if (dev->cfg->hds_config != ETHTOOL_TCP_DATA_SPLIT_ENABLED) {
+	if (unreadable &&
+	    dev->cfg->hds_config != ETHTOOL_TCP_DATA_SPLIT_ENABLED) {
 		NL_SET_ERR_MSG(extack, "tcp-data-split is disabled");
 		return -EINVAL;
 	}
-	if (dev->cfg->hds_thresh) {
+	if (unreadable && dev->cfg->hds_thresh) {
 		NL_SET_ERR_MSG(extack, "hds-thresh is not zero");
 		return -EINVAL;
 	}
-	if (dev_xdp_prog_count(dev)) {
-		NL_SET_ERR_MSG(extack, "unable to custom memory provider to device with XDP program attached");
+	if (unreadable && dev_xdp_prog_count(dev)) {
+		NL_SET_ERR_MSG(extack,
+			       "unable to use unreadable memory provider with XDP program attached");
 		return -EEXIST;
 	}
 	if (p->rx_page_size && !(qops->supported_params & QCFG_RX_PAGE_SIZE)) {
diff --git a/net/core/page_pool.c b/net/core/page_pool.c
index d7c88c0b67e5..d88a3ea528d4 100644
--- a/net/core/page_pool.c
+++ b/net/core/page_pool.c
@@ -188,6 +188,29 @@ static void page_pool_struct_check(void)
 				    PAGE_POOL_FRAG_GROUP_ALIGN);
 }
 
+static int page_pool_select_mp(struct page_pool *pool,
+			       const struct pp_memory_provider_params *mpp)
+{
+	const struct memory_provider_ops *ops = mpp->mp_ops;
+	unsigned int flags = pool->slow.flags;
+	bool readable;
+
+	if (!ops)
+		return 0;
+
+	readable = ops->caps & MP_CAP_READABLE;
+	if (flags & PP_FLAG_MEM_PROVIDER) {
+		if (!readable && !(flags & PP_FLAG_ALLOW_UNREADABLE_NETMEM))
+			return -EOPNOTSUPP;
+	} else if (readable) {
+		return 0;
+	}
+
+	pool->mp_priv = mpp->mp_priv;
+	pool->mp_ops = ops;
+	return 0;
+}
+
 static int page_pool_init(struct page_pool *pool,
 			  const struct page_pool_params *params,
 			  int cpuid)
@@ -272,12 +295,14 @@ static int page_pool_init(struct page_pool *pool,
 
 	xa_init_flags(&pool->dma_mapped, XA_FLAGS_ALLOC1);
 
-	if (pool->slow.flags & PP_FLAG_ALLOW_UNREADABLE_NETMEM) {
+	if (pool->slow.flags & (PP_FLAG_MEM_PROVIDER |
+				PP_FLAG_ALLOW_UNREADABLE_NETMEM)) {
 		netdev_assert_locked(pool->slow.netdev);
 		rxq = __netif_get_rx_queue(pool->slow.netdev,
 					   pool->slow.queue_idx);
-		pool->mp_priv = rxq->mp_params.mp_priv;
-		pool->mp_ops = rxq->mp_params.mp_ops;
+		err = page_pool_select_mp(pool, &rxq->mp_params);
+		if (err)
+			goto free_ptr_ring;
 	}
 
 	if (pool->mp_ops) {
@@ -1070,6 +1095,9 @@ netmem_ref page_pool_alloc_frag_netmem(struct page_pool *pool,
 	unsigned int max_size = PAGE_SIZE << pool->p.order;
 	netmem_ref netmem = pool->frag_page;
 
+	if (static_branch_unlikely(&page_pool_mem_providers) &&
+	    WARN_ON_ONCE(!page_pool_supports_frag(pool)))
+		return 0;
 	if (WARN_ON(size > max_size))
 		return 0;
 
diff --git a/net/ethtool/rings.c b/net/ethtool/rings.c
index e3810c0320e3..378e171d4e10 100644
--- a/net/ethtool/rings.c
+++ b/net/ethtool/rings.c
@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: GPL-2.0-only
 
 #include <net/netdev_queues.h>
+#include <net/page_pool/memory_provider.h>
 
 #include "common.h"
 #include "netlink.h"
@@ -258,15 +259,15 @@ ethnl_set_rings(struct ethnl_req_info *req_info, struct genl_info *info)
 		return -EINVAL;
 	}
 
-	if (dev_get_min_mp_channel_count(dev)) {
+	if (!netif_all_mps_have_cap(dev, MP_CAP_READABLE)) {
 		if (kernel_ringparam.tcp_data_split !=
 		    ETHTOOL_TCP_DATA_SPLIT_ENABLED) {
 			NL_SET_ERR_MSG(info->extack,
-				       "can't disable tcp-data-split while device has memory provider enabled");
+				       "can't disable tcp-data-split while device has unreadable memory provider enabled");
 			return -EINVAL;
 		} else if (kernel_ringparam.hds_thresh) {
 			NL_SET_ERR_MSG(info->extack,
-				       "can't set non-zero hds_thresh while device is memory provider enabled");
+				       "can't set non-zero hds_thresh while device has unreadable memory provider enabled");
 			return -EINVAL;
 		}
 	}
diff --git a/net/xdp/xsk_buff_pool.c b/net/xdp/xsk_buff_pool.c
index 7750d2ec4ee2..18966bbdb5b7 100644
--- a/net/xdp/xsk_buff_pool.c
+++ b/net/xdp/xsk_buff_pool.c
@@ -2,6 +2,7 @@
 
 #include <linux/netdevice.h>
 #include <net/netdev_lock.h>
+#include <net/page_pool/memory_provider.h>
 #include <net/xsk_buff_pool.h>
 #include <net/xdp_sock.h>
 #include <net/xdp_sock_drv.h>
@@ -241,7 +242,7 @@ int xp_assign_dev(struct xsk_buff_pool *pool,
 		goto err_unreg_pool;
 	}
 
-	if (dev_get_min_mp_channel_count(netdev)) {
+	if (!netif_all_mps_have_cap(netdev, MP_CAP_READABLE)) {
 		err = -EBUSY;
 		goto err_unreg_pool;
 	}
-- 
2.56.0

[PATCH net-next v1 03/11] xdp: Copy buffers with restricted lifetime

[Björn Töpel]

UMEM stays mapped writable in userspace, and it has a fixed number of
chunks. The stack must not keep its chunks in an skb or an xdp_frame.
Mark the providers whose buffers the stack may keep with
MP_CAP_RETAIN, and copy packets of the others on XDP_PASS, on
conversion to a frame, and on redirects other than XSKMAP.

The head chunk is writable by userspace, so it cannot hold
skb_shared_info, and virt_to_netmem() cannot find a net_iov head.
Drivers pass kernel memory for both to xdp_init_buff_netmem().

A static key keeps the check off while no such page pool exists. A
frame copy is linear, so a multi-buffer packet of such a provider
cannot go to a device or CPU map.

Signed-off-by: Björn Töpel <bjorn@kernel.org>
---
 include/linux/skbuff.h                  |  1 +
 include/net/netmem.h                    | 10 +++
 include/net/page_pool/memory_provider.h |  3 +
 include/net/xdp.h                       | 92 +++++++++++++++++++++++--
 io_uring/zcrx.c                         |  2 +-
 net/core/dev.h                          |  4 ++
 net/core/devmem.c                       |  2 +-
 net/core/filter.c                       | 20 +++++-
 net/core/page_pool.c                    | 30 ++++++++
 net/core/xdp.c                          | 78 +++++++++++++++++----
 10 files changed, 221 insertions(+), 21 deletions(-)

diff --git a/include/linux/skbuff.h b/include/linux/skbuff.h
index 8b0039d49905..a9cc5dd0211a 100644
--- a/include/linux/skbuff.h
+++ b/include/linux/skbuff.h
@@ -2691,6 +2691,7 @@ static inline void skb_add_rx_frag_netmem(struct sk_buff *skb, int i,
 					  int size, unsigned int truesize)
 {
 	DEBUG_NET_WARN_ON_ONCE(size > truesize);
+	DEBUG_NET_WARN_ON_ONCE(!netmem_can_retain(netmem));
 
 	skb_fill_netmem_desc(skb, i, netmem, off, size);
 	skb->len += size;
diff --git a/include/net/netmem.h b/include/net/netmem.h
index f196b014184c..b986ad616334 100644
--- a/include/net/netmem.h
+++ b/include/net/netmem.h
@@ -59,6 +59,7 @@ static_assert(sizeof(struct netmem_desc) <= offsetof(struct page, _refcount));
 /* net_iov */
 
 DECLARE_STATIC_KEY_FALSE(page_pool_mem_providers);
+DECLARE_STATIC_KEY_FALSE(page_pool_restricted);
 
 /*  We overload the LSB of the struct page pointer to indicate whether it's
  *  a page or net_iov.
@@ -414,6 +415,15 @@ static inline bool net_is_devmem_iov(const struct net_iov *niov)
 void net_iov_inc_ref(struct net_iov *niov);
 void net_iov_dec_ref(struct net_iov *niov);
 
+#if IS_ENABLED(CONFIG_PAGE_POOL)
+bool netmem_can_retain(netmem_ref netmem);
+#else
+static inline bool netmem_can_retain(netmem_ref netmem)
+{
+	return true;
+}
+#endif
+
 /**
  * netmem_inc_ref - acquire a non-page_pool reference on a netmem
  * @netmem: netmem to reference
diff --git a/include/net/page_pool/memory_provider.h b/include/net/page_pool/memory_provider.h
index 4417e54a0f81..e732b21be9a2 100644
--- a/include/net/page_pool/memory_provider.h
+++ b/include/net/page_pool/memory_provider.h
@@ -25,10 +25,13 @@ struct sk_buff;
  * @MP_CAP_READABLE: The CPU can access provider memory. It may back header
  *	page pools, and XDP programs may run on it.
  * @MP_CAP_FRAG: Page pool fragments may split provider buffers.
+ * @MP_CAP_RETAIN: The stack may keep page pool references to provider
+ *	buffers after the RX poll, for example in an skb or an xdp_frame.
  */
 enum mp_caps {
 	MP_CAP_READABLE		= BIT(0),
 	MP_CAP_FRAG		= BIT(1),
+	MP_CAP_RETAIN		= BIT(2),
 };
 
 struct memory_provider_ops {
diff --git a/include/net/xdp.h b/include/net/xdp.h
index 07231adfb5f8..1105bb9e3292 100644
--- a/include/net/xdp.h
+++ b/include/net/xdp.h
@@ -81,6 +81,16 @@ enum xdp_buff_flags {
 	 * XDP program is not attached.
 	 */
 	XDP_FLAGS_FRAGS_UNREADABLE	= BIT(2),
+	/* Internal: the head netmem and the shared info are in &xdp_buff.ext.
+	 * Not copied to xdp_frames or skbs.
+	 */
+	XDP_FLAGS_EXT			= BIT(3),
+};
+
+/* For a head buffer that cannot hold kernel data, see xdp_init_buff_netmem() */
+struct xdp_buff_ext {
+	netmem_ref netmem;
+	struct skb_shared_info sinfo;
 };
 
 struct xdp_buff {
@@ -89,7 +99,12 @@ struct xdp_buff {
 	void *data_meta;
 	void *data_hard_start;
 	struct xdp_rxq_info *rxq;
-	struct xdp_txq_info *txq;
+	union {
+		/* DEVMAP egress programs only */
+		struct xdp_txq_info *txq;
+		/* RX buffers with XDP_FLAGS_EXT */
+		struct xdp_buff_ext *ext;
+	};
 
 	union {
 		struct {
@@ -138,7 +153,7 @@ static __always_inline void xdp_buff_set_frag_unreadable(struct xdp_buff *xdp)
 
 static __always_inline u32 xdp_buff_get_skb_flags(const struct xdp_buff *xdp)
 {
-	return xdp->flags;
+	return xdp->flags & ~XDP_FLAGS_EXT;
 }
 
 static __always_inline void xdp_buff_clear_frag_pfmemalloc(struct xdp_buff *xdp)
@@ -176,6 +191,50 @@ xdp_prepare_buff(struct xdp_buff *xdp, unsigned char *hard_start,
 	xdp->data_meta = meta_valid ? data : data + 1;
 }
 
+/* See MP_CAP_RETAIN */
+static __always_inline bool xdp_netmem_restricted(netmem_ref netmem)
+{
+	return IS_ENABLED(CONFIG_PAGE_POOL) &&
+	       static_branch_unlikely(&page_pool_restricted) &&
+	       !netmem_can_retain(netmem);
+}
+
+/**
+ * xdp_init_buff_netmem - initialize an &xdp_buff for a netmem head buffer
+ * @xdp: &xdp_buff to initialize
+ * @frame_sz: frame size
+ * @rxq: RX queue info
+ * @netmem: head buffer
+ * @ext: kernel memory used if @netmem cannot hold the shared info, or the
+ *	stack must not keep it. It must live as long as @xdp.
+ *
+ * A driver that may receive into memory provider buffers uses this instead
+ * of xdp_init_buff().
+ */
+static __always_inline void
+xdp_init_buff_netmem(struct xdp_buff *xdp, u32 frame_sz,
+		     struct xdp_rxq_info *rxq, netmem_ref netmem,
+		     struct xdp_buff_ext *ext)
+{
+	xdp_init_buff(xdp, frame_sz, rxq);
+	if (likely(!netmem_is_net_iov(netmem) &&
+		   !xdp_netmem_restricted(netmem)))
+		return;
+
+	ext->netmem = netmem;
+	xdp->ext = ext;
+	xdp->flags |= XDP_FLAGS_EXT;
+}
+
+static __always_inline netmem_ref
+xdp_buff_head_netmem(const struct xdp_buff *xdp)
+{
+	if (unlikely(xdp->flags & XDP_FLAGS_EXT))
+		return xdp->ext->netmem;
+
+	return virt_to_netmem(xdp->data);
+}
+
 /* Reserve memory area at end-of data area.
  *
  * This macro reserves tailroom in the XDP buffer by limiting the
@@ -189,9 +248,28 @@ xdp_prepare_buff(struct xdp_buff *xdp, unsigned char *hard_start,
 static inline struct skb_shared_info *
 xdp_get_shared_info_from_buff(const struct xdp_buff *xdp)
 {
+	if (unlikely(xdp->flags & XDP_FLAGS_EXT))
+		return &xdp->ext->sinfo;
+
 	return (struct skb_shared_info *)xdp_data_hard_end(xdp);
 }
 
+bool __xdp_buff_needs_copy(const struct xdp_buff *xdp);
+
+/* Must the stack copy a buffer of @xdp before it keeps it? XSK buffers are
+ * copied by their own conversion.
+ */
+static __always_inline bool xdp_buff_needs_copy(const struct xdp_buff *xdp)
+{
+	if (unlikely(xdp->flags & XDP_FLAGS_EXT))
+		return true;
+	if (!IS_ENABLED(CONFIG_PAGE_POOL) ||
+	    !static_branch_unlikely(&page_pool_restricted))
+		return false;
+
+	return __xdp_buff_needs_copy(xdp);
+}
+
 static __always_inline unsigned int
 xdp_get_buff_len(const struct xdp_buff *xdp)
 {
@@ -379,7 +457,7 @@ xdp_update_skb_frags_info(struct sk_buff *skb, u8 nr_frags,
 void xdp_warn(const char *msg, const char *func, const int line);
 #define XDP_WARN(msg) xdp_warn(msg, __func__, __LINE__)
 
-struct sk_buff *xdp_build_skb_from_buff(const struct xdp_buff *xdp);
+struct sk_buff *xdp_build_skb_from_buff(struct xdp_buff *xdp);
 struct sk_buff *xdp_build_skb_from_zc(struct xdp_buff *xdp);
 struct xdp_frame *xdp_convert_zc_to_xdp_frame(struct xdp_buff *xdp);
 struct sk_buff *__xdp_build_skb_from_frame(struct xdp_frame *xdpf,
@@ -420,12 +498,15 @@ int xdp_update_frame_from_buff(const struct xdp_buff *xdp,
 		return -ENOSPC;
 	}
 
+	/* The stack may keep the frame, see xdp_buff_needs_copy(). */
+	DEBUG_NET_WARN_ON_ONCE(xdp_buff_needs_copy(xdp));
+
 	xdp_frame->data = xdp->data;
 	xdp_frame->len  = xdp->data_end - xdp->data;
 	xdp_frame->headroom = headroom - sizeof(*xdp_frame);
 	xdp_frame->metasize = metasize;
 	xdp_frame->frame_sz = xdp->frame_sz;
-	xdp_frame->flags = xdp->flags;
+	xdp_frame->flags = xdp_buff_get_skb_flags(xdp);
 
 	return 0;
 }
@@ -436,7 +517,8 @@ struct xdp_frame *xdp_convert_buff_to_frame(struct xdp_buff *xdp)
 {
 	struct xdp_frame *xdp_frame;
 
-	if (xdp->rxq->mem.type == MEM_TYPE_XSK_BUFF_POOL)
+	if (xdp->rxq->mem.type == MEM_TYPE_XSK_BUFF_POOL ||
+	    xdp_buff_needs_copy(xdp))
 		return xdp_convert_zc_to_xdp_frame(xdp);
 
 	/* Store info in top of packet */
diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c
index 3b5bf97aa203..98d83a8e45e4 100644
--- a/io_uring/zcrx.c
+++ b/io_uring/zcrx.c
@@ -1478,7 +1478,7 @@ static const struct memory_provider_ops io_uring_pp_zc_ops = {
 	.destroy		= io_pp_zc_destroy,
 	.nl_fill		= io_pp_nl_fill,
 	.uninstall		= io_pp_uninstall,
-	.caps			= MP_CAP_FRAG,
+	.caps			= MP_CAP_FRAG | MP_CAP_RETAIN,
 };
 
 static unsigned zcrx_parse_rq(netmem_ref *netmem_array, unsigned nr,
diff --git a/net/core/dev.h b/net/core/dev.h
index 4e9e8cd15697..e3337013ffc8 100644
--- a/net/core/dev.h
+++ b/net/core/dev.h
@@ -13,6 +13,10 @@ struct netlink_ext_ack;
 struct netdev_queue_config;
 struct cpumask;
 struct pp_memory_provider_params;
+struct xdp_buff;
+struct xdp_frame;
+
+struct xdp_frame *xdp_copy_buff_to_frame(const struct xdp_buff *xdp);
 
 /* Random bits of netdevice that don't need to be exposed */
 #define FLOW_LIMIT_HISTORY	(1 << 7)  /* must be ^2 and !overflow buckets */
diff --git a/net/core/devmem.c b/net/core/devmem.c
index 8f190b574beb..dea4a4a17e86 100644
--- a/net/core/devmem.c
+++ b/net/core/devmem.c
@@ -698,5 +698,5 @@ static const struct memory_provider_ops dmabuf_devmem_ops = {
 	.release_netmem		= mp_dmabuf_devmem_release_page,
 	.nl_fill		= mp_dmabuf_devmem_nl_fill,
 	.uninstall		= mp_dmabuf_devmem_uninstall,
-	.caps			= MP_CAP_FRAG,
+	.caps			= MP_CAP_FRAG | MP_CAP_RETAIN,
 };
diff --git a/net/core/filter.c b/net/core/filter.c
index 80380f89adf9..f93839b493b2 100644
--- a/net/core/filter.c
+++ b/net/core/filter.c
@@ -4774,12 +4774,28 @@ int xdp_do_redirect(struct net_device *dev, struct xdp_buff *xdp,
 {
 	struct bpf_redirect_info *ri = bpf_net_ctx_get_ri();
 	enum bpf_map_type map_type = ri->map_type;
+	struct xdp_frame *xdpf;
+	int err;
 
 	if (map_type == BPF_MAP_TYPE_XSKMAP)
 		return __xdp_do_redirect_xsk(ri, dev, xdp, xdp_prog);
 
-	return __xdp_do_redirect_frame(ri, dev, xdp_convert_buff_to_frame(xdp),
-				       xdp_prog);
+	if (likely(!xdp_buff_needs_copy(xdp)))
+		return __xdp_do_redirect_frame(ri, dev,
+					       xdp_convert_buff_to_frame(xdp),
+					       xdp_prog);
+
+	/* Release the source only after the target took the copy. */
+	xdpf = xdp_copy_buff_to_frame(xdp);
+	err = __xdp_do_redirect_frame(ri, dev, xdpf, xdp_prog);
+	if (unlikely(err)) {
+		if (xdpf)
+			xdp_return_frame(xdpf);
+		return err;
+	}
+
+	xdp_return_buff(xdp);
+	return 0;
 }
 EXPORT_SYMBOL_GPL(xdp_do_redirect);
 
diff --git a/net/core/page_pool.c b/net/core/page_pool.c
index d88a3ea528d4..8237fda598ab 100644
--- a/net/core/page_pool.c
+++ b/net/core/page_pool.c
@@ -34,6 +34,10 @@
 
 DEFINE_STATIC_KEY_FALSE(page_pool_mem_providers);
 
+/* Enabled while a pool of a provider without MP_CAP_RETAIN exists. */
+DEFINE_STATIC_KEY_FALSE(page_pool_restricted);
+EXPORT_SYMBOL(page_pool_restricted);
+
 #define DEFER_TIME (msecs_to_jiffies(1000))
 #define DEFER_WARN_INTERVAL (60 * HZ)
 
@@ -188,6 +192,28 @@ static void page_pool_struct_check(void)
 				    PAGE_POOL_FRAG_GROUP_ALIGN);
 }
 
+/**
+ * netmem_can_retain() - may the stack keep @netmem after the RX poll
+ * @netmem: netmem to check
+ *
+ * Context: the caller holds @netmem through its page pool. A plain page
+ * reference does not keep the page in its pool.
+ * Return: false if @netmem belongs to a memory provider without
+ * MP_CAP_RETAIN.
+ */
+bool netmem_can_retain(netmem_ref netmem)
+{
+	const struct memory_provider_ops *ops;
+
+	netmem = netmem_compound_head(netmem);
+	if (!netmem_is_pp(netmem))
+		return true;
+
+	ops = netmem_get_pp(netmem)->mp_ops;
+	return !ops || ops->caps & MP_CAP_RETAIN;
+}
+EXPORT_SYMBOL(netmem_can_retain);
+
 static int page_pool_select_mp(struct page_pool *pool,
 			       const struct pp_memory_provider_params *mpp)
 {
@@ -324,6 +350,8 @@ static int page_pool_init(struct page_pool *pool,
 		}
 
 		static_branch_inc(&page_pool_mem_providers);
+		if (!(pool->mp_ops->caps & MP_CAP_RETAIN))
+			static_branch_inc(&page_pool_restricted);
 	} else if (pool->p.order > MAX_PAGE_ORDER) {
 		err = -EINVAL;
 		goto free_ptr_ring;
@@ -352,6 +380,8 @@ static void page_pool_uninit(struct page_pool *pool)
 #endif
 
 	if (pool->mp_ops) {
+		if (!(pool->mp_ops->caps & MP_CAP_RETAIN))
+			static_branch_dec(&page_pool_restricted);
 		pool->mp_ops->destroy(pool);
 		static_branch_dec(&page_pool_mem_providers);
 	}
diff --git a/net/core/xdp.c b/net/core/xdp.c
index 1d679e8fd649..0cf46d492d91 100644
--- a/net/core/xdp.c
+++ b/net/core/xdp.c
@@ -23,6 +23,8 @@
 #include <trace/events/xdp.h>
 #include <net/xdp_sock_drv.h>
 
+#include "dev.h"
+
 #define REG_STATE_NEW		0x0
 #define REG_STATE_REGISTERED	0x1
 #define REG_STATE_UNREGISTERED	0x2
@@ -559,10 +561,41 @@ void xdp_return_buff(struct xdp_buff *xdp)
 			     xdp->rxq->mem.type, true, xdp);
 
 out:
-	__xdp_return(virt_to_netmem(xdp->data), xdp->rxq->mem.type, true, xdp);
+	__xdp_return(xdp_buff_head_netmem(xdp), xdp->rxq->mem.type, true, xdp);
 }
 EXPORT_SYMBOL_GPL(xdp_return_buff);
 
+bool __xdp_buff_needs_copy(const struct xdp_buff *xdp)
+{
+	const struct skb_shared_info *sinfo;
+	u32 i;
+
+	/* Provider buffers come only from page pool queues */
+	if (xdp->rxq->mem.type != MEM_TYPE_PAGE_POOL)
+		return false;
+	if (!netmem_can_retain(xdp_buff_head_netmem(xdp)))
+		return true;
+	if (likely(!xdp_buff_has_frags(xdp)))
+		return false;
+
+	sinfo = xdp_get_shared_info_from_buff(xdp);
+	for (i = 0; i < sinfo->nr_frags; i++) {
+		if (!netmem_can_retain(skb_frag_netmem(&sinfo->frags[i])))
+			return true;
+	}
+
+	return false;
+}
+EXPORT_SYMBOL_GPL(__xdp_buff_needs_copy);
+
+static void xdp_release_copied(struct xdp_buff *xdp)
+{
+	if (xdp->rxq->mem.type == MEM_TYPE_XSK_BUFF_POOL)
+		xsk_buff_free(xdp);
+	else
+		xdp_return_buff(xdp);
+}
+
 void xdp_attachment_setup(struct xdp_attachment_info *info,
 			  struct netdev_bpf *bpf)
 {
@@ -573,13 +606,19 @@ void xdp_attachment_setup(struct xdp_attachment_info *info,
 }
 EXPORT_SYMBOL_GPL(xdp_attachment_setup);
 
-struct xdp_frame *xdp_convert_zc_to_xdp_frame(struct xdp_buff *xdp)
+/* @xdp stays with the caller, also on success */
+struct xdp_frame *xdp_copy_buff_to_frame(const struct xdp_buff *xdp)
 {
 	unsigned int metasize, totsize;
 	void *addr, *data_to_copy;
 	struct xdp_frame *xdpf;
 	struct page *page;
 
+	/* Copies are linear. XSK frames lose their frags, as before. */
+	if (unlikely(xdp_buff_has_frags(xdp)) &&
+	    xdp->rxq->mem.type != MEM_TYPE_XSK_BUFF_POOL)
+		return NULL;
+
 	/* Clone into a MEM_TYPE_PAGE_ORDER0 xdp_frame. */
 	metasize = xdp_data_meta_unsupported(xdp) ? 0 :
 		   xdp->data - xdp->data_meta;
@@ -607,7 +646,17 @@ struct xdp_frame *xdp_convert_zc_to_xdp_frame(struct xdp_buff *xdp)
 	xdpf->frame_sz = PAGE_SIZE;
 	xdpf->mem_type = MEM_TYPE_PAGE_ORDER0;
 
-	xsk_buff_free(xdp);
+	return xdpf;
+}
+
+struct xdp_frame *xdp_convert_zc_to_xdp_frame(struct xdp_buff *xdp)
+{
+	struct xdp_frame *xdpf;
+
+	xdpf = xdp_copy_buff_to_frame(xdp);
+	if (xdpf)
+		xdp_release_copied(xdp);
+
 	return xdpf;
 }
 EXPORT_SYMBOL_GPL(xdp_convert_zc_to_xdp_frame);
@@ -627,10 +676,12 @@ EXPORT_SYMBOL_GPL(xdp_warn);
  * &xdp_buff: allocate an skb head from the NAPI percpu cache, initialize
  * skb data pointers and offsets, set the recycle bit if the buff is
  * PP-backed, Rx queue index, protocol and update frags info.
+ * A buffer that the stack must not keep is copied and released, see
+ * xdp_buff_needs_copy(). On error, @xdp is not released.
  *
  * Return: new &sk_buff on success, %NULL on error.
  */
-struct sk_buff *xdp_build_skb_from_buff(const struct xdp_buff *xdp)
+struct sk_buff *xdp_build_skb_from_buff(struct xdp_buff *xdp)
 {
 	const struct xdp_rxq_info *rxq = xdp->rxq;
 	const struct skb_shared_info *sinfo;
@@ -638,6 +689,9 @@ struct sk_buff *xdp_build_skb_from_buff(const struct xdp_buff *xdp)
 	u32 nr_frags = 0;
 	int metalen;
 
+	if (unlikely(xdp_buff_needs_copy(xdp)))
+		return xdp_build_skb_from_zc(xdp);
+
 	if (unlikely(xdp_buff_has_frags(xdp))) {
 		sinfo = xdp_get_shared_info_from_buff(xdp);
 		nr_frags = sinfo->nr_frags;
@@ -674,12 +728,12 @@ struct sk_buff *xdp_build_skb_from_buff(const struct xdp_buff *xdp)
 EXPORT_SYMBOL_GPL(xdp_build_skb_from_buff);
 
 /**
- * xdp_copy_frags_from_zc - copy frags from XSk buff to skb
+ * xdp_copy_frags_from_zc - copy frags from an &xdp_buff to an skb
  * @skb: skb to copy frags to
- * @xdp: XSk &xdp_buff from which the frags will be copied
+ * @xdp: XSk buff, or buffer that the stack must not keep
  * @pp: &page_pool backing page allocation, if available
  *
- * Copy all frags from XSk &xdp_buff to the skb to pass it up the stack.
+ * Copy all frags from @xdp to the skb to pass it up the stack.
  * Allocate a new buffer for each frag, copy it and attach to the skb.
  *
  * Return: true on success, false on netmem allocation fail.
@@ -724,14 +778,14 @@ static noinline bool xdp_copy_frags_from_zc(struct sk_buff *skb,
 }
 
 /**
- * xdp_build_skb_from_zc - create an skb from XSk &xdp_buff
- * @xdp: source XSk buff
+ * xdp_build_skb_from_zc - create an skb from a copy of an &xdp_buff
+ * @xdp: source XSk buff, or buffer that the stack must not keep
  *
- * Similar to xdp_build_skb_from_buff(), but for XSk frames. Allocate an skb
+ * Similar to xdp_build_skb_from_buff(), but copies @xdp. Allocate an skb
  * head, new buffer for the head, copy the data and initialize the skb fields.
  * If there are frags, allocate new buffers for them and copy.
  * Buffers are allocated from the system percpu pools to try recycling them.
- * If new skb was built successfully, @xdp is returned to XSk pool's freelist.
+ * If new skb was built successfully, @xdp is released to its owner.
  * On error, it remains untouched and the caller must take care of this.
  *
  * Return: new &sk_buff on success, %NULL on error.
@@ -781,7 +835,7 @@ struct sk_buff *xdp_build_skb_from_zc(struct xdp_buff *xdp)
 		goto out;
 	}
 
-	xsk_buff_free(xdp);
+	xdp_release_copied(xdp);
 
 	skb->protocol = eth_type_trans(skb, rxq->dev);
 
-- 
2.56.0

[PATCH net-next v1 04/11] xsk: Allocate UMEM chunks through page_pool

[Björn Töpel]

AF_XDP zero copy needs a second RX allocator in every driver. A
page_pool driver can take UMEM chunks from a memory provider instead,
and keep its refill, recycling and DMA sync code.

Each aligned chunk becomes a net_iov, whose address netmem_address()
finds without calling the provider. Userspace may post a chunk that
the kernel still owns, so the provider tracks the owner of each chunk
and drops such FILL entries. Chunks that a page pool releases stay in
the kernel for reuse. Nothing installs the provider yet.

Signed-off-by: Björn Töpel <bjorn@kernel.org>
---
 include/net/netmem.h |  25 ++++-
 include/net/xdp.h    |   2 +-
 net/xdp/Kconfig      |   1 +
 net/xdp/Makefile     |   2 +-
 net/xdp/xsk.h        |   6 ++
 net/xdp/xsk_pp.c     | 246 +++++++++++++++++++++++++++++++++++++++++++
 6 files changed, 279 insertions(+), 3 deletions(-)
 create mode 100644 net/xdp/xsk_pp.c

diff --git a/include/net/netmem.h b/include/net/netmem.h
index b986ad616334..a3b67ef32b21 100644
--- a/include/net/netmem.h
+++ b/include/net/netmem.h
@@ -69,6 +69,7 @@ DECLARE_STATIC_KEY_FALSE(page_pool_restricted);
 enum net_iov_type {
 	NET_IOV_DMABUF,
 	NET_IOV_IOURING,
+	NET_IOV_XSK,
 };
 
 /* A memory descriptor representing abstract networking I/O vectors.
@@ -107,6 +108,10 @@ struct net_iov_area {
 	size_t num_niovs;
 	/* Each net_iov covers 1 << niov_shift bytes. */
 	unsigned int niov_shift;
+	/* CPU address of the first net_iov, or NULL if the CPU cannot read
+	 * the area.
+	 */
+	void *vaddr;
 };
 
 static inline struct net_iov_area *net_iov_owner(const struct net_iov *niov)
@@ -119,6 +124,18 @@ static inline unsigned int net_iov_idx(const struct net_iov *niov)
 	return niov - net_iov_owner(niov)->niovs;
 }
 
+static inline void *net_iov_address(const struct net_iov *niov)
+{
+	const struct net_iov_area *area = net_iov_owner(niov);
+	unsigned long off;
+
+	if (!area->vaddr)
+		return NULL;
+
+	off = (unsigned long)net_iov_idx(niov) << area->niov_shift;
+	return area->vaddr + off;
+}
+
 /* Initialize a niov: stamp the owning area, the memory provider type.
  */
 static inline void net_iov_init(struct net_iov *niov,
@@ -345,10 +362,16 @@ static inline void *__netmem_address(netmem_ref netmem)
 	return page_address(__netmem_to_page(netmem));
 }
 
+/**
+ * netmem_address - get the CPU address of the memory behind @netmem
+ * @netmem: netmem to get the address of
+ *
+ * Return: the address, or NULL if the CPU cannot read @netmem.
+ */
 static inline void *netmem_address(netmem_ref netmem)
 {
 	if (netmem_is_net_iov(netmem))
-		return NULL;
+		return net_iov_address(netmem_to_net_iov(netmem));
 
 	return __netmem_address(netmem);
 }
diff --git a/include/net/xdp.h b/include/net/xdp.h
index 1105bb9e3292..5ab7802ac4e1 100644
--- a/include/net/xdp.h
+++ b/include/net/xdp.h
@@ -368,7 +368,7 @@ static inline bool xdp_buff_add_frag(struct xdp_buff *xdp, netmem_ref netmem,
 
 	if (unlikely(netmem_is_pfmemalloc(netmem)))
 		xdp_buff_set_frag_pfmemalloc(xdp);
-	if (unlikely(netmem_is_net_iov(netmem)))
+	if (unlikely(netmem_is_net_iov(netmem) && !netmem_address(netmem)))
 		xdp_buff_set_frag_unreadable(xdp);
 
 	return true;
diff --git a/net/xdp/Kconfig b/net/xdp/Kconfig
index 71af2febe72a..c9c68d3b3712 100644
--- a/net/xdp/Kconfig
+++ b/net/xdp/Kconfig
@@ -2,6 +2,7 @@
 config XDP_SOCKETS
 	bool "XDP sockets"
 	depends on BPF_SYSCALL
+	select PAGE_POOL
 	default n
 	help
 	  XDP sockets allows a channel between XDP programs and
diff --git a/net/xdp/Makefile b/net/xdp/Makefile
index 30cdc4315f42..6e9d60f9f7f1 100644
--- a/net/xdp/Makefile
+++ b/net/xdp/Makefile
@@ -1,4 +1,4 @@
 # SPDX-License-Identifier: GPL-2.0-only
 obj-$(CONFIG_XDP_SOCKETS) += xsk.o xdp_umem.o xsk_queue.o xskmap.o
-obj-$(CONFIG_XDP_SOCKETS) += xsk_buff_pool.o
+obj-$(CONFIG_XDP_SOCKETS) += xsk_buff_pool.o xsk_pp.o
 obj-$(CONFIG_XDP_SOCKETS_DIAG) += xsk_diag.o
diff --git a/net/xdp/xsk.h b/net/xdp/xsk.h
index 7c811b5cce76..4dab7b9a1487 100644
--- a/net/xdp/xsk.h
+++ b/net/xdp/xsk.h
@@ -34,6 +34,12 @@ static inline struct xdp_sock *xdp_sk(struct sock *sk)
 
 void xsk_map_try_sock_delete(struct xsk_map *map, struct xdp_sock *xs,
 			     struct xdp_sock __rcu **map_entry);
+struct xsk_pp;
+
+extern const struct memory_provider_ops xsk_pp_ops;
+
+struct xsk_pp *xsk_pp_create(struct xsk_buff_pool *pool);
+void xsk_pp_put(struct xsk_pp *pp);
 void xsk_clear_pool_at_qid(struct net_device *dev, u16 queue_id);
 int xsk_reg_pool_at_qid(struct net_device *dev, struct xsk_buff_pool *pool,
 			u16 queue_id);
diff --git a/net/xdp/xsk_pp.c b/net/xdp/xsk_pp.c
new file mode 100644
index 000000000000..fdd55265a6cd
--- /dev/null
+++ b/net/xdp/xsk_pp.c
@@ -0,0 +1,246 @@
+// SPDX-License-Identifier: GPL-2.0
+/* AF_XDP UMEM as a page pool memory provider
+ *
+ * Each aligned UMEM chunk is a readable net_iov with one owner at a time:
+ *
+ *   userspace --FILL--> page pool --release--> reuse --alloc--> page pool
+ *       ^                   |
+ *       +-----RX ring-------+
+ */
+
+#include <net/page_pool/helpers.h>
+#include <net/page_pool/memory_provider.h>
+#include <net/xdp_sock_drv.h>
+
+#include "xsk.h"
+#include "xsk_queue.h"
+
+enum xsk_pp_chunk {
+	XSK_PP_USER,	/* owned by userspace */
+	XSK_PP_POOL,	/* owned by a page pool */
+	XSK_PP_REUSE,	/* on the reuse stack */
+};
+
+struct xsk_pp {
+	struct net_iov_area area;
+	struct xsk_buff_pool *pool;
+	/* Protects fq, state and reuse */
+	spinlock_t lock;
+	struct xsk_queue *fq;
+	refcount_t refs;
+	u32 nr_pools;
+	bool dma_need_sync;
+	u8 *state;
+	u32 reuse_cnt;
+	u32 reuse[];
+};
+
+static void xsk_pp_free(struct xsk_pp *pp)
+{
+	kvfree(pp->state);
+	kvfree(pp->area.niovs);
+	kvfree(pp);
+}
+
+void xsk_pp_put(struct xsk_pp *pp)
+{
+	if (refcount_dec_and_test(&pp->refs))
+		xsk_pp_free(pp);
+}
+
+/* @pool must be DMA mapped and have aligned chunks. The provider must be
+ * the only consumer of its FILL ring.
+ */
+struct xsk_pp *xsk_pp_create(struct xsk_buff_pool *pool)
+{
+	struct net_iov *niov;
+	struct xsk_pp *pp;
+	dma_addr_t dma;
+	u32 i;
+
+	pp = kvzalloc_flex(*pp, reuse, pool->heads_cnt, GFP_KERNEL_ACCOUNT);
+	if (!pp)
+		return ERR_PTR(-ENOMEM);
+
+	pp->area.niovs = kvzalloc_objs(*pp->area.niovs, pool->heads_cnt,
+				       GFP_KERNEL_ACCOUNT);
+	pp->state = kvzalloc(pool->heads_cnt, GFP_KERNEL_ACCOUNT);
+	if (!pp->area.niovs || !pp->state) {
+		xsk_pp_free(pp);
+		return ERR_PTR(-ENOMEM);
+	}
+
+	pp->area.num_niovs = pool->heads_cnt;
+	pp->area.niov_shift = pool->chunk_shift;
+	pp->area.vaddr = pool->addrs;
+	pp->pool = pool;
+	pp->fq = pool->fq;
+	pp->dma_need_sync = dma_dev_need_sync(pool->dev);
+	spin_lock_init(&pp->lock);
+	refcount_set(&pp->refs, 1);
+
+	for (i = 0; i < pp->area.num_niovs; i++) {
+		niov = &pp->area.niovs[i];
+		dma = xp_raw_get_dma(pool, (u64)i << pool->chunk_shift);
+		net_iov_init(niov, &pp->area, NET_IOV_XSK);
+		if (net_mp_niov_set_dma_addr(niov, dma)) {
+			xsk_pp_free(pp);
+			return ERR_PTR(-ERANGE);
+		}
+	}
+
+	return pp;
+}
+
+static int xsk_pp_init(struct page_pool *pool)
+{
+	struct xsk_pp *pp = pool->mp_priv;
+
+	if (pool->p.order || pool->p.dma_dir != DMA_BIDIRECTIONAL ||
+	    pool->p.dev != pp->pool->dev)
+		return -EINVAL;
+	if (pool->p.max_len > pp->pool->chunk_size ||
+	    pool->p.offset > pp->pool->chunk_size - pool->p.max_len)
+		return -EINVAL;
+
+	refcount_inc(&pp->refs);
+	spin_lock_bh(&pp->lock);
+	pp->nr_pools++;
+	spin_unlock_bh(&pp->lock);
+	return 0;
+}
+
+static void xsk_pp_destroy(struct page_pool *pool)
+{
+	struct xsk_pp *pp = pool->mp_priv;
+
+	spin_lock_bh(&pp->lock);
+	pp->nr_pools--;
+	spin_unlock_bh(&pp->lock);
+	xsk_pp_put(pp);
+}
+
+static bool xsk_pp_claim(struct xsk_pp *pp, u32 idx, enum xsk_pp_chunk from)
+{
+	lockdep_assert_held(&pp->lock);
+
+	if (pp->state[idx] != from)
+		return false;
+
+	pp->state[idx] = XSK_PP_POOL;
+	return true;
+}
+
+static u32 xsk_pp_alloc_reuse(struct xsk_pp *pp, netmem_ref *netmems, u32 max)
+{
+	u32 idx, n = 0;
+
+	while (n < max && pp->reuse_cnt) {
+		idx = pp->reuse[--pp->reuse_cnt];
+		if (WARN_ON_ONCE(!xsk_pp_claim(pp, idx, XSK_PP_REUSE)))
+			continue;
+		netmems[n++] = net_iov_to_netmem(&pp->area.niovs[idx]);
+	}
+
+	return n;
+}
+
+static u32 xsk_pp_alloc_fill(struct xsk_pp *pp, netmem_ref *netmems, u32 max)
+{
+	struct xsk_buff_pool *pool = pp->pool;
+	struct xsk_queue *fq = pp->fq;
+	u32 i, entries, cons, idx, n = 0;
+	u64 addr;
+
+	if (!fq || !max)
+		return 0;
+
+	entries = xskq_cons_nb_entries(fq, max);
+	if (!entries) {
+		fq->queue_empty_descs++;
+		return 0;
+	}
+
+	cons = fq->cached_cons;
+	for (i = 0; i < entries; i++) {
+		__xskq_cons_read_addr_unchecked(fq, cons++, &addr);
+		addr &= pool->chunk_mask;
+		idx = addr >> pp->area.niov_shift;
+		if (unlikely(addr >= pool->addrs_cnt ||
+			     !xsk_pp_claim(pp, idx, XSK_PP_USER))) {
+			fq->invalid_descs++;
+			continue;
+		}
+		netmems[n++] = net_iov_to_netmem(&pp->area.niovs[idx]);
+	}
+
+	xskq_cons_release_n(fq, entries);
+	__xskq_cons_release(fq);
+	return n;
+}
+
+static netmem_ref xsk_pp_alloc_netmems(struct page_pool *pool, gfp_t gfp)
+{
+	netmem_ref *netmems = pool->alloc.cache;
+	struct xsk_pp *pp = pool->mp_priv;
+	struct net_iov *niov;
+	dma_addr_t dma;
+	u32 i, n;
+
+	if (WARN_ON_ONCE(pool->alloc.count))
+		return 0;
+
+	spin_lock_bh(&pp->lock);
+	n = xsk_pp_alloc_reuse(pp, netmems, PP_ALLOC_CACHE_REFILL);
+	n += xsk_pp_alloc_fill(pp, netmems + n, PP_ALLOC_CACHE_REFILL - n);
+	spin_unlock_bh(&pp->lock);
+	if (!n)
+		return 0;
+
+	for (i = 0; i < n; i++) {
+		niov = netmem_to_net_iov(netmems[i]);
+		dma = page_pool_get_dma_addr_netmem(netmems[i]);
+		if (pp->dma_need_sync)
+			dma_sync_single_range_for_device(pool->p.dev, dma,
+							 pool->p.offset,
+							 pool->p.max_len,
+							 pool->p.dma_dir);
+		net_mp_niov_set_page_pool(pool, niov);
+	}
+
+	pool->alloc.count = n - 1;
+	return netmems[n - 1];
+}
+
+static bool xsk_pp_release_netmem(struct page_pool *pool, netmem_ref netmem)
+{
+	struct net_iov *niov = netmem_to_net_iov(netmem);
+	struct xsk_pp *pp = pool->mp_priv;
+	u32 idx = net_iov_idx(niov);
+
+	spin_lock_bh(&pp->lock);
+	net_mp_niov_clear_page_pool(niov);
+	if (!WARN_ON_ONCE(pp->state[idx] != XSK_PP_POOL ||
+			  pp->reuse_cnt == pp->area.num_niovs)) {
+		pp->state[idx] = XSK_PP_REUSE;
+		pp->reuse[pp->reuse_cnt++] = idx;
+	}
+	spin_unlock_bh(&pp->lock);
+
+	return false;
+}
+
+static int xsk_pp_nl_fill(void *mp_priv, struct sk_buff *rsp,
+			  struct netdev_rx_queue *rxq)
+{
+	return 0;
+}
+
+const struct memory_provider_ops xsk_pp_ops = {
+	.init		= xsk_pp_init,
+	.destroy	= xsk_pp_destroy,
+	.alloc_netmems	= xsk_pp_alloc_netmems,
+	.release_netmem	= xsk_pp_release_netmem,
+	.nl_fill	= xsk_pp_nl_fill,
+	.caps		= MP_CAP_READABLE,
+};
-- 
2.56.0

[PATCH net-next v1 05/11] xsk: Install UMEM page pool providers in the core

[Björn Töpel]

A page_pool driver should not need its own AF_XDP setup. When it sets
netdev_queue_mgmt_ops.xsk_page_pool, a zero-copy bind installs the
UMEM provider through the queue API, as devmem and io_uring do. The
driver only judges the chunk size, headroom and multi-buffer mode
against its MTU and XDP program, in XDP_VALIDATE_XSK_POOL; MTU and
program changes of a bound device get the same check. The UMEM
headroom belongs to the socket, so the queue gets it as rx_headroom,
next to rx_page_size.

After a queue restart, the core checks that the new page pools took
the provider. A readable provider must back the header pool, where an
unsplit packet lands, so pools declare the buffers they supply in
rx_roles. The old queue's pools carry the same provider during the
check, so a per-queue generation, mp_gen, tells the new ones apart.

Generic XDP runs outside the queue NAPI, so it cannot feed the socket.

The provider needs no xdp_buff_xsk per chunk. The buffer pool
allocates them only for copy mode and driver zero copy, which saves
128 bytes per chunk.

Signed-off-by: Björn Töpel <bjorn@kernel.org>
---
 include/linux/netdevice.h     |   8 ++
 include/net/netdev_queues.h   |  12 ++
 include/net/netdev_rx_queue.h |   2 +
 include/net/page_pool/types.h |  12 ++
 include/net/xdp_sock.h        |   4 +
 include/net/xdp_sock_drv.h    |  45 +++++++
 include/net/xsk_buff_pool.h   |   3 +
 net/core/dev.c                |  16 +++
 net/core/netdev_config.c      |   2 +
 net/core/netdev_rx_queue.c    |   9 +-
 net/core/page_pool.c          |   6 +-
 net/core/page_pool_user.c     |  25 ++--
 net/xdp/xsk.h                 |   7 +-
 net/xdp/xsk_buff_pool.c       |  69 ++++++++---
 net/xdp/xsk_pp.c              | 221 ++++++++++++++++++++++++++++++++++
 15 files changed, 410 insertions(+), 31 deletions(-)

diff --git a/include/linux/netdevice.h b/include/linux/netdevice.h
index 435be7880d7e..11e51e684003 100644
--- a/include/linux/netdevice.h
+++ b/include/linux/netdevice.h
@@ -77,6 +77,7 @@ struct udp_tunnel_info;
 struct udp_tunnel_nic_info;
 struct udp_tunnel_nic;
 struct bpf_prog;
+struct xsk_rx_check;
 struct xdp_buff;
 struct xdp_frame;
 struct xdp_metadata_ops;
@@ -1010,6 +1011,8 @@ enum bpf_netdev_command {
 	BPF_OFFLOAD_MAP_ALLOC,
 	BPF_OFFLOAD_MAP_FREE,
 	XDP_SETUP_XSK_POOL,
+	/* Check an AF_XDP page pool setup, see netdev_queue_mgmt_ops */
+	XDP_VALIDATE_XSK_POOL,
 };
 
 struct bpf_prog_offload_ops;
@@ -1048,6 +1051,11 @@ struct netdev_bpf {
 			struct xsk_buff_pool *pool;
 			u16 queue_id;
 		} xsk;
+		/* XDP_VALIDATE_XSK_POOL */
+		struct {
+			const struct xsk_rx_check *check;
+			struct netlink_ext_ack *extack;
+		} xsk_check;
 	};
 };
 
diff --git a/include/net/netdev_queues.h b/include/net/netdev_queues.h
index c5335e935b7d..ad5f0420c21d 100644
--- a/include/net/netdev_queues.h
+++ b/include/net/netdev_queues.h
@@ -45,12 +45,15 @@ struct netdev_config {
 /**
  * struct netdev_queue_config - rendered configuration for an RX queue
  * @rx_page_size:	Size of one RX page-pool allocation.
+ * @rx_headroom:	Space before the packet in a head buffer, including
+ *			XDP_PACKET_HEADROOM. Zero selects the driver default.
  * @rx_ring_size:	Configured size of the regular RX ring.
  * @rx_mini_ring_size:	Configured size of the RX mini ring.
  * @rx_jumbo_ring_size:	Configured size of the RX jumbo ring.
  */
 struct netdev_queue_config {
 	u32	rx_page_size;
+	u32	rx_headroom;
 	u32	rx_ring_size;
 	u32	rx_mini_ring_size;
 	u32	rx_jumbo_ring_size;
@@ -156,6 +159,8 @@ void netdev_stat_queue_sum(struct net_device *netdev,
 enum {
 	/* The queue checks and honours the page size qcfg parameter */
 	QCFG_RX_PAGE_SIZE	= 0x1,
+	/* The queue checks and honours the headroom qcfg parameter */
+	QCFG_RX_HEADROOM	= 0x2,
 };
 
 /**
@@ -199,6 +204,12 @@ enum {
  *
  * @supported_params:	Bitmask of supported parameters, see QCFG_*.
  *
+ * @xsk_page_pool:	AF_XDP zero copy installs the UMEM as the page pool
+ *			memory provider of the RX queue. The driver selects it
+ *			with PP_FLAG_MEM_PROVIDER and PP_RX_ROLE_HEAD, honours
+ *			whole buffers and QCFG_RX_HEADROOM, and accepts or
+ *			rejects the setup in XDP_VALIDATE_XSK_POOL.
+ *
  * Note that @ndo_queue_mem_alloc and @ndo_queue_mem_free may be called while
  * the interface is closed. @ndo_queue_start and @ndo_queue_stop will only
  * be called for an interface which is open.
@@ -229,6 +240,7 @@ struct netdev_queue_mgmt_ops {
 				    struct netlink_ext_ack *extack);
 
 	unsigned int supported_params;
+	bool xsk_page_pool;
 };
 
 void netdev_queue_config(struct net_device *dev, int rxq,
diff --git a/include/net/netdev_rx_queue.h b/include/net/netdev_rx_queue.h
index 9415a94d333d..989120cbf452 100644
--- a/include/net/netdev_rx_queue.h
+++ b/include/net/netdev_rx_queue.h
@@ -31,6 +31,8 @@ struct netdev_rx_queue {
 	struct napi_struct		*napi;
 	struct netdev_queue_config	qcfg;
 	struct pp_memory_provider_params mp_params;
+	/* stamps the page pools of each queue allocation */
+	u32				mp_gen;
 
 	/* If a queue is leased, then the lease pointer is always
 	 * valid. From the physical device it points to the virtual
diff --git a/include/net/page_pool/types.h b/include/net/page_pool/types.h
index 0d48c05f1462..e9f3af91e305 100644
--- a/include/net/page_pool/types.h
+++ b/include/net/page_pool/types.h
@@ -40,6 +40,12 @@
  */
 #define PP_FLAG_MEM_PROVIDER	BIT(4)
 
+/* RX buffers a page pool supplies, see page_pool_params.rx_roles */
+enum pp_rx_role {
+	PP_RX_ROLE_HEAD		= BIT(0),	/* packet headers */
+	PP_RX_ROLE_PAYLOAD	= BIT(1),	/* payload after header split */
+};
+
 #define PP_FLAG_ALL		(PP_FLAG_DMA_MAP | PP_FLAG_DMA_SYNC_DEV | \
 				 PP_FLAG_SYSTEM_POOL | \
 				 PP_FLAG_ALLOW_UNREADABLE_NETMEM | \
@@ -94,6 +100,8 @@ struct pp_alloc_cache {
  * @queue_idx:	queue idx this page_pool is being created for.
  * @flags:	PP_FLAG_DMA_MAP, PP_FLAG_DMA_SYNC_DEV, PP_FLAG_SYSTEM_POOL,
  *		PP_FLAG_ALLOW_UNREADABLE_NETMEM, PP_FLAG_MEM_PROVIDER.
+ * @rx_roles:	&enum pp_rx_role mask, for pools that select a memory
+ *		provider. Zero means payload buffers.
  */
 struct page_pool_params {
 	struct_group_tagged(page_pool_params_fast, fast,
@@ -110,6 +118,7 @@ struct page_pool_params {
 		struct net_device *netdev;
 		unsigned int queue_idx;
 		unsigned int	flags;
+		unsigned int	rx_roles;
 /* private: used by test code only */
 		void (*init_callback)(netmem_ref netmem, void *arg);
 		void *init_arg;
@@ -182,6 +191,7 @@ struct pp_memory_provider_params {
 	void *mp_priv;
 	const struct memory_provider_ops *mp_ops;
 	u32 rx_page_size;
+	u32 rx_headroom;
 };
 
 struct page_pool {
@@ -245,6 +255,8 @@ struct page_pool {
 
 	void *mp_priv;
 	const struct memory_provider_ops *mp_ops;
+	/* queue allocation that selected the provider */
+	u32 mp_gen;
 
 	struct xarray dma_mapped;
 
diff --git a/include/net/xdp_sock.h b/include/net/xdp_sock.h
index 6e70b320b399..e0e22ef839e1 100644
--- a/include/net/xdp_sock.h
+++ b/include/net/xdp_sock.h
@@ -19,6 +19,10 @@
 struct net_device;
 struct xsk_queue;
 struct xdp_buff;
+struct memory_provider_ops;
+
+/* The memory provider of AF_XDP page pool zero copy */
+extern const struct memory_provider_ops xsk_pp_ops;
 
 struct xdp_umem {
 	void *addrs;
diff --git a/include/net/xdp_sock_drv.h b/include/net/xdp_sock_drv.h
index d94aeb506379..5bb1949a0aa7 100644
--- a/include/net/xdp_sock_drv.h
+++ b/include/net/xdp_sock_drv.h
@@ -22,8 +22,29 @@ struct xsk_cb_desc {
 	u8 bytes;
 };
 
+/**
+ * struct xsk_rx_check - AF_XDP page pool RX setup a driver must accept
+ * @buf_size: size of each RX buffer
+ * @headroom: space before the packet in a head buffer, including
+ *	XDP_PACKET_HEADROOM. The UMEM headroom is the rest.
+ * @mtu: MTU of the device
+ * @sg: the socket takes packets in more than one buffer
+ * @prog: XDP program to check, or NULL for the current one
+ */
+struct xsk_rx_check {
+	u32 buf_size;
+	u32 headroom;
+	unsigned int mtu;
+	bool sg;
+	struct bpf_prog *prog;
+};
+
 #ifdef CONFIG_XDP_SOCKETS
 
+int xsk_validate_dev(struct net_device *dev, unsigned int mtu,
+		     struct bpf_prog *prog, enum bpf_xdp_mode mode,
+		     struct netlink_ext_ack *extack);
+struct xsk_buff_pool *xsk_rxq_pool(struct net_device *dev, u16 queue_id);
 void xsk_tx_completed(struct xsk_buff_pool *pool, u32 nb_entries);
 bool xsk_tx_peek_desc(struct xsk_buff_pool *pool, struct xdp_desc *desc);
 u32 xsk_tx_peek_release_desc_batch(struct xsk_buff_pool *pool, u32 max);
@@ -36,6 +57,11 @@ void xsk_clear_rx_need_wakeup(struct xsk_buff_pool *pool);
 void xsk_clear_tx_need_wakeup(struct xsk_buff_pool *pool);
 bool xsk_uses_need_wakeup(struct xsk_buff_pool *pool);
 
+static inline bool xsk_pool_uses_sg(struct xsk_buff_pool *pool)
+{
+	return pool->umem->flags & XDP_UMEM_SG_FLAG;
+}
+
 static inline u32 xsk_pool_get_headroom(struct xsk_buff_pool *pool)
 {
 	return XDP_PACKET_HEADROOM + pool->headroom;
@@ -351,6 +377,20 @@ static inline void xsk_buff_raw_dma_sync_for_device(struct xsk_buff_pool *pool,
 
 #else
 
+static inline int xsk_validate_dev(struct net_device *dev, unsigned int mtu,
+				   struct bpf_prog *prog,
+				   enum bpf_xdp_mode mode,
+				   struct netlink_ext_ack *extack)
+{
+	return 0;
+}
+
+static inline struct xsk_buff_pool *xsk_rxq_pool(struct net_device *dev,
+						 u16 queue_id)
+{
+	return NULL;
+}
+
 static inline void xsk_tx_completed(struct xsk_buff_pool *pool, u32 nb_entries)
 {
 }
@@ -397,6 +437,11 @@ static inline bool xsk_uses_need_wakeup(struct xsk_buff_pool *pool)
 	return false;
 }
 
+static inline bool xsk_pool_uses_sg(struct xsk_buff_pool *pool)
+{
+	return false;
+}
+
 static inline u32 xsk_pool_get_headroom(struct xsk_buff_pool *pool)
 {
 	return 0;
diff --git a/include/net/xsk_buff_pool.h b/include/net/xsk_buff_pool.h
index 02e89350faf9..6d7cfdf99a8b 100644
--- a/include/net/xsk_buff_pool.h
+++ b/include/net/xsk_buff_pool.h
@@ -11,6 +11,7 @@
 #include <net/xdp.h>
 
 struct xsk_buff_pool;
+struct xsk_pp;
 struct xdp_rxq_info;
 struct xsk_cb_desc;
 struct xsk_queue;
@@ -54,6 +55,8 @@ struct xsk_buff_pool {
 	spinlock_t xsk_tx_list_lock;
 	refcount_t users;
 	struct xdp_umem *umem;
+	/* page pool memory provider, see net/xdp/xsk_pp.c */
+	struct xsk_pp *pp;
 	struct work_struct work;
 	/* Protects generic receive in shared and non-shared umem mode. */
 	spinlock_t rx_lock;
diff --git a/net/core/dev.c b/net/core/dev.c
index 663fe3e32ea0..7e985fb82488 100644
--- a/net/core/dev.c
+++ b/net/core/dev.c
@@ -163,6 +163,7 @@
 #include <net/page_pool/types.h>
 #include <net/page_pool/helpers.h>
 #include <net/page_pool/memory_provider.h>
+#include <net/xdp_sock_drv.h>
 #include <net/rps.h>
 #include <linux/phy_link_topology.h>
 
@@ -10535,6 +10536,8 @@ u8 dev_xdp_sb_prog_count(struct net_device *dev)
 
 int netif_xdp_propagate(struct net_device *dev, struct netdev_bpf *bpf)
 {
+	int err;
+
 	if (!dev->netdev_ops->ndo_bpf)
 		return -EOPNOTSUPP;
 
@@ -10552,6 +10555,13 @@ int netif_xdp_propagate(struct net_device *dev, struct netdev_bpf *bpf)
 		return -EBUSY;
 	}
 
+	if (bpf->command == XDP_SETUP_PROG && bpf->prog) {
+		err = xsk_validate_dev(dev, dev->mtu, bpf->prog, XDP_MODE_DRV,
+				       bpf->extack);
+		if (err)
+			return err;
+	}
+
 	return dev->netdev_ops->ndo_bpf(dev, bpf);
 }
 EXPORT_SYMBOL_GPL(netif_xdp_propagate);
@@ -10629,6 +10639,12 @@ static int dev_xdp_install(struct net_device *dev, enum bpf_xdp_mode mode,
 		return -EBUSY;
 	}
 
+	if (mode != XDP_MODE_HW && prog) {
+		err = xsk_validate_dev(dev, dev->mtu, prog, mode, extack);
+		if (err)
+			return err;
+	}
+
 	memset(&xdp, 0, sizeof(xdp));
 	xdp.command = mode == XDP_MODE_HW ? XDP_SETUP_PROG_HW : XDP_SETUP_PROG;
 	xdp.extack = extack;
diff --git a/net/core/netdev_config.c b/net/core/netdev_config.c
index 1975de42a60d..97e33764737b 100644
--- a/net/core/netdev_config.c
+++ b/net/core/netdev_config.c
@@ -88,6 +88,8 @@ static int __netdev_queue_config(struct net_device *dev, int rxq_idx,
 	mpp = &__netif_get_rx_queue(dev, rxq_idx)->mp_params;
 	if (mpp->rx_page_size)
 		qcfg->rx_page_size = mpp->rx_page_size;
+	if (mpp->rx_headroom)
+		qcfg->rx_headroom = mpp->rx_headroom;
 	err = validate_cb(dev, qcfg, extack);
 	if (err)
 		return err;
diff --git a/net/core/netdev_rx_queue.c b/net/core/netdev_rx_queue.c
index 9e14a747f9b5..4e2f906fbc73 100644
--- a/net/core/netdev_rx_queue.c
+++ b/net/core/netdev_rx_queue.c
@@ -6,6 +6,7 @@
 #include <net/netdev_queues.h>
 #include <net/netdev_rx_queue.h>
 #include <net/page_pool/memory_provider.h>
+#include <net/xdp_sock.h>
 
 #include "dev.h"
 #include "page_pool_priv.h"
@@ -141,6 +142,7 @@ static int netdev_rx_queue_reconfig(struct net_device *dev,
 		goto err_free_new_mem;
 	}
 
+	rxq->mp_gen++;
 	err = qops->ndo_queue_mem_alloc(dev, qcfg_new, new_mem, rxq_idx);
 	if (err)
 		goto err_free_old_mem;
@@ -235,6 +237,10 @@ static int __netif_mp_open_rxq(struct net_device *dev, unsigned int rxq_idx,
 		NL_SET_ERR_MSG(extack, "device does not support: rx_page_size");
 		return -EOPNOTSUPP;
 	}
+	if (p->rx_headroom && !(qops->supported_params & QCFG_RX_HEADROOM)) {
+		NL_SET_ERR_MSG(extack, "device does not support: rx_headroom");
+		return -EOPNOTSUPP;
+	}
 
 	rxq = __netif_get_rx_queue(dev, rxq_idx);
 	if (rxq->mp_params.mp_ops) {
@@ -242,7 +248,8 @@ static int __netif_mp_open_rxq(struct net_device *dev, unsigned int rxq_idx,
 		return -EEXIST;
 	}
 #ifdef CONFIG_XDP_SOCKETS
-	if (rxq->pool) {
+	/* An AF_XDP pool owns the queue, unless it installs its provider. */
+	if (rxq->pool && p->mp_ops != &xsk_pp_ops) {
 		NL_SET_ERR_MSG(extack, "designated queue already in use by AF_XDP");
 		return -EBUSY;
 	}
diff --git a/net/core/page_pool.c b/net/core/page_pool.c
index 8237fda598ab..e897e79d31f1 100644
--- a/net/core/page_pool.c
+++ b/net/core/page_pool.c
@@ -215,7 +215,8 @@ bool netmem_can_retain(netmem_ref netmem)
 EXPORT_SYMBOL(netmem_can_retain);
 
 static int page_pool_select_mp(struct page_pool *pool,
-			       const struct pp_memory_provider_params *mpp)
+			       const struct pp_memory_provider_params *mpp,
+			       u32 rxq_mp_gen)
 {
 	const struct memory_provider_ops *ops = mpp->mp_ops;
 	unsigned int flags = pool->slow.flags;
@@ -234,6 +235,7 @@ static int page_pool_select_mp(struct page_pool *pool,
 
 	pool->mp_priv = mpp->mp_priv;
 	pool->mp_ops = ops;
+	pool->mp_gen = rxq_mp_gen;
 	return 0;
 }
 
@@ -326,7 +328,7 @@ static int page_pool_init(struct page_pool *pool,
 		netdev_assert_locked(pool->slow.netdev);
 		rxq = __netif_get_rx_queue(pool->slow.netdev,
 					   pool->slow.queue_idx);
-		err = page_pool_select_mp(pool, &rxq->mp_params);
+		err = page_pool_select_mp(pool, &rxq->mp_params, rxq->mp_gen);
 		if (err)
 			goto free_ptr_ring;
 	}
diff --git a/net/core/page_pool_user.c b/net/core/page_pool_user.c
index f30f3779624c..b69da0f74756 100644
--- a/net/core/page_pool_user.c
+++ b/net/core/page_pool_user.c
@@ -398,28 +398,37 @@ void page_pool_unlist(struct page_pool *pool)
 	mutex_unlock(&page_pools_lock);
 }
 
+/* Check that the queue allocation that just ran put the provider in every
+ * pool role it needs: headers for a readable provider, payload otherwise.
+ * Pools of earlier allocations do not count.
+ */
 int page_pool_check_memory_provider(struct net_device *dev,
 				    struct netdev_rx_queue *rxq)
 {
-	void *binding = rxq->mp_params.mp_priv;
+	const struct memory_provider_ops *ops = rxq->mp_params.mp_ops;
+	void *mp_priv = rxq->mp_params.mp_priv;
+	unsigned int need, have = 0;
 	struct page_pool *pool;
 	struct hlist_node *n;
 
-	if (!binding)
+	if (!mp_priv)
 		return 0;
 
+	need = PP_RX_ROLE_PAYLOAD;
+	if (ops->caps & MP_CAP_READABLE)
+		need = PP_RX_ROLE_HEAD;
+
 	mutex_lock(&page_pools_lock);
 	hlist_for_each_entry_safe(pool, n, &dev->page_pools, user.list) {
-		if (pool->mp_priv != binding)
+		if (pool->mp_priv != mp_priv || pool->mp_gen != rxq->mp_gen ||
+		    pool->slow.queue_idx != get_netdev_rx_queue_index(rxq))
 			continue;
 
-		if (pool->slow.queue_idx == get_netdev_rx_queue_index(rxq)) {
-			mutex_unlock(&page_pools_lock);
-			return 0;
-		}
+		have |= pool->slow.rx_roles ? : PP_RX_ROLE_PAYLOAD;
 	}
 	mutex_unlock(&page_pools_lock);
-	return -ENODATA;
+
+	return (have & need) == need ? 0 : -ENODATA;
 }
 
 static void page_pool_unreg_netdev_wipe(struct net_device *netdev)
diff --git a/net/xdp/xsk.h b/net/xdp/xsk.h
index 4dab7b9a1487..9699e9a499a3 100644
--- a/net/xdp/xsk.h
+++ b/net/xdp/xsk.h
@@ -36,10 +36,13 @@ void xsk_map_try_sock_delete(struct xsk_map *map, struct xdp_sock *xs,
 			     struct xdp_sock __rcu **map_entry);
 struct xsk_pp;
 
-extern const struct memory_provider_ops xsk_pp_ops;
-
 struct xsk_pp *xsk_pp_create(struct xsk_buff_pool *pool);
 void xsk_pp_put(struct xsk_pp *pp);
+bool xsk_pp_supported(struct net_device *dev);
+int xsk_pp_bind(struct xsk_buff_pool *pool, struct net_device *dev,
+		u16 queue_id);
+void xsk_pp_unbind(struct xsk_buff_pool *pool);
+int xp_alloc_heads(struct xsk_buff_pool *pool);
 void xsk_clear_pool_at_qid(struct net_device *dev, u16 queue_id);
 int xsk_reg_pool_at_qid(struct net_device *dev, struct xsk_buff_pool *pool,
 			u16 queue_id);
diff --git a/net/xdp/xsk_buff_pool.c b/net/xdp/xsk_buff_pool.c
index 18966bbdb5b7..c620a0c32c17 100644
--- a/net/xdp/xsk_buff_pool.c
+++ b/net/xdp/xsk_buff_pool.c
@@ -38,6 +38,8 @@ void xp_destroy(struct xsk_buff_pool *pool)
 	if (!pool)
 		return;
 
+	if (pool->pp)
+		xsk_pp_put(pool->pp);
 	kvfree(pool->tx_descs);
 	kvfree(pool->heads);
 	kvfree(pool);
@@ -62,18 +64,13 @@ struct xsk_buff_pool *xp_create_and_assign_umem(struct xdp_sock *xs,
 {
 	bool unaligned = umem->flags & XDP_UMEM_UNALIGNED_CHUNK_FLAG;
 	struct xsk_buff_pool *pool;
-	struct xdp_buff_xsk *xskb;
-	u32 i, entries;
+	u32 entries;
 
 	entries = unaligned ? umem->chunks : 0;
 	pool = kvzalloc_flex(*pool, free_heads, entries);
 	if (!pool)
 		goto out;
 
-	pool->heads = kvzalloc_objs(*pool->heads, umem->chunks);
-	if (!pool->heads)
-		goto out;
-
 	if (xs->tx)
 		if (xp_alloc_tx_descs(pool, xs, max_segs))
 			goto out;
@@ -104,10 +101,30 @@ struct xsk_buff_pool *xp_create_and_assign_umem(struct xdp_sock *xs,
 	pool->fq = xs->fq_tmp;
 	pool->cq = xs->cq_tmp;
 
-	for (i = 0; i < pool->free_heads_cnt; i++) {
+	return pool;
+
+out:
+	xp_destroy(pool);
+	return NULL;
+}
+
+/* Only copy mode and driver zero copy use heads */
+int xp_alloc_heads(struct xsk_buff_pool *pool)
+{
+	struct xdp_buff_xsk *xskb;
+	u32 i;
+
+	if (pool->heads)
+		return 0;
+
+	pool->heads = kvzalloc_objs(*pool->heads, pool->heads_cnt);
+	if (!pool->heads)
+		return -ENOMEM;
+
+	for (i = 0; i < pool->heads_cnt; i++) {
 		xskb = &pool->heads[i];
 		xskb->pool = pool;
-		xskb->xdp.frame_sz = umem->chunk_size - umem->headroom;
+		xskb->xdp.frame_sz = pool->chunk_size - pool->headroom;
 		INIT_LIST_HEAD(&xskb->list_node);
 		if (pool->unaligned)
 			pool->free_heads[i] = xskb;
@@ -115,11 +132,7 @@ struct xsk_buff_pool *xp_create_and_assign_umem(struct xdp_sock *xs,
 			xp_init_xskb_addr(xskb, pool, (u64)i * pool->chunk_size);
 	}
 
-	return pool;
-
-out:
-	xp_destroy(pool);
-	return NULL;
+	return 0;
 }
 
 void xp_set_rxq_info(struct xsk_buff_pool *pool, struct xdp_rxq_info *rxq)
@@ -212,9 +225,24 @@ int xp_assign_dev(struct xsk_buff_pool *pool,
 	 */
 	hlist_add_head(&pool->dev_node, &netdev->xsk_pools);
 
-	if (force_copy)
+	if (force_copy) {
 		/* For copy-mode, we are done. */
+		err = xp_alloc_heads(pool);
+		goto out;
+	}
+
+	if (xsk_pp_supported(netdev)) {
+		err = xsk_pp_bind(pool, netdev, queue_id);
+		if (err)
+			goto err_unreg_pool;
+
+		pool->umem->zc = true;
 		return 0;
+	}
+
+	err = xp_alloc_heads(pool);
+	if (err)
+		goto out;
 
 	if ((netdev->xdp_features & NETDEV_XDP_ACT_XSK) != NETDEV_XDP_ACT_XSK) {
 		err = -EOPNOTSUPP;
@@ -267,8 +295,10 @@ int xp_assign_dev(struct xsk_buff_pool *pool,
 err_unreg_xsk:
 	xp_disable_drv_zc(pool);
 err_unreg_pool:
-	if (!force_zc)
-		err = 0; /* fallback to copy mode */
+	/* An old page pool may still own chunks of a failed provider. */
+	if (!force_zc && !pool->pp)
+		err = xp_alloc_heads(pool); /* fallback to copy mode */
+out:
 	if (err) {
 		pool->umem->flags = umem_flags;
 		xsk_clear_pool_at_qid(netdev, queue_id);
@@ -301,7 +331,10 @@ static void __xp_clear_dev(struct xsk_buff_pool *pool)
 	struct net_device *netdev = pool->netdev;
 
 	netdev_assert_locked_ops_compat(netdev);
-	xp_disable_drv_zc(pool);
+	if (pool->pp)
+		xsk_pp_unbind(pool);
+	else
+		xp_disable_drv_zc(pool);
 	xsk_clear_pool_at_qid(netdev, pool->queue_id);
 	hlist_del(&pool->dev_node);
 	pool->netdev = NULL;
@@ -474,7 +507,7 @@ static void xp_check_dma_contiguity(struct xsk_dma_map *dma_map)
 
 static int xp_init_dma_info(struct xsk_buff_pool *pool, struct xsk_dma_map *dma_map)
 {
-	if (!pool->unaligned) {
+	if (!pool->unaligned && pool->heads) {
 		u32 i;
 
 		for (i = 0; i < pool->heads_cnt; i++) {
diff --git a/net/xdp/xsk_pp.c b/net/xdp/xsk_pp.c
index fdd55265a6cd..c2214e718cdb 100644
--- a/net/xdp/xsk_pp.c
+++ b/net/xdp/xsk_pp.c
@@ -8,6 +8,9 @@
  *       +-----RX ring-------+
  */
 
+#include <net/netdev_lock.h>
+#include <net/netdev_queues.h>
+#include <net/netdev_rx_queue.h>
 #include <net/page_pool/helpers.h>
 #include <net/page_pool/memory_provider.h>
 #include <net/xdp_sock_drv.h>
@@ -15,6 +18,8 @@
 #include "xsk.h"
 #include "xsk_queue.h"
 
+#define XSK_PP_DMA_ATTR	(DMA_ATTR_WEAK_ORDERING | DMA_ATTR_SKIP_CPU_SYNC)
+
 enum xsk_pp_chunk {
 	XSK_PP_USER,	/* owned by userspace */
 	XSK_PP_POOL,	/* owned by a page pool */
@@ -30,6 +35,8 @@ struct xsk_pp {
 	refcount_t refs;
 	u32 nr_pools;
 	bool dma_need_sync;
+	/* RX queue of the provider, NULL once it is removed */
+	struct netdev_rx_queue *rxq;
 	u8 *state;
 	u32 reuse_cnt;
 	u32 reuse[];
@@ -236,11 +243,225 @@ static int xsk_pp_nl_fill(void *mp_priv, struct sk_buff *rsp,
 	return 0;
 }
 
+/* The queue no longer uses the provider, and its DMA stopped */
+static void xsk_pp_detach(struct xsk_pp *pp)
+{
+	pp->rxq = NULL;
+	spin_lock_bh(&pp->lock);
+	pp->fq = NULL;
+	spin_unlock_bh(&pp->lock);
+	xp_dma_unmap(pp->pool, XSK_PP_DMA_ATTR);
+}
+
+/* The device unregisters, and is closed */
+static void xsk_pp_uninstall(void *mp_priv, struct netdev_rx_queue *rxq)
+{
+	xsk_pp_detach(mp_priv);
+}
+
 const struct memory_provider_ops xsk_pp_ops = {
 	.init		= xsk_pp_init,
 	.destroy	= xsk_pp_destroy,
 	.alloc_netmems	= xsk_pp_alloc_netmems,
 	.release_netmem	= xsk_pp_release_netmem,
 	.nl_fill	= xsk_pp_nl_fill,
+	.uninstall	= xsk_pp_uninstall,
 	.caps		= MP_CAP_READABLE,
 };
+
+static bool xsk_pp_abandon(struct xsk_pp *pp)
+{
+	struct xsk_buff_pool *pool = pp->pool;
+	bool busy;
+	u32 idx;
+
+	spin_lock_bh(&pp->lock);
+	busy = pp->nr_pools;
+	while (!busy && pp->reuse_cnt) {
+		idx = pp->reuse[--pp->reuse_cnt];
+		pp->state[idx] = XSK_PP_USER;
+		xp_free(&pool->heads[idx]);
+	}
+	spin_unlock_bh(&pp->lock);
+
+	return !busy;
+}
+
+static int xsk_pp_validate(struct net_device *dev, struct xsk_buff_pool *pool,
+			   unsigned int mtu, struct bpf_prog *prog,
+			   struct netlink_ext_ack *extack)
+{
+	struct xsk_rx_check check = {
+		.buf_size = pool->chunk_size,
+		.headroom = xsk_pool_get_headroom(pool),
+		.mtu = mtu,
+		.sg = xsk_pool_uses_sg(pool),
+		.prog = prog,
+	};
+	struct netdev_bpf bpf = {
+		.command = XDP_VALIDATE_XSK_POOL,
+		.xsk_check.check = &check,
+		.xsk_check.extack = extack,
+	};
+
+	return dev->netdev_ops->ndo_bpf(dev, &bpf);
+}
+
+/**
+ * xsk_validate_dev - check AF_XDP page pool queues against a new setting
+ * @dev: device of the queues
+ * @mtu: MTU to check
+ * @prog: XDP program to check, or NULL for the current one
+ * @mode: XDP mode of @prog
+ * @extack: netlink extended ack
+ *
+ * Context: holds the instance lock of @dev, or RTNL for a device without it.
+ * Return: 0 if the driver accepts all queues of @dev that have an AF_XDP
+ * memory provider, or its error.
+ */
+int xsk_validate_dev(struct net_device *dev, unsigned int mtu,
+		     struct bpf_prog *prog, enum bpf_xdp_mode mode,
+		     struct netlink_ext_ack *extack)
+{
+	struct netdev_rx_queue *rxq;
+	struct xsk_pp *pp;
+	unsigned int i;
+	int err;
+
+	netdev_assert_locked_ops_compat(dev);
+
+	for (i = 0; i < dev->real_num_rx_queues; i++) {
+		rxq = __netif_get_rx_queue(dev, i);
+		if (rxq->mp_params.mp_ops != &xsk_pp_ops)
+			continue;
+
+		/* Generic XDP runs outside the queue NAPI */
+		if (mode == XDP_MODE_SKB) {
+			NL_SET_ERR_MSG(extack,
+				       "generic XDP cannot feed an AF_XDP page pool socket");
+			return -EBUSY;
+		}
+
+		pp = rxq->mp_params.mp_priv;
+		err = xsk_pp_validate(dev, pp->pool, mtu, prog, extack);
+		if (err)
+			return err;
+	}
+
+	return 0;
+}
+EXPORT_SYMBOL_GPL(xsk_validate_dev);
+
+/**
+ * xsk_rxq_pool - get the AF_XDP pool that provides memory to an RX queue
+ * @dev: device of the queue
+ * @queue_id: RX queue
+ *
+ * The pool stays valid while the queue that the caller allocates runs.
+ *
+ * Context: holds the instance lock of @dev.
+ * Return: the buffer pool, or NULL.
+ */
+struct xsk_buff_pool *xsk_rxq_pool(struct net_device *dev, u16 queue_id)
+{
+	struct netdev_rx_queue *rxq;
+	struct xsk_pp *pp;
+
+	netdev_assert_locked(dev);
+
+	if (queue_id >= dev->real_num_rx_queues)
+		return NULL;
+
+	rxq = __netif_get_rx_queue(dev, queue_id);
+	if (rxq->mp_params.mp_ops != &xsk_pp_ops)
+		return NULL;
+
+	pp = rxq->mp_params.mp_priv;
+	return pp->pool;
+}
+EXPORT_SYMBOL_GPL(xsk_rxq_pool);
+
+bool xsk_pp_supported(struct net_device *dev)
+{
+	const struct netdev_queue_mgmt_ops *qops = dev->queue_mgmt_ops;
+
+	return qops && qops->xsk_page_pool;
+}
+
+/* Called with RTNL and the instance lock of @dev. On error, @pool->pp stays
+ * set if an old page pool still owns chunks; copy mode must not take the pool.
+ */
+int xsk_pp_bind(struct xsk_buff_pool *pool, struct net_device *dev,
+		u16 queue_id)
+{
+	struct pp_memory_provider_params mp = { .mp_ops = &xsk_pp_ops };
+	struct device *dma_dev;
+	struct xsk_pp *pp;
+	int err;
+
+	if (queue_id >= dev->real_num_rx_queues || pool->unaligned)
+		return -EOPNOTSUPP;
+	if ((dev->xdp_features & NETDEV_XDP_ACT_XSK) != NETDEV_XDP_ACT_XSK ||
+	    (xsk_pool_uses_sg(pool) && dev->xdp_zc_max_segs == 1))
+		return -EOPNOTSUPP;
+	if (dev_xdp_prog_id(dev, XDP_MODE_SKB))
+		return -EOPNOTSUPP;
+	if (!netif_all_mps_have_cap(dev, MP_CAP_READABLE))
+		return -EBUSY;
+
+	err = xsk_pp_validate(dev, pool, dev->mtu, NULL, NULL);
+	if (err)
+		return err;
+
+	dma_dev = netdev_queue_get_dma_dev(dev, queue_id, NETDEV_QUEUE_TYPE_RX);
+	if (!dma_dev)
+		return -EOPNOTSUPP;
+
+	err = xp_dma_map(pool, dma_dev, XSK_PP_DMA_ATTR, pool->umem->pgs,
+			 pool->umem->npgs);
+	if (err)
+		return err;
+
+	pp = xsk_pp_create(pool);
+	if (IS_ERR(pp)) {
+		xp_dma_unmap(pool, XSK_PP_DMA_ATTR);
+		return PTR_ERR(pp);
+	}
+
+	mp.mp_priv = pp;
+	mp.rx_page_size = pool->chunk_size;
+	mp.rx_headroom = xsk_pool_get_headroom(pool);
+	err = netif_mp_open_rxq(dev, queue_id, &mp, NULL);
+	if (!err) {
+		pp->rxq = __netif_get_rx_queue(dev, queue_id);
+		pool->pp = pp;
+		pool->xdp_zc_max_segs = dev->xdp_zc_max_segs;
+		return 0;
+	}
+
+	xp_dma_unmap(pool, XSK_PP_DMA_ATTR);
+	/* Copy mode takes the chunks that the provider holds */
+	if (!xp_alloc_heads(pool) && xsk_pp_abandon(pp))
+		xsk_pp_put(pp);
+	else
+		pool->pp = pp;
+
+	return err;
+}
+
+/* Called with RTNL and the instance lock of the device */
+void xsk_pp_unbind(struct xsk_buff_pool *pool)
+{
+	struct xsk_pp *pp = pool->pp;
+	struct pp_memory_provider_params mp = {
+		.mp_ops = &xsk_pp_ops,
+		.mp_priv = pp,
+	};
+
+	/* Unregister removed the provider already */
+	if (!pp->rxq)
+		return;
+
+	netif_mp_close_rxq(pool->netdev, pool->queue_id, &mp);
+	xsk_pp_detach(pp);
+}
-- 
2.56.0

[PATCH net-next v1 06/11] page_pool: Coordinate refill with memory providers

[Björn Töpel]

A UMEM has a fixed number of chunks. When a refill finds none, and the
chunks come back later, nothing schedules the queue again: RX stalls.
Kicks after each queue operation are easy to miss.

Drivers report the end of each refill with page_pool_refill_done().
The provider then asks for another poll, or wakes the queue later
through the NAPI that the core publishes while the queue runs. Socket
wakeups use that NAPI too, so such drivers need no ndo_xsk_wakeup.

With XDP_USE_NEED_WAKEUP, the FILL flag stays set: a flag set only
when starved can miss an entry posted at the same time.

Signed-off-by: Björn Töpel <bjorn@kernel.org>
---
 Documentation/networking/netmem.rst     | 10 +++
 include/net/netdev_rx_queue.h           |  6 ++
 include/net/page_pool/helpers.h         | 24 ++++++
 include/net/page_pool/memory_provider.h |  2 +
 include/net/page_pool/types.h           |  6 ++
 net/core/dev.c                          |  4 +
 net/core/dev.h                          |  4 +
 net/core/netdev_rx_queue.c              | 97 ++++++++++++++++++++++++-
 net/xdp/xsk.c                           |  4 +
 net/xdp/xsk.h                           |  1 +
 net/xdp/xsk_pp.c                        | 73 ++++++++++++++++++-
 11 files changed, 226 insertions(+), 5 deletions(-)

diff --git a/Documentation/networking/netmem.rst b/Documentation/networking/netmem.rst
index 39cb75339704..74c73d7d126e 100644
--- a/Documentation/networking/netmem.rst
+++ b/Documentation/networking/netmem.rst
@@ -136,6 +136,16 @@ Driver RX Requirements
    circulation times, such as when userspace holds a reference in zerocopy
    scenarios.
 
+8. After each RX ring refill, call page_pool_refill_done() once per page
+   pool, with PP_REFILL_FULL only if all rings that use the pool are full.
+   Rings of a queue that take buffers from one such provider share one
+   page pool. Keep buffers in each of them, first in a ring that every
+   packet needs, like a header ring, so that a provider with few buffers
+   does not leave them all in one ring. If the call returns false, poll
+   again soon, for example by returning the full NAPI budget. Register the
+   NAPI of each RX queue with netif_queue_set_napi(); the core schedules
+   it when provider buffers become available.
+
 Driver TX Requirements
 ======================
 
diff --git a/include/net/netdev_rx_queue.h b/include/net/netdev_rx_queue.h
index 989120cbf452..ca12cbf7b825 100644
--- a/include/net/netdev_rx_queue.h
+++ b/include/net/netdev_rx_queue.h
@@ -33,6 +33,12 @@ struct netdev_rx_queue {
 	struct pp_memory_provider_params mp_params;
 	/* stamps the page pools of each queue allocation */
 	u32				mp_gen;
+	/* NAPI of the running queue, for memory provider wakeups. Set while
+	 * mp_armed, and if the provider has a refill_done op. Written under
+	 * the ops lock, read under RCU, see netif_mp_wake().
+	 */
+	struct napi_struct __rcu	*mp_napi;
+	bool				mp_armed;
 
 	/* If a queue is leased, then the lease pointer is always
 	 * valid. From the physical device it points to the virtual
diff --git a/include/net/page_pool/helpers.h b/include/net/page_pool/helpers.h
index 96e6b62cdebb..245a904e0b2b 100644
--- a/include/net/page_pool/helpers.h
+++ b/include/net/page_pool/helpers.h
@@ -523,6 +523,30 @@ static inline bool page_pool_is_unreadable(struct page_pool *pool)
 	return pool->mp_ops && !(pool->mp_ops->caps & MP_CAP_READABLE);
 }
 
+/**
+ * page_pool_refill_done() - report the end of an RX ring refill
+ * @pool: page pool of the rings
+ * @state: whether all rings that take buffers from @pool are full
+ *
+ * A memory provider with a fixed set of buffers must know when the rings
+ * are full, and when they wait for buffers. Call this once per page pool
+ * after each refill, also when the rings were full already.
+ *
+ * Context: the NAPI poll that allocates from @pool, or the driver while that
+ * NAPI is disabled, with IRQs enabled.
+ * Return: true if the refill is done, false if NAPI should retry it soon.
+ */
+static inline bool page_pool_refill_done(struct page_pool *pool,
+					 enum pp_refill state)
+{
+	const struct memory_provider_ops *ops = pool->mp_ops;
+
+	if (ops && ops->refill_done)
+		return ops->refill_done(pool, state);
+
+	return true;
+}
+
 /**
  * page_pool_supports_frag() - may page pool buffers be split
  * @pool: queried page pool
diff --git a/include/net/page_pool/memory_provider.h b/include/net/page_pool/memory_provider.h
index e732b21be9a2..daec73c9dcee 100644
--- a/include/net/page_pool/memory_provider.h
+++ b/include/net/page_pool/memory_provider.h
@@ -42,6 +42,7 @@ struct memory_provider_ops {
 	int (*nl_fill)(void *mp_priv, struct sk_buff *rsp,
 		       struct netdev_rx_queue *rxq);
 	void (*uninstall)(void *mp_priv, struct netdev_rx_queue *rxq);
+	bool (*refill_done)(struct page_pool *pool, enum pp_refill state);
 	u32 caps;
 };
 
@@ -55,6 +56,7 @@ int netif_mp_open_rxq(struct net_device *dev, unsigned int rxq_idx,
 int netif_mp_close_rxq(struct net_device *dev, unsigned int rxq_idx,
 		       const struct pp_memory_provider_params *old_p);
 bool netif_all_mps_have_cap(struct net_device *dev, u32 cap);
+bool netif_mp_wake(struct netdev_rx_queue *rxq);
 
 /**
   * net_mp_netmem_place_in_cache() - give a netmem to a page pool
diff --git a/include/net/page_pool/types.h b/include/net/page_pool/types.h
index e9f3af91e305..58ca414fa3d2 100644
--- a/include/net/page_pool/types.h
+++ b/include/net/page_pool/types.h
@@ -40,6 +40,12 @@
  */
 #define PP_FLAG_MEM_PROVIDER	BIT(4)
 
+/* Ring state after a refill, see page_pool_refill_done() */
+enum pp_refill {
+	PP_REFILL_PARTIAL,	/* some ring of the pool has room left */
+	PP_REFILL_FULL,		/* all rings of the pool are full */
+};
+
 /* RX buffers a page pool supplies, see page_pool_params.rx_roles */
 enum pp_rx_role {
 	PP_RX_ROLE_HEAD		= BIT(0),	/* packet headers */
diff --git a/net/core/dev.c b/net/core/dev.c
index 7e985fb82488..6033fc61a666 100644
--- a/net/core/dev.c
+++ b/net/core/dev.c
@@ -1771,6 +1771,7 @@ static int __dev_open(struct net_device *dev, struct netlink_ext_ack *extack)
 		dev_set_rx_mode(dev);
 		dev_activate(dev);
 		add_device_randomness(dev->dev_addr, dev->addr_len);
+		netif_mp_arm_dev(dev);
 	}
 
 	return ret;
@@ -1833,6 +1834,7 @@ static void __dev_close_many(struct list_head *head)
 
 		netdev_assert_locked_ops_compat(dev);
 
+		netif_mp_disarm_dev(dev);
 		if (ops->ndo_stop)
 			ops->ndo_stop(dev);
 
@@ -7559,6 +7561,7 @@ void netif_queue_set_napi(struct net_device *dev, unsigned int queue_index,
 	case NETDEV_QUEUE_TYPE_RX:
 		rxq = __netif_get_rx_queue(dev, queue_index);
 		rxq->napi = napi;
+		netif_mp_update_napi(rxq);
 		return;
 	case NETDEV_QUEUE_TYPE_TX:
 		txq = netdev_get_tx_queue(dev, queue_index);
@@ -7960,6 +7963,7 @@ void __netif_napi_del_locked(struct napi_struct *napi)
 	}
 
 	list_del_rcu(&napi->dev_list);
+	netif_mp_napi_del(napi);
 	napi_free_frags(napi);
 
 	gro_cleanup(&napi->gro);
diff --git a/net/core/dev.h b/net/core/dev.h
index e3337013ffc8..56386fd8c767 100644
--- a/net/core/dev.h
+++ b/net/core/dev.h
@@ -125,6 +125,10 @@ bool netif_is_queue_leasee(const struct net_device *dev);
 
 void __netif_mp_uninstall_rxq(struct netdev_rx_queue *rxq,
 			      const struct pp_memory_provider_params *p);
+void netif_mp_update_napi(struct netdev_rx_queue *rxq);
+void netif_mp_arm_dev(struct net_device *dev);
+void netif_mp_disarm_dev(struct net_device *dev);
+void netif_mp_napi_del(struct napi_struct *napi);
 
 void netif_rxq_cleanup_unlease(struct netdev_rx_queue *phys_rxq,
 			       struct netdev_rx_queue *virt_rxq);
diff --git a/net/core/netdev_rx_queue.c b/net/core/netdev_rx_queue.c
index 4e2f906fbc73..3fe904b56791 100644
--- a/net/core/netdev_rx_queue.c
+++ b/net/core/netdev_rx_queue.c
@@ -116,6 +116,94 @@ bool netif_all_mps_have_cap(struct net_device *dev, u32 cap)
 	return true;
 }
 
+/**
+ * netif_mp_wake - schedule the NAPI that refills a memory provider queue
+ * @rxq: physical RX queue of the provider
+ *
+ * Context: IRQs enabled, not in hardirq. The caller keeps the device of
+ * @rxq alive.
+ *
+ * Return: false if no NAPI refills the queue.
+ */
+bool netif_mp_wake(struct netdev_rx_queue *rxq)
+{
+	struct napi_struct *napi;
+	bool running = false;
+
+	rcu_read_lock();
+	napi = rcu_dereference(rxq->mp_napi);
+	if (napi) {
+		local_bh_disable();
+		napi_schedule(napi);
+		local_bh_enable();
+		running = true;
+	}
+	rcu_read_unlock();
+
+	return running;
+}
+
+/* Point wakeups at the NAPI of the running queue, and schedule it once: a
+ * refill during the start may have found no buffers.
+ */
+void netif_mp_update_napi(struct netdev_rx_queue *rxq)
+{
+	const struct memory_provider_ops *ops = rxq->mp_params.mp_ops;
+	struct napi_struct *napi = NULL;
+
+	if (rxq->mp_armed && ops && ops->refill_done)
+		napi = rxq->napi;
+	if (rcu_access_pointer(rxq->mp_napi) == napi)
+		return;
+
+	rcu_assign_pointer(rxq->mp_napi, napi);
+	if (napi)
+		netif_mp_wake(rxq);
+}
+
+static void netif_mp_arm_napi(struct netdev_rx_queue *rxq)
+{
+	rxq->mp_armed = true;
+	netif_mp_update_napi(rxq);
+}
+
+static void netif_mp_disarm_napi(struct netdev_rx_queue *rxq)
+{
+	rxq->mp_armed = false;
+	RCU_INIT_POINTER(rxq->mp_napi, NULL);
+}
+
+void netif_mp_arm_dev(struct net_device *dev)
+{
+	unsigned int i;
+
+	for (i = 0; i < dev->real_num_rx_queues; i++)
+		netif_mp_arm_napi(__netif_get_rx_queue(dev, i));
+}
+
+void netif_mp_disarm_dev(struct net_device *dev)
+{
+	unsigned int i;
+
+	for (i = 0; i < dev->num_rx_queues; i++)
+		netif_mp_disarm_napi(&dev->_rx[i]);
+}
+
+/* The grace period of netif_napi_del() covers wakeups that found @napi. */
+void netif_mp_napi_del(struct napi_struct *napi)
+{
+	struct net_device *dev = napi->dev;
+	unsigned int i;
+
+	/* free_netdev() frees the queues before it deletes the last NAPIs */
+	if (dev->reg_state != NETREG_REGISTERED)
+		return;
+
+	for (i = 0; i < dev->num_rx_queues; i++)
+		if (rcu_access_pointer(dev->_rx[i].mp_napi) == napi)
+			RCU_INIT_POINTER(dev->_rx[i].mp_napi, NULL);
+}
+
 static int netdev_rx_queue_reconfig(struct net_device *dev,
 				    unsigned int rxq_idx,
 				    struct netdev_queue_config *qcfg_old,
@@ -152,13 +240,15 @@ static int netdev_rx_queue_reconfig(struct net_device *dev,
 		goto err_free_new_queue_mem;
 
 	if (netif_running(dev)) {
+		netif_mp_disarm_napi(rxq);
 		err = qops->ndo_queue_stop(dev, old_mem, rxq_idx);
 		if (err)
-			goto err_free_new_queue_mem;
+			goto err_arm_old_queue;
 
 		err = qops->ndo_queue_start(dev, qcfg_new, new_mem, rxq_idx);
 		if (err)
 			goto err_start_queue;
+		netif_mp_arm_napi(rxq);
 	} else {
 		swap(new_mem, old_mem);
 	}
@@ -183,8 +273,13 @@ static int netdev_rx_queue_reconfig(struct net_device *dev,
 		     "Failed to restart old queue in error path. RX queue %d may be unhealthy.",
 		     rxq_idx);
 		qops->ndo_queue_mem_free(dev, old_mem);
+		goto err_free_new_queue_mem;
 	}
 
+err_arm_old_queue:
+	/* The old queue may run */
+	netif_mp_arm_napi(rxq);
+
 err_free_new_queue_mem:
 	qops->ndo_queue_mem_free(dev, new_mem);
 
diff --git a/net/xdp/xsk.c b/net/xdp/xsk.c
index 781ba46eb152..113b0b08356d 100644
--- a/net/xdp/xsk.c
+++ b/net/xdp/xsk.c
@@ -794,6 +794,10 @@ static int xsk_wakeup(struct xdp_sock *xs, u8 flags)
 {
 	struct net_device *dev = xs->dev;
 
+	/* The core wakes the queue for both RX and TX. */
+	if (xs->pool->pp)
+		return xsk_pp_wakeup(xs->pool->pp);
+
 	return dev->netdev_ops->ndo_xsk_wakeup(dev, xs->queue_id, flags);
 }
 
diff --git a/net/xdp/xsk.h b/net/xdp/xsk.h
index 9699e9a499a3..d7d6bcf0a893 100644
--- a/net/xdp/xsk.h
+++ b/net/xdp/xsk.h
@@ -42,6 +42,7 @@ bool xsk_pp_supported(struct net_device *dev);
 int xsk_pp_bind(struct xsk_buff_pool *pool, struct net_device *dev,
 		u16 queue_id);
 void xsk_pp_unbind(struct xsk_buff_pool *pool);
+int xsk_pp_wakeup(struct xsk_pp *pp);
 int xp_alloc_heads(struct xsk_buff_pool *pool);
 void xsk_clear_pool_at_qid(struct net_device *dev, u16 queue_id);
 int xsk_reg_pool_at_qid(struct net_device *dev, struct xsk_buff_pool *pool,
diff --git a/net/xdp/xsk_pp.c b/net/xdp/xsk_pp.c
index c2214e718cdb..a77213f7fef6 100644
--- a/net/xdp/xsk_pp.c
+++ b/net/xdp/xsk_pp.c
@@ -36,7 +36,8 @@ struct xsk_pp {
 	u32 nr_pools;
 	bool dma_need_sync;
 	/* RX queue of the provider, NULL once it is removed */
-	struct netdev_rx_queue *rxq;
+	struct netdev_rx_queue __rcu *rxq;
+	bool starved;
 	u8 *state;
 	u32 reuse_cnt;
 	u32 reuse[];
@@ -219,11 +220,31 @@ static netmem_ref xsk_pp_alloc_netmems(struct page_pool *pool, gfp_t gfp)
 	return netmems[n - 1];
 }
 
+static bool xsk_pp_wake(struct xsk_pp *pp)
+{
+	struct netdev_rx_queue *rxq;
+	bool running = false;
+
+	rcu_read_lock();
+	rxq = rcu_dereference(pp->rxq);
+	if (rxq)
+		running = netif_mp_wake(rxq);
+	rcu_read_unlock();
+
+	return running;
+}
+
+int xsk_pp_wakeup(struct xsk_pp *pp)
+{
+	return xsk_pp_wake(pp) ? 0 : -ENETDOWN;
+}
+
 static bool xsk_pp_release_netmem(struct page_pool *pool, netmem_ref netmem)
 {
 	struct net_iov *niov = netmem_to_net_iov(netmem);
 	struct xsk_pp *pp = pool->mp_priv;
 	u32 idx = net_iov_idx(niov);
+	bool starved;
 
 	spin_lock_bh(&pp->lock);
 	net_mp_niov_clear_page_pool(niov);
@@ -232,11 +253,47 @@ static bool xsk_pp_release_netmem(struct page_pool *pool, netmem_ref netmem)
 		pp->state[idx] = XSK_PP_REUSE;
 		pp->reuse[pp->reuse_cnt++] = idx;
 	}
+	starved = pp->starved;
+	WRITE_ONCE(pp->starved, false);
 	spin_unlock_bh(&pp->lock);
 
+	/* A starved queue may wait for this release */
+	if (starved)
+		xsk_pp_wake(pp);
+
 	return false;
 }
 
+static bool xsk_pp_refill_done(struct page_pool *pool, enum pp_refill state)
+{
+	struct xsk_pp *pp = pool->mp_priv;
+	struct xsk_buff_pool *xp = pp->pool;
+	bool wait = xp->uses_need_wakeup;
+	bool more;
+
+	if (state == PP_REFILL_FULL && !READ_ONCE(pp->starved))
+		return true;
+
+	spin_lock_bh(&pp->lock);
+	if (!pp->fq) {
+		spin_unlock_bh(&pp->lock);
+		return true;
+	}
+
+	if (state == PP_REFILL_FULL) {
+		WRITE_ONCE(pp->starved, false);
+		spin_unlock_bh(&pp->lock);
+		return true;
+	}
+
+	/* With need_wakeup, userspace wakes the queue after it posts FILL */
+	more = pp->reuse_cnt || xskq_cons_nb_entries(pp->fq, 1);
+	WRITE_ONCE(pp->starved, !more);
+	spin_unlock_bh(&pp->lock);
+
+	return !more && wait;
+}
+
 static int xsk_pp_nl_fill(void *mp_priv, struct sk_buff *rsp,
 			  struct netdev_rx_queue *rxq)
 {
@@ -246,10 +303,12 @@ static int xsk_pp_nl_fill(void *mp_priv, struct sk_buff *rsp,
 /* The queue no longer uses the provider, and its DMA stopped */
 static void xsk_pp_detach(struct xsk_pp *pp)
 {
-	pp->rxq = NULL;
+	RCU_INIT_POINTER(pp->rxq, NULL);
 	spin_lock_bh(&pp->lock);
 	pp->fq = NULL;
 	spin_unlock_bh(&pp->lock);
+	/* Wait for wakeups that found the queue */
+	synchronize_net();
 	xp_dma_unmap(pp->pool, XSK_PP_DMA_ATTR);
 }
 
@@ -266,6 +325,7 @@ const struct memory_provider_ops xsk_pp_ops = {
 	.release_netmem	= xsk_pp_release_netmem,
 	.nl_fill	= xsk_pp_nl_fill,
 	.uninstall	= xsk_pp_uninstall,
+	.refill_done	= xsk_pp_refill_done,
 	.caps		= MP_CAP_READABLE,
 };
 
@@ -428,17 +488,22 @@ int xsk_pp_bind(struct xsk_buff_pool *pool, struct net_device *dev,
 		return PTR_ERR(pp);
 	}
 
+	rcu_assign_pointer(pp->rxq, __netif_get_rx_queue(dev, queue_id));
 	mp.mp_priv = pp;
 	mp.rx_page_size = pool->chunk_size;
 	mp.rx_headroom = xsk_pool_get_headroom(pool);
 	err = netif_mp_open_rxq(dev, queue_id, &mp, NULL);
 	if (!err) {
-		pp->rxq = __netif_get_rx_queue(dev, queue_id);
 		pool->pp = pp;
 		pool->xdp_zc_max_segs = dev->xdp_zc_max_segs;
+		/* See xsk_pp_refill_done() */
+		if (pool->uses_need_wakeup)
+			xsk_set_rx_need_wakeup(pool);
 		return 0;
 	}
 
+	RCU_INIT_POINTER(pp->rxq, NULL);
+	synchronize_net();
 	xp_dma_unmap(pool, XSK_PP_DMA_ATTR);
 	/* Copy mode takes the chunks that the provider holds */
 	if (!xp_alloc_heads(pool) && xsk_pp_abandon(pp))
@@ -459,7 +524,7 @@ void xsk_pp_unbind(struct xsk_buff_pool *pool)
 	};
 
 	/* Unregister removed the provider already */
-	if (!pp->rxq)
+	if (!rcu_access_pointer(pp->rxq))
 		return;
 
 	netif_mp_close_rxq(pool->netdev, pool->queue_id, &mp);
-- 
2.56.0

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

[Björn Töpel]

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);
 }
 
@@ -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;
+	}
 	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);
+	}
 	if (!err)
 		xdp_return_buff(xdp);
 	return err;
diff --git a/net/xdp/xsk.h b/net/xdp/xsk.h
index d7d6bcf0a893..a5f52ef382a6 100644
--- a/net/xdp/xsk.h
+++ b/net/xdp/xsk.h
@@ -43,6 +43,11 @@ int xsk_pp_bind(struct xsk_buff_pool *pool, struct net_device *dev,
 		u16 queue_id);
 void xsk_pp_unbind(struct xsk_buff_pool *pool);
 int xsk_pp_wakeup(struct xsk_pp *pp);
+void xsk_pp_flush(struct xsk_pp *pp);
+bool xsk_pp_queue_buf(struct xsk_pp *pp, struct xdp_buff *xdp);
+int xsk_pp_rcv_zc(struct xsk_pp *pp, struct xdp_sock *xs,
+		  struct xdp_buff *xdp);
+bool xsk_pp_claim_copy(struct xsk_pp *pp, u64 *addrs, u32 n);
 int xp_alloc_heads(struct xsk_buff_pool *pool);
 void xsk_clear_pool_at_qid(struct net_device *dev, u16 queue_id);
 int xsk_reg_pool_at_qid(struct net_device *dev, struct xsk_buff_pool *pool,
diff --git a/net/xdp/xsk_pp.c b/net/xdp/xsk_pp.c
index a77213f7fef6..a1c2ca00f016 100644
--- a/net/xdp/xsk_pp.c
+++ b/net/xdp/xsk_pp.c
@@ -19,11 +19,13 @@
 #include "xsk_queue.h"
 
 #define XSK_PP_DMA_ATTR	(DMA_ATTR_WEAK_ORDERING | DMA_ATTR_SKIP_CPU_SYNC)
+#define XSK_PP_BATCH	64
 
 enum xsk_pp_chunk {
 	XSK_PP_USER,	/* owned by userspace */
 	XSK_PP_POOL,	/* owned by a page pool */
 	XSK_PP_REUSE,	/* on the reuse stack */
+	XSK_PP_COPY,	/* copied into, not yet on the RX ring */
 };
 
 struct xsk_pp {
@@ -38,6 +40,9 @@ struct xsk_pp {
 	/* RX queue of the provider, NULL once it is removed */
 	struct netdev_rx_queue __rcu *rxq;
 	bool starved;
+	/* Chunks of unpublished RX descriptors, NAPI only */
+	u32 pending_cnt;
+	netmem_ref pending[XSK_PP_BATCH];
 	u8 *state;
 	u32 reuse_cnt;
 	u32 reuse[];
@@ -294,6 +299,166 @@ static bool xsk_pp_refill_done(struct page_pool *pool, enum pp_refill state)
 	return !more && wait;
 }
 
+static u32 xsk_pp_chunk_idx(netmem_ref netmem)
+{
+	return net_iov_idx(netmem_to_net_iov(netmem));
+}
+
+/* Called from the queue NAPI, before the RX ring publishes the descriptors */
+void xsk_pp_flush(struct xsk_pp *pp)
+{
+	struct page_pool *pool;
+	u32 i, j, n = 0;
+	u32 idx;
+
+	if (!pp->pending_cnt)
+		return;
+
+	spin_lock_bh(&pp->lock);
+	for (i = 0; i < pp->pending_cnt; i++) {
+		idx = xsk_pp_chunk_idx(pp->pending[i]);
+		if (pp->state[idx] == XSK_PP_COPY) {
+			pp->state[idx] = XSK_PP_USER;
+			continue;
+		}
+
+		DEBUG_NET_WARN_ON_ONCE(pp->state[idx] != XSK_PP_POOL);
+		pp->pending[n++] = pp->pending[i];
+	}
+
+	/* A queue restart may change the page pool within a batch. */
+	for (i = 0; i < n; i = j) {
+		pool = netmem_get_pp(pp->pending[i]);
+		for (j = i + 1; j < n; j++)
+			if (netmem_get_pp(pp->pending[j]) != pool)
+				break;
+
+		net_mp_release_page_pool_bulk(pool, pp->pending + i, j - i);
+		while (i < j) {
+			idx = xsk_pp_chunk_idx(pp->pending[i++]);
+			pp->state[idx] = XSK_PP_USER;
+		}
+	}
+	pp->pending_cnt = 0;
+	spin_unlock_bh(&pp->lock);
+}
+
+bool xsk_pp_queue_buf(struct xsk_pp *pp, struct xdp_buff *xdp)
+{
+	netmem_ref head = xdp_buff_head_netmem(xdp);
+	struct netdev_rx_queue *rxq;
+	struct page_pool *pool;
+	bool ours = false;
+
+	/* An XDP test run for the queue brings buffers of its own */
+	if (!netmem_is_net_iov(head))
+		return false;
+
+	pool = netmem_get_pp(head);
+	rcu_read_lock();
+	rxq = rcu_dereference(pp->rxq);
+	if (pool && rxq)
+		ours = pool->slow.netdev == rxq->dev &&
+		       pool->slow.queue_idx == get_netdev_rx_queue_index(rxq);
+	rcu_read_unlock();
+
+	return ours;
+}
+
+static bool xsk_pp_owns(struct xsk_pp *pp, netmem_ref netmem)
+{
+	struct page_pool *pool;
+
+	if (!netmem_is_net_iov(netmem))
+		return false;
+
+	pool = netmem_get_pp(netmem);
+	return pool && pool->mp_priv == pp;
+}
+
+/* Returns -EXDEV for a packet that must be copied. Called from the NAPI. */
+int xsk_pp_rcv_zc(struct xsk_pp *pp, struct xdp_sock *xs,
+		  struct xdp_buff *xdp)
+{
+	struct skb_shared_info *sinfo = NULL;
+	void *umem = xs->pool->addrs;
+	u32 i, nr_frags = 0, num_desc;
+	netmem_ref head;
+	skb_frag_t *frag;
+
+	head = xdp_buff_head_netmem(xdp);
+	if (!xsk_pp_owns(pp, head))
+		return -EXDEV;
+
+	if (unlikely(xdp_buff_has_frags(xdp))) {
+		if (!xs->sg)
+			return -EXDEV;
+
+		sinfo = xdp_get_shared_info_from_buff(xdp);
+		nr_frags = sinfo->nr_frags;
+		for (i = 0; i < nr_frags; i++)
+			if (!xsk_pp_owns(pp, skb_frag_netmem(&sinfo->frags[i])))
+				return -EXDEV;
+	}
+
+	num_desc = nr_frags + 1;
+	if (xskq_prod_nb_free(xs->rx, num_desc) < num_desc) {
+		xs->rx_queue_full++;
+		return -ENOBUFS;
+	}
+	if (pp->pending_cnt + num_desc > XSK_PP_BATCH)
+		xsk_pp_flush(pp);
+
+	__xskq_prod_reserve_desc(xs->rx, xdp->data - umem,
+				 xdp->data_end - xdp->data,
+				 nr_frags ? XDP_PKT_CONTD : 0);
+	pp->pending[pp->pending_cnt++] = head;
+
+	for (i = 0; i < nr_frags; i++) {
+		frag = &sinfo->frags[i];
+		__xskq_prod_reserve_desc(xs->rx, skb_frag_address(frag) - umem,
+					 skb_frag_size(frag),
+					 i + 1 < nr_frags ? XDP_PKT_CONTD : 0);
+		pp->pending[pp->pending_cnt++] = skb_frag_netmem(frag);
+	}
+
+	return 0;
+}
+
+/* The chunks go to userspace with the next xsk_pp_flush(). Called from the
+ * NAPI. On failure the chunks taken from FILL stay with the provider.
+ */
+bool xsk_pp_claim_copy(struct xsk_pp *pp, u64 *addrs, u32 n)
+{
+	netmem_ref *netmems;
+	u32 i, got, idx;
+
+	BUILD_BUG_ON(MAX_SKB_FRAGS + 1 > XSK_PP_BATCH);
+	if (pp->pending_cnt + n > XSK_PP_BATCH)
+		xsk_pp_flush(pp);
+
+	netmems = pp->pending + pp->pending_cnt;
+	spin_lock_bh(&pp->lock);
+	got = xsk_pp_alloc_reuse(pp, netmems, n);
+	got += xsk_pp_alloc_fill(pp, netmems + got, n - got);
+	for (i = 0; i < got; i++) {
+		idx = xsk_pp_chunk_idx(netmems[i]);
+		if (got < n) {
+			pp->state[idx] = XSK_PP_REUSE;
+			pp->reuse[pp->reuse_cnt++] = idx;
+			continue;
+		}
+
+		pp->state[idx] = XSK_PP_COPY;
+		addrs[i] = (u64)idx << pp->area.niov_shift;
+	}
+	if (got == n)
+		pp->pending_cnt += n;
+	spin_unlock_bh(&pp->lock);
+
+	return got == n;
+}
+
 static int xsk_pp_nl_fill(void *mp_priv, struct sk_buff *rsp,
 			  struct netdev_rx_queue *rxq)
 {
diff --git a/net/xdp/xsk_queue.h b/net/xdp/xsk_queue.h
index 1bc42c8902f4..b9464b917a88 100644
--- a/net/xdp/xsk_queue.h
+++ b/net/xdp/xsk_queue.h
@@ -156,17 +156,24 @@ static inline bool xp_unused_options_set(u32 options)
 	return options & ~(XDP_PKT_CONTD | XDP_TX_METADATA);
 }
 
+static inline u32 xp_desc_meta_len(struct xsk_buff_pool *pool,
+				   struct xdp_desc *desc)
+{
+	return desc->options & XDP_TX_METADATA ? pool->tx_metadata_len : 0;
+}
+
 static inline bool xp_aligned_validate_desc(struct xsk_buff_pool *pool,
 					    struct xdp_desc *desc)
 {
+	u32 meta = xp_desc_meta_len(pool, desc);
 	u64 len = desc->len;
 	u64 addr, offset;
 
 	if (!len)
 		return false;
 
-	/* Can overflow if desc->addr < pool->tx_metadata_len */
-	if (check_sub_overflow(desc->addr, pool->tx_metadata_len, &addr))
+	/* Can overflow if desc->addr < meta */
+	if (check_sub_overflow(desc->addr, meta, &addr))
 		return false;
 
 	offset = addr & (pool->chunk_size - 1);
@@ -176,7 +183,7 @@ static inline bool xp_aligned_validate_desc(struct xsk_buff_pool *pool,
 	 * (pool->chunk_size is ``u32``), @len is guaranteed
 	 * to be <= ``U32_MAX``.
 	 */
-	if (offset + len + pool->tx_metadata_len > pool->chunk_size)
+	if (offset + len + meta > pool->chunk_size)
 		return false;
 
 	if (addr >= pool->addrs_cnt)
@@ -191,6 +198,7 @@ static inline bool xp_aligned_validate_desc(struct xsk_buff_pool *pool,
 static inline bool xp_unaligned_validate_desc(struct xsk_buff_pool *pool,
 					      struct xdp_desc *desc)
 {
+	u32 meta = xp_desc_meta_len(pool, desc);
 	u64 len = desc->len;
 	u64 addr, end;
 
@@ -198,13 +206,13 @@ static inline bool xp_unaligned_validate_desc(struct xsk_buff_pool *pool,
 		return false;
 
 	/* Can't overflow: @len is guaranteed to be <= ``U32_MAX`` */
-	len += pool->tx_metadata_len;
+	len += meta;
 	if (len > pool->chunk_size)
 		return false;
 
 	/* Can overflow if desc->addr is close to 0 */
 	if (check_sub_overflow(xp_unaligned_add_offset_to_addr(desc->addr),
-			       pool->tx_metadata_len, &addr))
+			       meta, &addr))
 		return false;
 
 	if (addr >= pool->addrs_cnt)
-- 
2.56.0

[PATCH net-next v1 08/11] eth: fbnic: Support XDP redirect

[Björn Töpel]

fbnic treats XDP_REDIRECT as an invalid action and drops the packet,
so AF_XDP, devmap and cpumap do not work. A failed redirect counts as
received, as XDP_DROP does, and as dropped.

Signed-off-by: Björn Töpel <bjorn@kernel.org>
---
 .../net/ethernet/meta/fbnic/fbnic_netdev.c    |  4 +++-
 drivers/net/ethernet/meta/fbnic/fbnic_txrx.c  | 19 +++++++++++++++++++
 2 files changed, 22 insertions(+), 1 deletion(-)

diff --git a/drivers/net/ethernet/meta/fbnic/fbnic_netdev.c b/drivers/net/ethernet/meta/fbnic/fbnic_netdev.c
index 8a6703afca38..ae2585508690 100644
--- a/drivers/net/ethernet/meta/fbnic/fbnic_netdev.c
+++ b/drivers/net/ethernet/meta/fbnic/fbnic_netdev.c
@@ -822,7 +822,9 @@ struct net_device *fbnic_netdev_alloc(struct fbnic_dev *fbd)
 	netdev->hw_enc_features |= netdev->features;
 	netdev->features |= NETIF_F_NTUPLE;
 
-	netdev->xdp_features = NETDEV_XDP_ACT_BASIC | NETDEV_XDP_ACT_RX_SG;
+	netdev->xdp_features = NETDEV_XDP_ACT_BASIC |
+			       NETDEV_XDP_ACT_REDIRECT |
+			       NETDEV_XDP_ACT_RX_SG;
 
 	netdev->min_mtu = IPV6_MIN_MTU;
 	netdev->max_mtu = FBNIC_MAX_JUMBO_FRAME_SIZE - ETH_HLEN;
diff --git a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c
index 0bcee152d47c..51d573c9c5b8 100644
--- a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c
+++ b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c
@@ -22,6 +22,8 @@ enum {
 	FBNIC_XDP_PASS = 0,
 	FBNIC_XDP_CONSUME,
 	FBNIC_XDP_TX,
+	FBNIC_XDP_REDIRECT,
+	FBNIC_XDP_REDIRECT_ERR,
 	FBNIC_XDP_LEN_ERR,
 };
 
@@ -1211,6 +1213,10 @@ static struct sk_buff *fbnic_run_xdp(struct fbnic_napi_vector *nv,
 		return fbnic_build_skb(nv, pkt);
 	case XDP_TX:
 		return ERR_PTR(fbnic_pkt_tx(nv, pkt));
+	case XDP_REDIRECT:
+		if (xdp_do_redirect(nv->napi.dev, &pkt->buff, xdp_prog))
+			return ERR_PTR(-FBNIC_XDP_REDIRECT_ERR);
+		return ERR_PTR(-FBNIC_XDP_REDIRECT);
 	default:
 		bpf_warn_invalid_xdp_action(nv->napi.dev, xdp_prog, act);
 		fallthrough;
@@ -1280,6 +1286,7 @@ static int fbnic_clean_rcq(struct fbnic_napi_vector *nv,
 	struct fbnic_ring *rcq = &qt->cmpl;
 	struct fbnic_pkt_buff *pkt;
 	__le64 *raw_rcd, done;
+	bool redirect = false;
 	u32 head = rcq->head;
 
 	done = (head & (rcq->size_mask + 1)) ? cpu_to_le64(FBNIC_RCD_DONE) : 0;
@@ -1336,6 +1343,16 @@ static int fbnic_clean_rcq(struct fbnic_napi_vector *nv,
 				pkt_tail = nv->qt[0].sub1.tail;
 			} else if (PTR_ERR(skb) == -FBNIC_XDP_CONSUME) {
 				fbnic_put_pkt_buff(qt, pkt, 1);
+			} else if (skb == ERR_PTR(-FBNIC_XDP_REDIRECT)) {
+				redirect = true;
+			} else if (skb == ERR_PTR(-FBNIC_XDP_REDIRECT_ERR)) {
+				/* A broadcast may queue frames, then fail. XDP
+				 * consumed the packet, so it also counts as
+				 * received.
+				 */
+				redirect = true;
+				dropped++;
+				fbnic_put_pkt_buff(qt, pkt, 1);
 			} else {
 				if (!skb)
 					alloc_failed++;
@@ -1377,6 +1394,8 @@ static int fbnic_clean_rcq(struct fbnic_napi_vector *nv,
 
 	if (pkt_tail >= 0)
 		fbnic_pkt_commit_tail(nv, pkt_tail);
+	if (redirect)
+		xdp_do_flush();
 
 	/* Unmap and free processed buffers */
 	if (head0 >= 0)
-- 
2.56.0

[PATCH net-next v1 09/11] eth: fbnic: Use netmem for RX headers and XDP_TX

[Björn Töpel]

fbnic reaches header and XDP_TX buffers through struct page, which a
provider net_iov does not have. Use netmem for both, and set up the
xdp_buff with xdp_init_buff_netmem(), so XDP_PASS copies heads that
must not leave the RX poll.

Signed-off-by: Björn Töpel <bjorn@kernel.org>
---
 drivers/net/ethernet/meta/fbnic/fbnic_txrx.c | 47 +++++++++-----------
 drivers/net/ethernet/meta/fbnic/fbnic_txrx.h |  3 ++
 2 files changed, 25 insertions(+), 25 deletions(-)

diff --git a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c
index 51d573c9c5b8..bfaa284e497e 100644
--- a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c
+++ b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c
@@ -665,14 +665,14 @@ static void fbnic_clean_twq1(struct fbnic_napi_vector *nv, bool pp_allow_direct,
 	unsigned int head = ring->head;
 
 	while (hw_head != head) {
-		struct page *page;
+		netmem_ref netmem;
 		u64 twd;
 
 		if (unlikely(!(ring->desc[head] & FBNIC_TWD_TYPE(AL))))
 			goto next_desc;
 
 		twd = le64_to_cpu(ring->desc[head]);
-		page = ring->tx_buf[head];
+		netmem = ring->tx_netmem[head];
 
 		/* TYPE_AL is 2, TYPE_LAST_AL is 3. So this trick gives
 		 * us one increment per packet, with no branches.
@@ -681,8 +681,8 @@ static void fbnic_clean_twq1(struct fbnic_napi_vector *nv, bool pp_allow_direct,
 				 FBNIC_TWD_TYPE_AL;
 		total_bytes += FIELD_GET(FBNIC_TWD_LEN_MASK, twd);
 
-		page_pool_put_page(pp_page_to_nmdesc(page)->pp, page, -1,
-				   pp_allow_direct);
+		page_pool_put_netmem(netmem_get_pp(netmem), netmem, -1,
+				     pp_allow_direct);
 next_desc:
 		head++;
 		head &= ring->size_mask;
@@ -766,15 +766,14 @@ static void fbnic_page_pool_init(struct fbnic_ring *ring, unsigned int idx,
 	rx_buf->netmem = netmem;
 }
 
-static struct page *
+static netmem_ref
 fbnic_page_pool_get_head(struct fbnic_q_triad *qt, unsigned int idx)
 {
 	struct fbnic_rx_buf *rx_buf = &qt->sub0.rx_buf[idx];
 
 	rx_buf->pagecnt_bias--;
 
-	/* sub0 is always fed system pages, from the NAPI-level page_pool */
-	return netmem_to_page(rx_buf->netmem);
+	return rx_buf->netmem;
 }
 
 static netmem_ref
@@ -1003,12 +1002,13 @@ static void fbnic_pkt_prepare(struct fbnic_napi_vector *nv, u64 rcd,
 	unsigned int hdr_pg_idx = fbnic_rcd_bd_idx(&qt->sub0, rcd);
 	unsigned int frame_sz, hdr_pg_start, hdr_pg_end, headroom;
 	unsigned char *hdr_start;
-	struct page *page;
+	netmem_ref netmem;
 
 	/* data_hard_start should always be NULL when this is called */
 	WARN_ON_ONCE(pkt->buff.data_hard_start);
 
-	page = fbnic_page_pool_get_head(qt, hdr_pg_idx);
+	netmem = fbnic_page_pool_get_head(qt, hdr_pg_idx);
+	pkt->head = netmem;
 
 	/* Short-cut the end calculation if we know page is fully consumed */
 	hdr_pg_end = FIELD_GET(FBNIC_RCD_AL_PAGE_FIN, rcd) ?
@@ -1017,16 +1017,16 @@ static void fbnic_pkt_prepare(struct fbnic_napi_vector *nv, u64 rcd,
 
 	headroom = hdr_pg_off - hdr_pg_start + FBNIC_RX_PAD;
 	frame_sz = hdr_pg_end - hdr_pg_start;
-	xdp_init_buff(&pkt->buff, frame_sz, &qt->xdp_rxq);
+	xdp_init_buff_netmem(&pkt->buff, frame_sz, &qt->xdp_rxq, netmem,
+			     &pkt->ext);
 	hdr_pg_start += fbnic_rcd_bd_page_offset(&qt->sub0, rcd);
 
 	/* Sync DMA buffer */
-	dma_sync_single_range_for_cpu(nv->dev, page_pool_get_dma_addr(page),
-				      hdr_pg_start, frame_sz,
-				      DMA_BIDIRECTIONAL);
+	page_pool_dma_sync_netmem_for_cpu(qt->sub0.page_pool, netmem,
+					  hdr_pg_start, frame_sz);
 
 	/* Build frame around buffer */
-	hdr_start = page_address(page) + hdr_pg_start;
+	hdr_start = netmem_address(netmem) + hdr_pg_start;
 	net_prefetch(hdr_start + headroom);
 	xdp_prepare_buff(&pkt->buff, hdr_start, headroom,
 			 len - FBNIC_RX_PAD, true);
@@ -1073,8 +1073,6 @@ static void fbnic_add_rx_frag(struct fbnic_napi_vector *nv, u64 rcd,
 static void fbnic_put_pkt_buff(struct fbnic_q_triad *qt,
 			       struct fbnic_pkt_buff *pkt, int budget)
 {
-	struct page *page;
-
 	if (!pkt->buff.data_hard_start)
 		return;
 
@@ -1093,8 +1091,7 @@ static void fbnic_put_pkt_buff(struct fbnic_q_triad *qt,
 		}
 	}
 
-	page = virt_to_page(pkt->buff.data_hard_start);
-	page_pool_put_full_page(qt->sub0.page_pool, page, !!budget);
+	page_pool_put_full_netmem(qt->sub0.page_pool, pkt->head, !!budget);
 }
 
 static struct sk_buff *fbnic_build_skb(struct fbnic_napi_vector *nv,
@@ -1121,7 +1118,7 @@ static long fbnic_pkt_tx(struct fbnic_napi_vector *nv,
 	unsigned int tail = ring->tail;
 	struct skb_shared_info *shinfo;
 	skb_frag_t *frag = NULL;
-	struct page *page;
+	netmem_ref netmem;
 	dma_addr_t dma;
 	__le64 *twd;
 
@@ -1139,9 +1136,9 @@ static long fbnic_pkt_tx(struct fbnic_napi_vector *nv,
 		return -FBNIC_XDP_CONSUME;
 	}
 
-	page = virt_to_page(pkt->buff.data_hard_start);
-	offset = offset_in_page(pkt->buff.data);
-	dma = page_pool_get_dma_addr(page);
+	netmem = pkt->head;
+	offset = (u8 *)pkt->buff.data - (u8 *)netmem_address(netmem);
+	dma = page_pool_get_dma_addr_netmem(netmem);
 
 	size = pkt->buff.data_end - pkt->buff.data;
 
@@ -1150,7 +1147,7 @@ static long fbnic_pkt_tx(struct fbnic_napi_vector *nv,
 						 DMA_BIDIRECTIONAL);
 		dma += offset;
 
-		ring->tx_buf[tail] = page;
+		ring->tx_netmem[tail] = netmem;
 
 		twd = &ring->desc[tail];
 		*twd = cpu_to_le64(FIELD_PREP(FBNIC_TWD_ADDR_MASK, dma) |
@@ -1165,8 +1162,8 @@ static long fbnic_pkt_tx(struct fbnic_napi_vector *nv,
 			break;
 
 		offset = skb_frag_off(frag);
-		page = skb_frag_page(frag);
-		dma = page_pool_get_dma_addr(page);
+		netmem = skb_frag_netmem(frag);
+		dma = page_pool_get_dma_addr_netmem(netmem);
 
 		size = skb_frag_size(frag);
 		data_len -= size;
diff --git a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.h b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.h
index 6ee3cce3d942..d036c8399364 100644
--- a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.h
+++ b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.h
@@ -82,7 +82,9 @@ struct fbnic_net;
 struct fbnic_pkt_buff {
 	struct xdp_buff buff;
 	ktime_t hwtstamp;
+	netmem_ref head;
 	bool add_frag_failed;
+	struct xdp_buff_ext ext;
 };
 
 struct fbnic_queue_stats {
@@ -123,6 +125,7 @@ struct fbnic_ring {
 	union {
 		struct fbnic_pkt_buff *pkt;	/* RCQ */
 		struct fbnic_rx_buf *rx_buf;	/* BDQ */
+		netmem_ref *tx_netmem;		/* TWQ1 */
 		void **tx_buf;			/* TWQ */
 		void *buffer;			/* Generic pointer */
 	};
-- 
2.56.0

[PATCH net-next v1 10/11] eth: fbnic: Support RX buffers it cannot split

[Björn Töpel]

fbnic packs several headers, and several payloads, into one page. A
provider without MP_CAP_FRAG needs one descriptor per buffer. Give
such a buffer a single reference, program the queue so each header and
payload takes a buffer of its own, and drop packets that break this.

Header split is now set per queue, so link up changes only the drop
mode.

Signed-off-by: Björn Töpel <bjorn@kernel.org>
---
 drivers/net/ethernet/meta/fbnic/fbnic_txrx.c | 119 ++++++++++++++++---
 drivers/net/ethernet/meta/fbnic/fbnic_txrx.h |   2 +
 2 files changed, 102 insertions(+), 19 deletions(-)

diff --git a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c
index bfaa284e497e..d96c957d6fba 100644
--- a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c
+++ b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c
@@ -761,8 +761,14 @@ static void fbnic_page_pool_init(struct fbnic_ring *ring, unsigned int idx,
 {
 	struct fbnic_rx_buf *rx_buf = &ring->rx_buf[idx];
 
-	page_pool_fragment_netmem(netmem, FBNIC_PAGECNT_BIAS_MAX);
-	rx_buf->pagecnt_bias = FBNIC_PAGECNT_BIAS_MAX;
+	/* A whole buffer has one reference. The packet takes it, so the buffer
+	 * may be reused before its BDQ entry is cleaned.
+	 */
+	rx_buf->pagecnt_bias = 1;
+	if (page_pool_supports_frag(ring->page_pool)) {
+		page_pool_fragment_netmem(netmem, FBNIC_PAGECNT_BIAS_MAX);
+		rx_buf->pagecnt_bias = FBNIC_PAGECNT_BIAS_MAX;
+	}
 	rx_buf->netmem = netmem;
 }
 
@@ -792,7 +798,8 @@ static void fbnic_page_pool_drain(struct fbnic_ring *ring, unsigned int idx,
 	struct fbnic_rx_buf *rx_buf = &ring->rx_buf[idx];
 	netmem_ref netmem = rx_buf->netmem;
 
-	if (!page_pool_unref_netmem(netmem, rx_buf->pagecnt_bias))
+	if (rx_buf->pagecnt_bias &&
+	    !page_pool_unref_netmem(netmem, rx_buf->pagecnt_bias))
 		page_pool_put_unrefed_netmem(ring->page_pool, netmem, -1,
 					     !!budget);
 
@@ -993,6 +1000,19 @@ static unsigned int fbnic_hdr_pg_end(unsigned int pg_off, unsigned int len)
 	return ALIGN(pg_off, 128) - FBNIC_RX_HROOM;
 }
 
+/* A buffer that cannot be split has one reference and takes one descriptor */
+static bool fbnic_bd_one_desc(struct fbnic_napi_vector *nv,
+			      const struct fbnic_ring *bdq, unsigned int idx,
+			      bool ok)
+{
+	if (likely(page_pool_supports_frag(bdq->page_pool) ||
+		   (bdq->rx_buf[idx].pagecnt_bias == 1 && ok)))
+		return true;
+
+	netdev_err_once(nv->napi.dev, "Bad descriptor for a whole RX buffer\n");
+	return false;
+}
+
 static void fbnic_pkt_prepare(struct fbnic_napi_vector *nv, u64 rcd,
 			      struct fbnic_pkt_buff *pkt,
 			      struct fbnic_q_triad *qt)
@@ -1007,6 +1027,17 @@ static void fbnic_pkt_prepare(struct fbnic_napi_vector *nv, u64 rcd,
 	/* data_hard_start should always be NULL when this is called */
 	WARN_ON_ONCE(pkt->buff.data_hard_start);
 
+	/* MAX_HDR leaves no room for a second header in a whole buffer, so
+	 * the device closes it, with PAGE_FIN, after the first.
+	 */
+	pkt->add_frag_failed = false;
+	if (unlikely(!fbnic_bd_one_desc(nv, &qt->sub0, hdr_pg_idx,
+					FIELD_GET(FBNIC_RCD_AL_PAGE_FIN,
+						  rcd)))) {
+		pkt->add_frag_failed = true;
+		return;
+	}
+
 	netmem = fbnic_page_pool_get_head(qt, hdr_pg_idx);
 	pkt->head = netmem;
 
@@ -1032,7 +1063,6 @@ static void fbnic_pkt_prepare(struct fbnic_napi_vector *nv, u64 rcd,
 			 len - FBNIC_RX_PAD, true);
 
 	pkt->hwtstamp = 0;
-	pkt->add_frag_failed = false;
 }
 
 static void fbnic_add_rx_frag(struct fbnic_napi_vector *nv, u64 rcd,
@@ -1046,8 +1076,20 @@ static void fbnic_add_rx_frag(struct fbnic_napi_vector *nv, u64 rcd,
 	netmem_ref netmem;
 	bool added;
 
+	if (unlikely(!fbnic_bd_one_desc(nv, &qt->sub1, pg_idx, !pg_off))) {
+		pkt->add_frag_failed = true;
+		return;
+	}
+
 	netmem = fbnic_page_pool_get_data(qt, pg_idx);
 
+	/* Without a header buffer, the packet is dropped. */
+	if (unlikely(!pkt->buff.data_hard_start)) {
+		page_pool_put_full_netmem(qt->sub1.page_pool, netmem, true);
+		pkt->add_frag_failed = true;
+		return;
+	}
+
 	truesize = FIELD_GET(FBNIC_RCD_AL_PAGE_FIN, rcd) ?
 		   FBNIC_BD_PAGE_SIZE - pg_off :
 		   ALIGN(len, FBNIC_RX_PAYLD_ALIGN);
@@ -1608,6 +1650,13 @@ void fbnic_free_napi_vectors(struct fbnic_net *fbn)
 			fbnic_free_napi_vector(fbn, fbn->napi[i]);
 }
 
+/* No second header fits behind the first in a whole buffer */
+static u32 fbnic_whole_hdr_max(u32 hroom)
+{
+	return FBNIC_BD_PAGE_SIZE - ALIGN(hroom, FBNIC_RX_HDR_ALIGN) -
+	       FBNIC_RX_TROOM - FBNIC_RX_PAD;
+}
+
 static int
 fbnic_alloc_qt_page_pools(struct fbnic_net *fbn, struct fbnic_q_triad *qt,
 			  unsigned int rxq_idx, u32 rx_page_size)
@@ -1644,6 +1693,9 @@ fbnic_alloc_qt_page_pools(struct fbnic_net *fbn, struct fbnic_q_triad *qt,
 		return PTR_ERR(pp);
 
 	qt->sub0.page_pool = pp;
+	qt->hdr_max = 0;
+	if (!page_pool_supports_frag(pp))
+		qt->hdr_max = fbnic_whole_hdr_max(FBNIC_RX_HROOM);
 	if (netif_rxq_has_unreadable_mp(fbn->netdev, rxq_idx)) {
 		pp_params.order = get_order(rx_page_size);
 		pp_params.max_len = rx_page_size;
@@ -2672,6 +2724,14 @@ static void fbnic_enable_bdq(struct fbnic_ring *hpq, struct fbnic_ring *ppq)
 	fbnic_ring_wr32(hpq, FBNIC_QUEUE_BDQ_CTL, bdq_ctl);
 }
 
+static u32 fbnic_drop_mode(struct fbnic_net *fbn, bool tx_pause)
+{
+	if (!tx_pause && fbn->num_rx_queues > 1)
+		return FBNIC_QUEUE_RDE_CTL0_DROP_IMMEDIATE;
+
+	return FBNIC_QUEUE_RDE_CTL0_DROP_NEVER;
+}
+
 static void fbnic_config_drop_mode_rcq(struct fbnic_napi_vector *nv,
 				       struct fbnic_ring *rcq, bool tx_pause,
 				       bool hdr_split)
@@ -2679,10 +2739,7 @@ static void fbnic_config_drop_mode_rcq(struct fbnic_napi_vector *nv,
 	struct fbnic_net *fbn = netdev_priv(nv->napi.dev);
 	u32 drop_mode, rcq_ctl;
 
-	if (!tx_pause && fbn->num_rx_queues > 1)
-		drop_mode = FBNIC_QUEUE_RDE_CTL0_DROP_IMMEDIATE;
-	else
-		drop_mode = FBNIC_QUEUE_RDE_CTL0_DROP_NEVER;
+	drop_mode = fbnic_drop_mode(fbn, tx_pause);
 
 	/* Specify packet layout */
 	rcq_ctl = FIELD_PREP(FBNIC_QUEUE_RDE_CTL0_DROP_MODE_MASK, drop_mode) |
@@ -2693,20 +2750,32 @@ static void fbnic_config_drop_mode_rcq(struct fbnic_napi_vector *nv,
 	fbnic_ring_wr32(rcq, FBNIC_QUEUE_RDE_CTL0, rcq_ctl);
 }
 
+static u32 fbnic_rcq_hds_thresh(const struct fbnic_net *fbn,
+				const struct fbnic_ring *rcq)
+{
+	const struct fbnic_q_triad *qt;
+
+	qt = container_of_const(rcq, struct fbnic_q_triad, cmpl);
+	return qt->hdr_max ? : fbn->hds_thresh;
+}
+
+/* Only the drop mode: a queue start may write the rest of the register */
 void fbnic_config_drop_mode(struct fbnic_net *fbn, bool txp)
 {
-	bool hds;
+	struct fbnic_ring *rcq;
+	u32 drop_mode, rcq_ctl;
 	int i, t;
 
-	hds = fbn->hds_thresh < FBNIC_HDR_BYTES_MIN;
-
+	drop_mode = fbnic_drop_mode(fbn, txp);
 	for (i = 0; i < fbn->num_napi; i++) {
 		struct fbnic_napi_vector *nv = fbn->napi[i];
 
 		for (t = 0; t < nv->rxt_count; t++) {
-			struct fbnic_q_triad *qt = &nv->qt[nv->txt_count + t];
-
-			fbnic_config_drop_mode_rcq(nv, &qt->cmpl, txp, hds);
+			rcq = &nv->qt[nv->txt_count + t].cmpl;
+			rcq_ctl = fbnic_ring_rd32(rcq, FBNIC_QUEUE_RDE_CTL0);
+			FIELD_MODIFY(FBNIC_QUEUE_RDE_CTL0_DROP_MODE_MASK,
+				     &rcq_ctl, drop_mode);
+			fbnic_ring_wr32(rcq, FBNIC_QUEUE_RDE_CTL0, rcq_ctl);
 		}
 	}
 }
@@ -2771,26 +2840,38 @@ void fbnic_config_rx_frames(struct fbnic_napi_vector *nv)
 static void fbnic_enable_rcq(struct fbnic_napi_vector *nv,
 			     struct fbnic_ring *rcq)
 {
+	struct fbnic_q_triad *qt = container_of_const(rcq, struct fbnic_q_triad,
+						     cmpl);
 	struct fbnic_net *fbn = netdev_priv(nv->napi.dev);
+	u32 payld_pg_cl = FBNIC_RX_PAYLD_PG_CL;
+	u32 payld_off = FBNIC_RX_PAYLD_OFFSET;
 	u32 log_size = fls(rcq->size_mask);
 	u32 rcq_ctl = 0;
 	bool hdr_split;
 	u32 hds_thresh;
 
+	/* At 4 KiB, PAYLD_OFF puts each payload in a fresh buffer, and
+	 * PAYLD_PG_CL closes it after the payload.
+	 */
+	if (!page_pool_supports_frag(qt->sub1.page_pool)) {
+		payld_off = ilog2(FBNIC_BD_PAGE_SIZE / FBNIC_RX_PAYLD_ALIGN);
+		payld_pg_cl = payld_off;
+	}
+
 	/* Force lower bound on MAX_HEADER_BYTES. Below this, all frames should
 	 * be split at L4. It would also result in the frames being split at
 	 * L2/L3 depending on the frame size.
 	 */
-	hdr_split = fbn->hds_thresh < FBNIC_HDR_BYTES_MIN;
+	hds_thresh = fbnic_rcq_hds_thresh(fbn, rcq);
+	hdr_split = hds_thresh < FBNIC_HDR_BYTES_MIN;
 	fbnic_config_drop_mode_rcq(nv, rcq, fbn->tx_pause, hdr_split);
 
-	hds_thresh = max(fbn->hds_thresh, FBNIC_HDR_BYTES_MIN);
+	hds_thresh = max(hds_thresh, FBNIC_HDR_BYTES_MIN);
 	rcq_ctl |= FIELD_PREP(FBNIC_QUEUE_RDE_CTL1_PADLEN_MASK, FBNIC_RX_PAD) |
 		   FIELD_PREP(FBNIC_QUEUE_RDE_CTL1_MAX_HDR_MASK, hds_thresh) |
-		   FIELD_PREP(FBNIC_QUEUE_RDE_CTL1_PAYLD_OFF_MASK,
-			      FBNIC_RX_PAYLD_OFFSET) |
+		   FIELD_PREP(FBNIC_QUEUE_RDE_CTL1_PAYLD_OFF_MASK, payld_off) |
 		   FIELD_PREP(FBNIC_QUEUE_RDE_CTL1_PAYLD_PG_CL_MASK,
-			      FBNIC_RX_PAYLD_PG_CL);
+			      payld_pg_cl);
 	fbnic_ring_wr32(rcq, FBNIC_QUEUE_RDE_CTL1, rcq_ctl);
 
 	/* Reset head/tail */
diff --git a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.h b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.h
index d036c8399364..3077d9bbd587 100644
--- a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.h
+++ b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.h
@@ -65,6 +65,7 @@ struct fbnic_net;
 #define FBNIC_RX_HROOM \
 	(ALIGN(FBNIC_RX_TROOM + FBNIC_RX_HROOM_PAD, 128) - FBNIC_RX_TROOM)
 #define FBNIC_RX_PAD			0
+#define FBNIC_RX_HDR_ALIGN		128
 #define FBNIC_RX_PAYLD_ALIGN		128
 #define FBNIC_RX_PAYLD_OFFSET		0
 #define FBNIC_RX_PAYLD_PG_CL		0
@@ -165,6 +166,7 @@ struct fbnic_ring {
 struct fbnic_q_triad {
 	struct fbnic_ring sub0, sub1, cmpl;
 	struct xdp_rxq_info xdp_rxq;
+	u16 hdr_max;	/* Header bytes for whole header buffers, or 0 */
 };
 
 struct fbnic_napi_vector {
-- 
2.56.0

[PATCH net-next v1 11/11] eth: fbnic: Support AF_XDP page pool zero copy

[Björn Töpel]

Header and payload rings share one provider page pool, so refills even
them out, and a short pool does not starve one of them. Payload
buffers start at chunk offset 0, and the UMEM headroom rules require
headroom only in the first buffer. A chunk is one 4 KiB device page,
also with larger kernel pages, so rings fed by the provider post one
descriptor per chunk.

Socket TX shares the XDP TX ring and runs after RX: XDP_TX drops
frames that do not fit, while socket descriptors can wait. The TX
need_wakeup flag stays set, as for ice, since clearing it can miss the
last wakeup.

Signed-off-by: Björn Töpel <bjorn@kernel.org>
---
 Documentation/networking/af_xdp.rst           |  58 +++
 .../net/ethernet/meta/fbnic/fbnic_netdev.c    |  18 +
 drivers/net/ethernet/meta/fbnic/fbnic_txrx.c  | 428 ++++++++++++++----
 drivers/net/ethernet/meta/fbnic/fbnic_txrx.h  |  25 +-
 4 files changed, 446 insertions(+), 83 deletions(-)

diff --git a/Documentation/networking/af_xdp.rst b/Documentation/networking/af_xdp.rst
index e73a0005e169..f152ae8812d1 100644
--- a/Documentation/networking/af_xdp.rst
+++ b/Documentation/networking/af_xdp.rst
@@ -346,6 +346,64 @@ Note that a UMEM can be shared between sockets on the same queue id
 and device, as well as between queues on the same device and between
 devices at the same time.
 
+Zero-copy through page pool
+---------------------------
+
+Some drivers, for example fbnic, receive into the UMEM through a page
+pool memory provider instead of their own allocator. The socket API
+does not change: zero-copy mode is selected as before, and XDP_OPTIONS
+reports it.
+
+The requirements are:
+
+* Aligned chunks of a size the driver supports, 4 KiB for fbnic.
+* The queue id has an RX queue, also for a socket that only transmits.
+  Such a socket needs the FILL and COMPLETION rings, like any other.
+* The driver accepts the headroom, MTU and XDP program. fbnic accepts
+  a UMEM headroom of 0, 128 or 256 bytes. An MTU larger than its header
+  buffer needs XDP_USE_SG, and frags support in an XDP program. The
+  first buffer of a larger frame may hold only its headers, and the
+  other buffers start at offset 0 of their chunk. Such a frame can take
+  one chunk more than in copy mode.
+
+If the device cannot do zero-copy for the socket, a bind with
+XDP_ZEROCOPY fails with EOPNOTSUPP, while a bind without it falls back
+to copy mode. MTU and XDP program changes that a bound socket cannot
+handle are refused.
+
+Binding restarts the RX queue. The queue takes chunks from the FILL
+ring and returns them on the RX ring. A FILL entry for a chunk that
+the RX side still owns is dropped; TX chunks stay with userspace until
+their completion, as before. XDP_PASS copies the packet. Redirects to
+anything but a socket copy single-buffer packets and drop others. A
+redirect to a socket of another queue fails, and generic XDP cannot
+redirect to the socket.
+
+When the queue has no chunks left to post, RX waits for the FILL ring.
+With XDP_USE_NEED_WAKEUP, the flag on the FILL ring stays set, and
+userspace must call poll(), or recvfrom() with an RX ring or sendto()
+with a TX ring, after it posts entries. Without the flag, the queue
+polls the FILL ring, which costs CPU time while the ring is empty.
+Post FILL entries before bind() if possible.
+
+A driver supports this mode when it:
+
+* uses netmem for header buffers and XDP_TX, and handles page pools
+  whose buffers cannot be split (see page_pool_supports_frag()),
+* selects the provider with PP_FLAG_MEM_PROVIDER and pp_rx_role for
+  one page pool of each queue, and honours the rx_headroom queue
+  configuration,
+* calls page_pool_refill_done() after each refill,
+* sends socket TX descriptors from the NAPI of the RX queue, with
+  xsk_tx_peek_release_desc_batch() for the pool that xsk_rxq_pool()
+  returns,
+* advertises NETDEV_XDP_ACT_XSK_ZEROCOPY and xdp_zc_max_segs,
+* checks the setup in XDP_VALIDATE_XSK_POOL, and sets xsk_page_pool
+  in its netdev_queue_mgmt_ops.
+
+The core schedules that NAPI for RX and TX wakeups, so no
+ndo_xsk_wakeup is needed.
+
 XDP_USE_NEED_WAKEUP bind flag
 -----------------------------
 
diff --git a/drivers/net/ethernet/meta/fbnic/fbnic_netdev.c b/drivers/net/ethernet/meta/fbnic/fbnic_netdev.c
index ae2585508690..8cb5b5b93c70 100644
--- a/drivers/net/ethernet/meta/fbnic/fbnic_netdev.c
+++ b/drivers/net/ethernet/meta/fbnic/fbnic_netdev.c
@@ -5,6 +5,7 @@
 #include <linux/ipv6.h>
 #include <linux/types.h>
 #include <net/netdev_queues.h>
+#include <net/xdp_sock_drv.h>
 
 #include "fbnic.h"
 #include "fbnic_netdev.h"
@@ -276,6 +277,7 @@ static int fbnic_set_mac(struct net_device *netdev, void *p)
 static int fbnic_change_mtu(struct net_device *dev, int new_mtu)
 {
 	struct fbnic_net *fbn = netdev_priv(dev);
+	int err;
 
 	if (fbnic_check_split_frames(fbn->xdp_prog, new_mtu, fbn->hds_thresh)) {
 		dev_err(&dev->dev,
@@ -285,6 +287,13 @@ static int fbnic_change_mtu(struct net_device *dev, int new_mtu)
 		return -EINVAL;
 	}
 
+	err = xsk_validate_dev(dev, new_mtu, NULL, XDP_MODE_DRV, NULL);
+	if (err) {
+		dev_err(&dev->dev, "MTU %d does not fit a bound AF_XDP socket\n",
+			new_mtu);
+		return err;
+	}
+
 	WRITE_ONCE(dev->mtu, new_mtu);
 
 	return 0;
@@ -538,6 +547,13 @@ static int fbnic_bpf(struct net_device *netdev, struct netdev_bpf *bpf)
 {
 	struct bpf_prog *prog = bpf->prog, *prev_prog;
 	struct fbnic_net *fbn = netdev_priv(netdev);
+	const struct xsk_rx_check *check;
+
+	if (bpf->command == XDP_VALIDATE_XSK_POOL) {
+		check = bpf->xsk_check.check;
+		prog = check->prog ? : READ_ONCE(fbn->xdp_prog);
+		return fbnic_xsk_validate(check, prog, bpf->xsk_check.extack);
+	}
 
 	if (bpf->command != XDP_SETUP_PROG)
 		return -EINVAL;
@@ -825,6 +841,8 @@ struct net_device *fbnic_netdev_alloc(struct fbnic_dev *fbd)
 	netdev->xdp_features = NETDEV_XDP_ACT_BASIC |
 			       NETDEV_XDP_ACT_REDIRECT |
 			       NETDEV_XDP_ACT_RX_SG;
+	netdev->xdp_features |= NETDEV_XDP_ACT_XSK_ZEROCOPY;
+	netdev->xdp_zc_max_segs = FBNIC_MAX_RX_BUFS;
 
 	netdev->min_mtu = IPV6_MIN_MTU;
 	netdev->max_mtu = FBNIC_MAX_JUMBO_FRAME_SIZE - ETH_HLEN;
diff --git a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c
index d96c957d6fba..c358003f053b 100644
--- a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c
+++ b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.c
@@ -12,6 +12,7 @@
 #include <net/page_pool/helpers.h>
 #include <net/tcp.h>
 #include <net/xdp.h>
+#include <net/xdp_sock_drv.h>
 
 #include "fbnic.h"
 #include "fbnic_csr.h"
@@ -43,6 +44,8 @@ struct fbnic_xmit_cb {
 
 #define FBNIC_XMIT_NOUNMAP	((void *)1)
 
+#define FBNIC_XSK_TX_BUDGET	64
+
 u32 __iomem *fbnic_ring_csr_base(const struct fbnic_ring *ring)
 {
 	unsigned long csr_base = (unsigned long)ring->doorbell;
@@ -657,22 +660,45 @@ static void fbnic_clean_twq0(struct fbnic_napi_vector *nv, int napi_budget,
 	}
 }
 
+struct fbnic_xsk_done {
+	struct xsk_buff_pool *pool;
+	u32 descs;
+};
+
+static void fbnic_xsk_done_flush(struct fbnic_xsk_done *done)
+{
+	if (done->descs)
+		xsk_tx_completed(done->pool, done->descs);
+	done->descs = 0;
+}
+
+static void fbnic_xsk_done_add(struct fbnic_xsk_done *done,
+			       struct xsk_buff_pool *pool)
+{
+	if (done->pool != pool) {
+		fbnic_xsk_done_flush(done);
+		done->pool = pool;
+	}
+	done->descs++;
+}
+
 static void fbnic_clean_twq1(struct fbnic_napi_vector *nv, bool pp_allow_direct,
 			     struct fbnic_ring *ring, bool discard,
 			     unsigned int hw_head)
 {
 	u64 total_bytes = 0, total_packets = 0;
+	struct fbnic_xsk_done done = {};
 	unsigned int head = ring->head;
 
 	while (hw_head != head) {
-		netmem_ref netmem;
+		struct fbnic_xdp_tx_buf *buf;
 		u64 twd;
 
 		if (unlikely(!(ring->desc[head] & FBNIC_TWD_TYPE(AL))))
 			goto next_desc;
 
 		twd = le64_to_cpu(ring->desc[head]);
-		netmem = ring->tx_netmem[head];
+		buf = &ring->xdp_buf[head];
 
 		/* TYPE_AL is 2, TYPE_LAST_AL is 3. So this trick gives
 		 * us one increment per packet, with no branches.
@@ -681,13 +707,18 @@ static void fbnic_clean_twq1(struct fbnic_napi_vector *nv, bool pp_allow_direct,
 				 FBNIC_TWD_TYPE_AL;
 		total_bytes += FIELD_GET(FBNIC_TWD_LEN_MASK, twd);
 
-		page_pool_put_netmem(netmem_get_pp(netmem), netmem, -1,
-				     pp_allow_direct);
+		if (buf->xsk)
+			fbnic_xsk_done_add(&done, buf->xsk);
+		else
+			page_pool_put_netmem(netmem_get_pp(buf->netmem),
+					     buf->netmem, -1, pp_allow_direct);
 next_desc:
 		head++;
 		head &= ring->size_mask;
 	}
 
+	fbnic_xsk_done_flush(&done);
+
 	if (!total_bytes)
 		return;
 
@@ -938,66 +969,146 @@ static void fbnic_bd_prep(struct fbnic_ring *bdq, u16 id, netmem_ref netmem)
 	} while (--i);
 }
 
+static bool fbnic_fill_bdq_one(struct fbnic_ring *bdq, unsigned int *tail)
+{
+	netmem_ref netmem;
+
+	netmem = page_pool_dev_alloc_netmems(bdq->page_pool);
+	if (!netmem) {
+		u64_stats_update_begin(&bdq->stats.syncp);
+		bdq->stats.bdq.alloc_failed++;
+		u64_stats_update_end(&bdq->stats.syncp);
+
+		return false;
+	}
+
+	fbnic_page_pool_init(bdq, *tail, netmem);
+	fbnic_bd_prep(bdq, *tail, netmem);
+
+	*tail = (*tail + 1) & bdq->size_mask;
+	return true;
+}
+
+static void fbnic_fill_bdq_done(struct fbnic_ring *bdq, unsigned int tail)
+{
+	if (bdq->tail == tail)
+		return;
+
+	bdq->tail = tail;
+
+	/* Force DMA writes to flush before writing to tail */
+	dma_wmb();
+
+	writel(tail * fbnic_bd_page_count(bdq), bdq->doorbell);
+}
+
+static enum pp_refill fbnic_bdq_state(struct fbnic_ring *bdq)
+{
+	return fbnic_desc_unused(bdq) ? PP_REFILL_PARTIAL : PP_REFILL_FULL;
+}
+
 static void fbnic_fill_bdq(struct fbnic_ring *bdq)
 {
 	unsigned int count = fbnic_desc_unused(bdq);
-	unsigned int i = bdq->tail;
+	unsigned int tail = bdq->tail;
 
-	if (!count)
-		return;
+	while (count-- && fbnic_fill_bdq_one(bdq, &tail))
+		;
 
-	do {
-		netmem_ref netmem;
-
-		netmem = page_pool_dev_alloc_netmems(bdq->page_pool);
-		if (!netmem) {
-			u64_stats_update_begin(&bdq->stats.syncp);
-			bdq->stats.bdq.alloc_failed++;
-			u64_stats_update_end(&bdq->stats.syncp);
-
-			break;
-		}
-
-		fbnic_page_pool_init(bdq, i, netmem);
-		fbnic_bd_prep(bdq, i, netmem);
-
-		i++;
-		i &= bdq->size_mask;
-
-		count--;
-	} while (count);
-
-	if (bdq->tail != i) {
-		bdq->tail = i;
-
-		/* Force DMA writes to flush before writing to tail */
-		dma_wmb();
-
-		writel(i * fbnic_bd_page_count(bdq), bdq->doorbell);
-	}
+	fbnic_fill_bdq_done(bdq, tail);
 }
 
-static unsigned int fbnic_hdr_pg_start(unsigned int pg_off)
+/* Buffers each ring of a shared page pool gets first */
+#define FBNIC_SHARED_BDQ_MIN	NAPI_POLL_WEIGHT
+
+/* Until both rings hold FBNIC_SHARED_BDQ_MIN buffers, the one with fewer
+ * gets the next buffer. Then the one with the smaller share of its capacity
+ * posted does, so a ring that uses more buffers gets more.
+ */
+static bool fbnic_hpq_takes_next(const struct fbnic_ring *hpq,
+				 unsigned int hpq_cnt,
+				 const struct fbnic_ring *ppq,
+				 unsigned int ppq_cnt)
 {
-	/* The headroom of the first header may be larger than FBNIC_RX_HROOM
+	u64 hpq_posted = hpq->size_mask - hpq_cnt;
+	u64 ppq_posted = ppq->size_mask - ppq_cnt;
+
+	if (!hpq_cnt)
+		return false;
+	if (!ppq_cnt)
+		return true;
+	if (min(hpq_posted, ppq_posted) < FBNIC_SHARED_BDQ_MIN)
+		return hpq_posted <= ppq_posted;
+
+	return hpq_posted * ppq->size_mask <= ppq_posted * hpq->size_mask;
+}
+
+static void fbnic_fill_shared_bdqs(struct fbnic_ring *hpq,
+				   struct fbnic_ring *ppq)
+{
+	unsigned int hpq_cnt = fbnic_desc_unused(hpq);
+	unsigned int ppq_cnt = fbnic_desc_unused(ppq);
+	unsigned int hpq_tail = hpq->tail;
+	unsigned int ppq_tail = ppq->tail;
+	bool more = true;
+
+	while (more && (hpq_cnt || ppq_cnt)) {
+		if (fbnic_hpq_takes_next(hpq, hpq_cnt, ppq, ppq_cnt)) {
+			more = fbnic_fill_bdq_one(hpq, &hpq_tail);
+			hpq_cnt -= more;
+		} else {
+			more = fbnic_fill_bdq_one(ppq, &ppq_tail);
+			ppq_cnt -= more;
+		}
+	}
+
+	fbnic_fill_bdq_done(hpq, hpq_tail);
+	fbnic_fill_bdq_done(ppq, ppq_tail);
+}
+
+static bool fbnic_fill_bdqs(struct fbnic_q_triad *qt)
+{
+	struct fbnic_ring *hpq = &qt->sub0, *ppq = &qt->sub1;
+	enum pp_refill state;
+	bool done;
+
+	if (hpq->page_pool == ppq->page_pool) {
+		fbnic_fill_shared_bdqs(hpq, ppq);
+		state = fbnic_bdq_state(hpq);
+		if (state == PP_REFILL_FULL)
+			state = fbnic_bdq_state(ppq);
+		return page_pool_refill_done(hpq->page_pool, state);
+	}
+
+	fbnic_fill_bdq(hpq);
+	fbnic_fill_bdq(ppq);
+	done = page_pool_refill_done(hpq->page_pool, fbnic_bdq_state(hpq));
+	done &= page_pool_refill_done(ppq->page_pool, fbnic_bdq_state(ppq));
+	return done;
+}
+
+static unsigned int fbnic_hdr_pg_start(const struct fbnic_q_triad *qt,
+				       unsigned int pg_off)
+{
+	/* The headroom of the first header may be larger than rx_hroom
 	 * due to alignment. So account for that by just making the page
 	 * offset 0 if we are starting at the first header.
 	 */
-	if (ALIGN(FBNIC_RX_HROOM, 128) > FBNIC_RX_HROOM &&
-	    pg_off == ALIGN(FBNIC_RX_HROOM, 128))
+	if (pg_off == qt->rx_hdr_first)
 		return 0;
 
-	return pg_off - FBNIC_RX_HROOM;
+	return pg_off - qt->rx_hroom;
 }
 
-static unsigned int fbnic_hdr_pg_end(unsigned int pg_off, unsigned int len)
+static unsigned int fbnic_hdr_pg_end(unsigned int pg_off, unsigned int len,
+				     unsigned int hroom)
 {
 	/* Determine the end of the buffer by finding the start of the next
 	 * and then subtracting the headroom from that frame.
 	 */
-	pg_off += len + FBNIC_RX_TROOM + FBNIC_RX_HROOM;
+	pg_off += len + FBNIC_RX_TROOM + hroom;
 
-	return ALIGN(pg_off, 128) - FBNIC_RX_HROOM;
+	return ALIGN(pg_off, 128) - hroom;
 }
 
 /* A buffer that cannot be split has one reference and takes one descriptor */
@@ -1043,8 +1154,11 @@ static void fbnic_pkt_prepare(struct fbnic_napi_vector *nv, u64 rcd,
 
 	/* Short-cut the end calculation if we know page is fully consumed */
 	hdr_pg_end = FIELD_GET(FBNIC_RCD_AL_PAGE_FIN, rcd) ?
-		     FBNIC_BD_PAGE_SIZE : fbnic_hdr_pg_end(hdr_pg_off, len);
-	hdr_pg_start = fbnic_hdr_pg_start(hdr_pg_off);
+		     FBNIC_BD_PAGE_SIZE :
+		     fbnic_hdr_pg_end(hdr_pg_off, len, qt->rx_hroom);
+	hdr_pg_start = fbnic_hdr_pg_start(qt, hdr_pg_off);
+	/* The frame starts after the UMEM headroom */
+	hdr_pg_start += qt->rx_frame_off;
 
 	headroom = hdr_pg_off - hdr_pg_start + FBNIC_RX_PAD;
 	frame_sz = hdr_pg_end - hdr_pg_start;
@@ -1189,7 +1303,8 @@ static long fbnic_pkt_tx(struct fbnic_napi_vector *nv,
 						 DMA_BIDIRECTIONAL);
 		dma += offset;
 
-		ring->tx_netmem[tail] = netmem;
+		ring->xdp_buf[tail].netmem = netmem;
+		ring->xdp_buf[tail].xsk = NULL;
 
 		twd = &ring->desc[tail];
 		*twd = cpu_to_le64(FIELD_PREP(FBNIC_TWD_ADDR_MASK, dma) |
@@ -1327,6 +1442,7 @@ static int fbnic_clean_rcq(struct fbnic_napi_vector *nv,
 	__le64 *raw_rcd, done;
 	bool redirect = false;
 	u32 head = rcq->head;
+	bool filled;
 
 	done = (head & (rcq->size_mask + 1)) ? cpu_to_le64(FBNIC_RCD_DONE) : 0;
 	raw_rcd = &rcq->desc[head & rcq->size_mask];
@@ -1439,11 +1555,10 @@ static int fbnic_clean_rcq(struct fbnic_napi_vector *nv,
 	/* Unmap and free processed buffers */
 	if (head0 >= 0)
 		fbnic_clean_bdq(&qt->sub0, head0, budget);
-	fbnic_fill_bdq(&qt->sub0);
-
 	if (head1 >= 0)
 		fbnic_clean_bdq(&qt->sub1, head1, budget);
-	fbnic_fill_bdq(&qt->sub1);
+	/* Netpoll has no budget and receives nothing */
+	filled = !budget || fbnic_fill_bdqs(qt);
 
 	/* Record the current head/tail of the queue */
 	if (rcq->head != head) {
@@ -1451,7 +1566,8 @@ static int fbnic_clean_rcq(struct fbnic_napi_vector *nv,
 		writel(head & rcq->size_mask, rcq->doorbell);
 	}
 
-	return packets;
+	/* A full budget polls again, to retry the refill */
+	return filled ? packets : budget;
 }
 
 static void fbnic_nv_irq_disable(struct fbnic_napi_vector *nv)
@@ -1471,6 +1587,75 @@ static void fbnic_nv_irq_rearm(struct fbnic_napi_vector *nv)
 		   FBNIC_INTR_CQ_REARM_INTR_UNMASK);
 }
 
+static bool fbnic_xsk_tx_pool(struct fbnic_ring *ring,
+			      struct xsk_buff_pool *pool)
+{
+	unsigned int budget, tail = ring->tail;
+	struct xdp_desc *desc;
+	u32 i, count = 0;
+	dma_addr_t dma;
+	u64 type;
+
+	/* As for ice: a flag cleared with descriptors in flight can miss one */
+	if (xsk_uses_need_wakeup(pool))
+		xsk_set_tx_need_wakeup(pool);
+
+	budget = min(fbnic_desc_unused(ring), FBNIC_XSK_TX_BUDGET);
+	if (budget)
+		count = xsk_tx_peek_release_desc_batch(pool, budget);
+
+	for (i = 0; i < count; i++) {
+		desc = &pool->tx_descs[i];
+		dma = xsk_buff_raw_get_dma(pool, desc->addr);
+		xsk_buff_raw_dma_sync_for_device(pool, dma, desc->len);
+
+		ring->xdp_buf[tail].netmem = 0;
+		ring->xdp_buf[tail].xsk = pool;
+
+		type = xsk_is_eop_desc(desc) ? FBNIC_TWD_TYPE_LAST_AL :
+					       FBNIC_TWD_TYPE_AL;
+		ring->desc[tail] =
+			cpu_to_le64(FIELD_PREP(FBNIC_TWD_ADDR_MASK, dma) |
+				    FIELD_PREP(FBNIC_TWD_LEN_MASK, desc->len) |
+				    FIELD_PREP(FBNIC_TWD_TYPE_MASK, type));
+
+		tail++;
+		tail &= ring->size_mask;
+	}
+	ring->tail = tail;
+
+	/* The batch stops before a packet that does not fit the budget */
+	return !count || budget - count >= pool->xdp_zc_max_segs;
+}
+
+/* After RX: XDP_TX drops frames that do not fit, socket descriptors wait */
+static bool fbnic_xsk_tx(struct fbnic_napi_vector *nv)
+{
+	struct fbnic_ring *ring = &nv->qt[0].sub1;
+	unsigned int tail = ring->tail;
+	struct xsk_buff_pool *pool;
+	unsigned int i, t;
+	bool done = true;
+
+	for (i = 0; i < nv->rxt_count; i++) {
+		t = (nv->xsk_tx_next + i) % nv->rxt_count;
+		pool = nv->qt[nv->txt_count + t].xsk_pool;
+		if (pool && !fbnic_xsk_tx_pool(ring, pool))
+			done = false;
+	}
+	if (nv->rxt_count)
+		nv->xsk_tx_next = (nv->xsk_tx_next + 1) % nv->rxt_count;
+
+	if (tail != ring->tail) {
+		/* Force DMA writes to flush before writing to tail */
+		dma_wmb();
+
+		writel(ring->tail, ring->doorbell);
+	}
+
+	return done;
+}
+
 static int fbnic_poll(struct napi_struct *napi, int budget)
 {
 	struct fbnic_napi_vector *nv = container_of(napi,
@@ -1484,6 +1669,9 @@ static int fbnic_poll(struct napi_struct *napi, int budget)
 	for (j = 0; j < nv->rxt_count; j++, i++)
 		work_done += fbnic_clean_rcq(nv, &nv->qt[i], budget);
 
+	if (budget && !fbnic_xsk_tx(nv))
+		work_done = budget;
+
 	if (work_done >= budget)
 		return budget;
 
@@ -1650,6 +1838,16 @@ void fbnic_free_napi_vectors(struct fbnic_net *fbn)
 			fbnic_free_napi_vector(fbn, fbn->napi[i]);
 }
 
+static u8 fbnic_bdq_page_shift(u32 page_size)
+{
+	return ilog2(page_size / FBNIC_BD_PAGE_SIZE);
+}
+
+static void fbnic_bdq_set_page_size(struct fbnic_ring *bdq, u32 page_size)
+{
+	bdq->bd_page_shift = fbnic_bdq_page_shift(page_size);
+}
+
 /* No second header fits behind the first in a whole buffer */
 static u32 fbnic_whole_hdr_max(u32 hroom)
 {
@@ -1657,10 +1855,39 @@ static u32 fbnic_whole_hdr_max(u32 hroom)
 	       FBNIC_RX_TROOM - FBNIC_RX_PAD;
 }
 
+int fbnic_xsk_validate(const struct xsk_rx_check *check,
+		       struct bpf_prog *prog, struct netlink_ext_ack *extack)
+{
+	if (check->buf_size != FBNIC_BD_PAGE_SIZE) {
+		NL_SET_ERR_MSG_MOD(extack, "AF_XDP chunks must be 4 KiB");
+		return -EOPNOTSUPP;
+	}
+	if (check->headroom > FBNIC_RX_HROOM_MAX ||
+	    !IS_ALIGNED(check->headroom, FBNIC_RX_HDR_ALIGN)) {
+		NL_SET_ERR_MSG_MOD(extack, "AF_XDP headroom is not supported");
+		return -EOPNOTSUPP;
+	}
+
+	/* Frames of the MTU, with two VLAN tags, fit one header buffer */
+	if (check->mtu + ETH_HLEN + 2 * VLAN_HLEN <=
+	    fbnic_whole_hdr_max(check->headroom))
+		return 0;
+
+	if (!check->sg || (prog && !prog->aux->xdp_has_frags)) {
+		NL_SET_ERR_MSG_MOD(extack,
+				   "MTU needs XDP_USE_SG, and frags in an XDP program");
+		return -EOPNOTSUPP;
+	}
+
+	return 0;
+}
+
 static int
 fbnic_alloc_qt_page_pools(struct fbnic_net *fbn, struct fbnic_q_triad *qt,
-			  unsigned int rxq_idx, u32 rx_page_size)
+			  unsigned int rxq_idx,
+			  const struct netdev_queue_config *qcfg)
 {
+	u32 rx_page_size = qcfg->rx_page_size;
 	struct page_pool_params pp_params = {
 		.order = 0,
 		.flags = PP_FLAG_DMA_MAP |
@@ -1674,8 +1901,29 @@ fbnic_alloc_qt_page_pools(struct fbnic_net *fbn, struct fbnic_q_triad *qt,
 		.netdev	= fbn->netdev,
 		.queue_idx = rxq_idx,
 	};
+	struct xsk_buff_pool *xsk;
 	struct page_pool *pp;
 
+	qt->rx_hroom = FBNIC_RX_HROOM;
+	qt->rx_frame_off = 0;
+	if (qcfg->rx_headroom) {
+		qt->rx_hroom = qcfg->rx_headroom;
+		qt->rx_frame_off = qcfg->rx_headroom - XDP_PACKET_HEADROOM;
+	}
+	qt->rx_hdr_first = ALIGN(qt->rx_hroom, FBNIC_RX_HDR_ALIGN);
+
+	/* Payload buffers come from UMEM only for multi-buffer sockets */
+	xsk = xsk_rxq_pool(fbn->netdev, rxq_idx);
+	qt->xsk_pool = xsk;
+	if (xsk) {
+		pp_params.flags |= PP_FLAG_MEM_PROVIDER;
+		pp_params.rx_roles = PP_RX_ROLE_HEAD;
+		/* A chunk can be smaller than a page */
+		pp_params.max_len = rx_page_size;
+		if (xsk_pool_uses_sg(xsk))
+			pp_params.rx_roles |= PP_RX_ROLE_PAYLOAD;
+	}
+
 	/* Page pool cannot exceed a size of 32768. This doesn't limit the
 	 * pages on the ring but the number we can have cached waiting on
 	 * the next use.
@@ -1693,15 +1941,25 @@ fbnic_alloc_qt_page_pools(struct fbnic_net *fbn, struct fbnic_q_triad *qt,
 		return PTR_ERR(pp);
 
 	qt->sub0.page_pool = pp;
+	fbnic_bdq_set_page_size(&qt->sub0, xsk ? rx_page_size : PAGE_SIZE);
 	qt->hdr_max = 0;
 	if (!page_pool_supports_frag(pp))
-		qt->hdr_max = fbnic_whole_hdr_max(FBNIC_RX_HROOM);
+		qt->hdr_max = fbnic_whole_hdr_max(qt->rx_hroom);
 	if (netif_rxq_has_unreadable_mp(fbn->netdev, rxq_idx)) {
 		pp_params.order = get_order(rx_page_size);
 		pp_params.max_len = rx_page_size;
 		pp_params.flags |= PP_FLAG_ALLOW_UNREADABLE_NETMEM;
 		pp_params.dma_dir = DMA_FROM_DEVICE;
 
+		pp = page_pool_create(&pp_params);
+		if (IS_ERR(pp))
+			goto err_destroy_sub0;
+	} else if (xsk && !xsk_pool_uses_sg(xsk)) {
+		pp_params.flags &= ~PP_FLAG_MEM_PROVIDER;
+		pp_params.rx_roles = 0;
+		pp_params.max_len = PAGE_SIZE;
+		rx_page_size = PAGE_SIZE;
+
 		pp = page_pool_create(&pp_params);
 		if (IS_ERR(pp))
 			goto err_destroy_sub0;
@@ -1709,6 +1967,7 @@ fbnic_alloc_qt_page_pools(struct fbnic_net *fbn, struct fbnic_q_triad *qt,
 		page_pool_get(pp);
 	}
 	qt->sub1.page_pool = pp;
+	fbnic_bdq_set_page_size(&qt->sub1, rx_page_size);
 
 	return 0;
 
@@ -1717,16 +1976,6 @@ fbnic_alloc_qt_page_pools(struct fbnic_net *fbn, struct fbnic_q_triad *qt,
 	return PTR_ERR(pp);
 }
 
-static u8 fbnic_bdq_page_shift(u32 page_size)
-{
-	return ilog2(page_size / FBNIC_BD_PAGE_SIZE);
-}
-
-static void fbnic_bdq_set_page_size(struct fbnic_ring *bdq, u32 page_size)
-{
-	bdq->bd_page_shift = fbnic_bdq_page_shift(page_size);
-}
-
 static void fbnic_ring_init(struct fbnic_ring *ring, u32 __iomem *doorbell,
 			    int q_idx, u8 flags)
 {
@@ -1830,7 +2079,8 @@ static int fbnic_alloc_napi_vector(struct fbnic_dev *fbd, struct fbnic_net *fbn,
 		if (xdp_count > 0) {
 			unsigned int xdp_idx = FBNIC_MAX_TXQS + rxq_idx;
 
-			fbnic_ring_init(&qt->sub1, db, xdp_idx, flags);
+			fbnic_ring_init(&qt->sub1, db, xdp_idx,
+					flags | FBNIC_RING_F_XDP);
 			fbn->tx[xdp_idx] = &qt->sub1;
 			xdp_count--;
 		} else {
@@ -1977,6 +2227,9 @@ static int fbnic_alloc_tx_ring_buffer(struct fbnic_ring *txr)
 {
 	size_t size = array_size(sizeof(*txr->tx_buf), txr->size_mask + 1);
 
+	if (txr->flags & FBNIC_RING_F_XDP)
+		size = array_size(sizeof(*txr->xdp_buf), txr->size_mask + 1);
+
 	txr->tx_buf = kvzalloc(size, GFP_KERNEL | __GFP_NOWARN);
 
 	return txr->tx_buf ? 0 : -ENOMEM;
@@ -2126,20 +2379,19 @@ static int fbnic_alloc_tx_qt_resources(struct fbnic_net *fbn,
 	return err;
 }
 
-static int fbnic_alloc_rx_qt_resources(struct fbnic_net *fbn,
-				       struct fbnic_napi_vector *nv,
-				       struct fbnic_q_triad *qt,
-				       u32 rx_page_size)
+static int
+fbnic_alloc_rx_qt_resources(struct fbnic_net *fbn,
+			    struct fbnic_napi_vector *nv,
+			    struct fbnic_q_triad *qt,
+			    const struct netdev_queue_config *qcfg)
 {
 	struct device *dev = fbn->netdev->dev.parent;
 	int err;
 
-	err = fbnic_alloc_qt_page_pools(fbn, qt, qt->cmpl.q_idx, rx_page_size);
+	err = fbnic_alloc_qt_page_pools(fbn, qt, qt->cmpl.q_idx, qcfg);
 	if (err)
 		return err;
 
-	fbnic_bdq_set_page_size(&qt->sub1, rx_page_size);
-
 	err = xdp_rxq_info_reg(&qt->xdp_rxq, fbn->netdev, qt->cmpl.q_idx,
 			       nv->napi.napi_id);
 	if (err)
@@ -2203,8 +2455,7 @@ static int fbnic_alloc_nv_resources(struct fbnic_net *fbn,
 		struct netdev_queue_config qcfg;
 
 		netdev_queue_config(fbn->netdev, nv->qt[i].cmpl.q_idx, &qcfg);
-		err = fbnic_alloc_rx_qt_resources(fbn, nv, &nv->qt[i],
-						  qcfg.rx_page_size);
+		err = fbnic_alloc_rx_qt_resources(fbn, nv, &nv->qt[i], &qcfg);
 		if (err)
 			goto free_qt_resources;
 	}
@@ -2598,9 +2849,8 @@ static void fbnic_nv_fill(struct fbnic_napi_vector *nv)
 	for (j = 0, t = nv->txt_count; j < nv->rxt_count; j++, t++) {
 		struct fbnic_q_triad *qt = &nv->qt[t];
 
-		/* Populate the header and payload BDQs */
-		fbnic_fill_bdq(&qt->sub0);
-		fbnic_fill_bdq(&qt->sub1);
+		/* Populate the header and payload BDQs; NAPI retries */
+		fbnic_fill_bdqs(qt);
 	}
 }
 
@@ -2736,14 +2986,20 @@ static void fbnic_config_drop_mode_rcq(struct fbnic_napi_vector *nv,
 				       struct fbnic_ring *rcq, bool tx_pause,
 				       bool hdr_split)
 {
+	struct fbnic_q_triad *qt = container_of_const(rcq, struct fbnic_q_triad,
+						     cmpl);
 	struct fbnic_net *fbn = netdev_priv(nv->napi.dev);
-	u32 drop_mode, rcq_ctl;
+	u32 drop_mode, hroom, rcq_ctl;
+
+	/* The device aligns the maximum value up to FBNIC_RX_HROOM_MAX. */
+	hroom = min_t(u32, qt->rx_hroom,
+		      FIELD_MAX(FBNIC_QUEUE_RDE_CTL0_MIN_HROOM_MASK));
 
 	drop_mode = fbnic_drop_mode(fbn, tx_pause);
 
 	/* Specify packet layout */
 	rcq_ctl = FIELD_PREP(FBNIC_QUEUE_RDE_CTL0_DROP_MODE_MASK, drop_mode) |
-	    FIELD_PREP(FBNIC_QUEUE_RDE_CTL0_MIN_HROOM_MASK, FBNIC_RX_HROOM) |
+	    FIELD_PREP(FBNIC_QUEUE_RDE_CTL0_MIN_HROOM_MASK, hroom) |
 	    FIELD_PREP(FBNIC_QUEUE_RDE_CTL0_MIN_TROOM_MASK, FBNIC_RX_TROOM) |
 	    FIELD_PREP(FBNIC_QUEUE_RDE_CTL0_EN_HDR_SPLIT, hdr_split);
 
@@ -3028,8 +3284,7 @@ static int fbnic_queue_mem_alloc(struct net_device *dev,
 	struct fbnic_napi_vector *nv;
 
 	if (!netif_running(dev))
-		return fbnic_alloc_qt_page_pools(fbn, qt, idx,
-						 qcfg->rx_page_size);
+		return fbnic_alloc_qt_page_pools(fbn, qt, idx, qcfg);
 
 	/* A failed PCIe recovery or resume can leave the datapath torn down
 	 * while netif_running() is still true.  This ndo runs before
@@ -3049,7 +3304,7 @@ static int fbnic_queue_mem_alloc(struct net_device *dev,
 	fbnic_ring_init(&qt->cmpl, real->cmpl.doorbell, real->cmpl.q_idx,
 			real->cmpl.flags);
 
-	return fbnic_alloc_rx_qt_resources(fbn, nv, qt, qcfg->rx_page_size);
+	return fbnic_alloc_rx_qt_resources(fbn, nv, qt, qcfg);
 }
 
 static void fbnic_default_qcfg(struct net_device *dev,
@@ -3086,6 +3341,14 @@ static int fbnic_validate_qcfg(struct net_device *dev,
 		return -EINVAL;
 	}
 
+	if (qcfg->rx_headroom &&
+	    (qcfg->rx_headroom < XDP_PACKET_HEADROOM ||
+	     qcfg->rx_headroom > FBNIC_RX_HROOM_MAX ||
+	     !IS_ALIGNED(qcfg->rx_headroom, FBNIC_RX_HDR_ALIGN))) {
+		NL_SET_ERR_MSG_MOD(extack, "rx_headroom is not supported");
+		return -EINVAL;
+	}
+
 	/* Payload fragments occupy multiples of FBNIC_RX_PAYLD_ALIGN bytes.
 	 * Keep at least one reference in the bias until fbnic_clean_bdq()
 	 * observes a completion from a subsequent allocation.
@@ -3216,5 +3479,6 @@ const struct netdev_queue_mgmt_ops fbnic_queue_mgmt_ops = {
 	.ndo_queue_stop		= fbnic_queue_stop,
 	.ndo_default_qcfg	= fbnic_default_qcfg,
 	.ndo_validate_qcfg	= fbnic_validate_qcfg,
-	.supported_params	= QCFG_RX_PAGE_SIZE,
+	.supported_params	= QCFG_RX_PAGE_SIZE | QCFG_RX_HEADROOM,
+	.xsk_page_pool		= true,
 };
diff --git a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.h b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.h
index 3077d9bbd587..b4baaa914585 100644
--- a/drivers/net/ethernet/meta/fbnic/fbnic_txrx.h
+++ b/drivers/net/ethernet/meta/fbnic/fbnic_txrx.h
@@ -14,7 +14,11 @@
 
 #include "fbnic_csr.h"
 
+struct bpf_prog;
 struct fbnic_net;
+struct netlink_ext_ack;
+struct xsk_buff_pool;
+struct xsk_rx_check;
 
 /* Guarantee we have space needed for storing the buffer
  * To store the buffer we need:
@@ -36,6 +40,8 @@ struct fbnic_net;
  */
 #define FBNIC_MAX_RX_PKT_DESC	7
 #define FBNIC_RX_DESC_MIN	roundup_pow_of_two(FBNIC_MAX_RX_PKT_DESC * 2)
+/* All descriptors of a packet but the two metadata ones */
+#define FBNIC_MAX_RX_BUFS	(FBNIC_MAX_RX_PKT_DESC - 2)
 
 #define FBNIC_MAX_TXQS			128u
 #define FBNIC_MAX_RXQS			128u
@@ -66,6 +72,9 @@ struct fbnic_net;
 	(ALIGN(FBNIC_RX_TROOM + FBNIC_RX_HROOM_PAD, 128) - FBNIC_RX_TROOM)
 #define FBNIC_RX_PAD			0
 #define FBNIC_RX_HDR_ALIGN		128
+#define FBNIC_RX_HROOM_MAX \
+	ALIGN(FIELD_MAX(FBNIC_QUEUE_RDE_CTL0_MIN_HROOM_MASK), \
+	      FBNIC_RX_HDR_ALIGN)
 #define FBNIC_RX_PAYLD_ALIGN		128
 #define FBNIC_RX_PAYLD_OFFSET		0
 #define FBNIC_RX_PAYLD_PG_CL		0
@@ -73,6 +82,7 @@ struct fbnic_net;
 #define FBNIC_RING_F_DISABLED		BIT(0)
 #define FBNIC_RING_F_CTX		BIT(1)
 #define FBNIC_RING_F_STATS		BIT(2)	/* Ring's stats may be used */
+#define FBNIC_RING_F_XDP		BIT(3)	/* XDP TX ring */
 
 #define FBNIC_HDS_THRESH_MAX \
 	(4096 - FBNIC_RX_HROOM - FBNIC_RX_TROOM - FBNIC_RX_PAD)
@@ -121,12 +131,18 @@ struct fbnic_rx_buf {
 	long pagecnt_bias;
 };
 
+/* XDP TX ring entry: an XDP_TX buffer, or an AF_XDP TX descriptor */
+struct fbnic_xdp_tx_buf {
+	netmem_ref netmem;
+	struct xsk_buff_pool *xsk;
+};
+
 struct fbnic_ring {
 	/* Pointer to buffer specific info */
 	union {
 		struct fbnic_pkt_buff *pkt;	/* RCQ */
 		struct fbnic_rx_buf *rx_buf;	/* BDQ */
-		netmem_ref *tx_netmem;		/* TWQ1 */
+		struct fbnic_xdp_tx_buf *xdp_buf; /* TWQ1 */
 		void **tx_buf;			/* TWQ */
 		void *buffer;			/* Generic pointer */
 	};
@@ -167,6 +183,10 @@ struct fbnic_q_triad {
 	struct fbnic_ring sub0, sub1, cmpl;
 	struct xdp_rxq_info xdp_rxq;
 	u16 hdr_max;	/* Header bytes for whole header buffers, or 0 */
+	u16 rx_hroom;	/* Headroom before a header */
+	u16 rx_hdr_first; /* Offset of the first header in a buffer */
+	u16 rx_frame_off; /* Offset of the frame in a header buffer */
+	struct xsk_buff_pool *xsk_pool;	/* AF_XDP pool of an RX queue */
 };
 
 struct fbnic_napi_vector {
@@ -178,6 +198,7 @@ struct fbnic_napi_vector {
 	u16 v_idx;
 	u8 txt_count;
 	u8 rxt_count;
+	u8 xsk_tx_next;		/* RX queue that sends socket TX first */
 
 	struct fbnic_q_triad qt[];
 };
@@ -203,6 +224,8 @@ void fbnic_aggregate_ring_tx_counters(struct fbnic_net *fbn,
 void fbnic_aggregate_ring_xdp_counters(struct fbnic_net *fbn,
 				       struct fbnic_ring *xdpr);
 
+int fbnic_xsk_validate(const struct xsk_rx_check *check,
+		       struct bpf_prog *prog, struct netlink_ext_ack *extack);
 int fbnic_alloc_napi_vectors(struct fbnic_net *fbn);
 void fbnic_free_napi_vectors(struct fbnic_net *fbn);
 int fbnic_alloc_resources(struct fbnic_net *fbn);
-- 
2.56.0

Re: [PATCH net-next v1 04/11] xsk: Allocate UMEM chunks through page_pool

[Magnus Karlsson]

On Tue, 10 Nov 2026 at 12:26, Björn Töpel <bjorn@kernel.org> wrote:
>
> AF_XDP zero copy needs a second RX allocator in every driver. A
> page_pool driver can take UMEM chunks from a memory provider instead,
> and keep its refill, recycling and DMA sync code.
>
> Each aligned chunk becomes a net_iov, whose address netmem_address()
> finds without calling the provider. Userspace may post a chunk that
> the kernel still owns, so the provider tracks the owner of each chunk
> and drops such FILL entries. Chunks that a page pool releases stay in
> the kernel for reuse. Nothing installs the provider yet.
>
> Signed-off-by: Björn Töpel <bjorn@kernel.org>
> ---
>  include/net/netmem.h |  25 ++++-
>  include/net/xdp.h    |   2 +-
>  net/xdp/Kconfig      |   1 +
>  net/xdp/Makefile     |   2 +-
>  net/xdp/xsk.h        |   6 ++
>  net/xdp/xsk_pp.c     | 246 +++++++++++++++++++++++++++++++++++++++++++
>  6 files changed, 279 insertions(+), 3 deletions(-)
>  create mode 100644 net/xdp/xsk_pp.c
>
> diff --git a/include/net/netmem.h b/include/net/netmem.h
> index b986ad616334..a3b67ef32b21 100644
> --- a/include/net/netmem.h
> +++ b/include/net/netmem.h
> @@ -69,6 +69,7 @@ DECLARE_STATIC_KEY_FALSE(page_pool_restricted);
>  enum net_iov_type {
>         NET_IOV_DMABUF,
>         NET_IOV_IOURING,
> +       NET_IOV_XSK,
>  };
>
>  /* A memory descriptor representing abstract networking I/O vectors.
> @@ -107,6 +108,10 @@ struct net_iov_area {
>         size_t num_niovs;
>         /* Each net_iov covers 1 << niov_shift bytes. */
>         unsigned int niov_shift;
> +       /* CPU address of the first net_iov, or NULL if the CPU cannot read
> +        * the area.
> +        */
> +       void *vaddr;
>  };
>
>  static inline struct net_iov_area *net_iov_owner(const struct net_iov *niov)
> @@ -119,6 +124,18 @@ static inline unsigned int net_iov_idx(const struct net_iov *niov)
>         return niov - net_iov_owner(niov)->niovs;
>  }
>
> +static inline void *net_iov_address(const struct net_iov *niov)
> +{
> +       const struct net_iov_area *area = net_iov_owner(niov);
> +       unsigned long off;
> +
> +       if (!area->vaddr)
> +               return NULL;
> +
> +       off = (unsigned long)net_iov_idx(niov) << area->niov_shift;
> +       return area->vaddr + off;
> +}
> +
>  /* Initialize a niov: stamp the owning area, the memory provider type.
>   */
>  static inline void net_iov_init(struct net_iov *niov,
> @@ -345,10 +362,16 @@ static inline void *__netmem_address(netmem_ref netmem)
>         return page_address(__netmem_to_page(netmem));
>  }
>
> +/**
> + * netmem_address - get the CPU address of the memory behind @netmem
> + * @netmem: netmem to get the address of
> + *
> + * Return: the address, or NULL if the CPU cannot read @netmem.
> + */
>  static inline void *netmem_address(netmem_ref netmem)
>  {
>         if (netmem_is_net_iov(netmem))
> -               return NULL;
> +               return net_iov_address(netmem_to_net_iov(netmem));
>
>         return __netmem_address(netmem);
>  }
> diff --git a/include/net/xdp.h b/include/net/xdp.h
> index 1105bb9e3292..5ab7802ac4e1 100644
> --- a/include/net/xdp.h
> +++ b/include/net/xdp.h
> @@ -368,7 +368,7 @@ static inline bool xdp_buff_add_frag(struct xdp_buff *xdp, netmem_ref netmem,
>
>         if (unlikely(netmem_is_pfmemalloc(netmem)))
>                 xdp_buff_set_frag_pfmemalloc(xdp);
> -       if (unlikely(netmem_is_net_iov(netmem)))
> +       if (unlikely(netmem_is_net_iov(netmem) && !netmem_address(netmem)))
>                 xdp_buff_set_frag_unreadable(xdp);
>
>         return true;
> diff --git a/net/xdp/Kconfig b/net/xdp/Kconfig
> index 71af2febe72a..c9c68d3b3712 100644
> --- a/net/xdp/Kconfig
> +++ b/net/xdp/Kconfig
> @@ -2,6 +2,7 @@
>  config XDP_SOCKETS
>         bool "XDP sockets"
>         depends on BPF_SYSCALL
> +       select PAGE_POOL
>         default n
>         help
>           XDP sockets allows a channel between XDP programs and
> diff --git a/net/xdp/Makefile b/net/xdp/Makefile
> index 30cdc4315f42..6e9d60f9f7f1 100644
> --- a/net/xdp/Makefile
> +++ b/net/xdp/Makefile
> @@ -1,4 +1,4 @@
>  # SPDX-License-Identifier: GPL-2.0-only
>  obj-$(CONFIG_XDP_SOCKETS) += xsk.o xdp_umem.o xsk_queue.o xskmap.o
> -obj-$(CONFIG_XDP_SOCKETS) += xsk_buff_pool.o
> +obj-$(CONFIG_XDP_SOCKETS) += xsk_buff_pool.o xsk_pp.o
>  obj-$(CONFIG_XDP_SOCKETS_DIAG) += xsk_diag.o
> diff --git a/net/xdp/xsk.h b/net/xdp/xsk.h
> index 7c811b5cce76..4dab7b9a1487 100644
> --- a/net/xdp/xsk.h
> +++ b/net/xdp/xsk.h
> @@ -34,6 +34,12 @@ static inline struct xdp_sock *xdp_sk(struct sock *sk)
>
>  void xsk_map_try_sock_delete(struct xsk_map *map, struct xdp_sock *xs,
>                              struct xdp_sock __rcu **map_entry);
> +struct xsk_pp;

How about a header file instead of a forward declaration?

> +extern const struct memory_provider_ops xsk_pp_ops;
> +
> +struct xsk_pp *xsk_pp_create(struct xsk_buff_pool *pool);
> +void xsk_pp_put(struct xsk_pp *pp);
>  void xsk_clear_pool_at_qid(struct net_device *dev, u16 queue_id);
>  int xsk_reg_pool_at_qid(struct net_device *dev, struct xsk_buff_pool *pool,
>                         u16 queue_id);
> diff --git a/net/xdp/xsk_pp.c b/net/xdp/xsk_pp.c
> new file mode 100644
> index 000000000000..fdd55265a6cd
> --- /dev/null
> +++ b/net/xdp/xsk_pp.c
> @@ -0,0 +1,246 @@
> +// SPDX-License-Identifier: GPL-2.0
> +/* AF_XDP UMEM as a page pool memory provider
> + *
> + * Each aligned UMEM chunk is a readable net_iov with one owner at a time:
> + *
> + *   userspace --FILL--> page pool --release--> reuse --alloc--> page pool
> + *       ^                   |
> + *       +-----RX ring-------+
> + */
> +
> +#include <net/page_pool/helpers.h>
> +#include <net/page_pool/memory_provider.h>
> +#include <net/xdp_sock_drv.h>
> +
> +#include "xsk.h"
> +#include "xsk_queue.h"
> +
> +enum xsk_pp_chunk {
> +       XSK_PP_USER,    /* owned by userspace */
> +       XSK_PP_POOL,    /* owned by a page pool */
> +       XSK_PP_REUSE,   /* on the reuse stack */
> +};
> +
> +struct xsk_pp {
> +       struct net_iov_area area;
> +       struct xsk_buff_pool *pool;
> +       /* Protects fq, state and reuse */
> +       spinlock_t lock;
> +       struct xsk_queue *fq;
> +       refcount_t refs;
> +       u32 nr_pools;
> +       bool dma_need_sync;
> +       u8 *state;
> +       u32 reuse_cnt;
> +       u32 reuse[];

Is the reason that you have a reuse stack here and not use the one in
xsk_buff_pool that it only exists in unaligned mode, which you are not
supporting? Could you still use the "pre-calculated" heads in aligned
mode and link them up? There is already a xskb_list and free_list in
there that is used for that purpose. Just an idea that could cut down
the state and maybe the amount of code.

> +};
> +
> +static void xsk_pp_free(struct xsk_pp *pp)
> +{
> +       kvfree(pp->state);
> +       kvfree(pp->area.niovs);
> +       kvfree(pp);
> +}
> +
> +void xsk_pp_put(struct xsk_pp *pp)
> +{
> +       if (refcount_dec_and_test(&pp->refs))
> +               xsk_pp_free(pp);
> +}
> +
> +/* @pool must be DMA mapped and have aligned chunks. The provider must be
> + * the only consumer of its FILL ring.
> + */
> +struct xsk_pp *xsk_pp_create(struct xsk_buff_pool *pool)
> +{
> +       struct net_iov *niov;
> +       struct xsk_pp *pp;
> +       dma_addr_t dma;
> +       u32 i;
> +
> +       pp = kvzalloc_flex(*pp, reuse, pool->heads_cnt, GFP_KERNEL_ACCOUNT);
> +       if (!pp)
> +               return ERR_PTR(-ENOMEM);
> +
> +       pp->area.niovs = kvzalloc_objs(*pp->area.niovs, pool->heads_cnt,
> +                                      GFP_KERNEL_ACCOUNT);
> +       pp->state = kvzalloc(pool->heads_cnt, GFP_KERNEL_ACCOUNT);
> +       if (!pp->area.niovs || !pp->state) {
> +               xsk_pp_free(pp);
> +               return ERR_PTR(-ENOMEM);
> +       }
> +
> +       pp->area.num_niovs = pool->heads_cnt;
> +       pp->area.niov_shift = pool->chunk_shift;
> +       pp->area.vaddr = pool->addrs;
> +       pp->pool = pool;
> +       pp->fq = pool->fq;
> +       pp->dma_need_sync = dma_dev_need_sync(pool->dev);

Are you copying these for performance reasons, i.e. no extra pointer
dereference in the fast path? In the same manner as the pool copies
some information from the umem?

> +       spin_lock_init(&pp->lock);
> +       refcount_set(&pp->refs, 1);
> +
> +       for (i = 0; i < pp->area.num_niovs; i++) {
> +               niov = &pp->area.niovs[i];
> +               dma = xp_raw_get_dma(pool, (u64)i << pool->chunk_shift);
> +               net_iov_init(niov, &pp->area, NET_IOV_XSK);
> +               if (net_mp_niov_set_dma_addr(niov, dma)) {
> +                       xsk_pp_free(pp);
> +                       return ERR_PTR(-ERANGE);
> +               }
> +       }
> +
> +       return pp;
> +}
> +
> +static int xsk_pp_init(struct page_pool *pool)
> +{
> +       struct xsk_pp *pp = pool->mp_priv;
> +
> +       if (pool->p.order || pool->p.dma_dir != DMA_BIDIRECTIONAL ||
> +           pool->p.dev != pp->pool->dev)
> +               return -EINVAL;
> +       if (pool->p.max_len > pp->pool->chunk_size ||
> +           pool->p.offset > pp->pool->chunk_size - pool->p.max_len)
> +               return -EINVAL;
> +
> +       refcount_inc(&pp->refs);
> +       spin_lock_bh(&pp->lock);
> +       pp->nr_pools++;

What is the difference between this counter and the refcount?

> +       spin_unlock_bh(&pp->lock);
> +       return 0;
> +}
> +
> +static void xsk_pp_destroy(struct page_pool *pool)
> +{
> +       struct xsk_pp *pp = pool->mp_priv;
> +
> +       spin_lock_bh(&pp->lock);
> +       pp->nr_pools--;
> +       spin_unlock_bh(&pp->lock);
> +       xsk_pp_put(pp);
> +}
> +
> +static bool xsk_pp_claim(struct xsk_pp *pp, u32 idx, enum xsk_pp_chunk from)
> +{
> +       lockdep_assert_held(&pp->lock);
> +
> +       if (pp->state[idx] != from)
> +               return false;
> +
> +       pp->state[idx] = XSK_PP_POOL;
> +       return true;
> +}
> +
> +static u32 xsk_pp_alloc_reuse(struct xsk_pp *pp, netmem_ref *netmems, u32 max)
> +{
> +       u32 idx, n = 0;
> +
> +       while (n < max && pp->reuse_cnt) {
> +               idx = pp->reuse[--pp->reuse_cnt];
> +               if (WARN_ON_ONCE(!xsk_pp_claim(pp, idx, XSK_PP_REUSE)))
> +                       continue;
> +               netmems[n++] = net_iov_to_netmem(&pp->area.niovs[idx]);
> +       }
> +
> +       return n;
> +}
> +
> +static u32 xsk_pp_alloc_fill(struct xsk_pp *pp, netmem_ref *netmems, u32 max)
> +{
> +       struct xsk_buff_pool *pool = pp->pool;
> +       struct xsk_queue *fq = pp->fq;
> +       u32 i, entries, cons, idx, n = 0;
> +       u64 addr;
> +
> +       if (!fq || !max)
> +               return 0;
> +
> +       entries = xskq_cons_nb_entries(fq, max);
> +       if (!entries) {
> +               fq->queue_empty_descs++;
> +               return 0;
> +       }
> +
> +       cons = fq->cached_cons;
> +       for (i = 0; i < entries; i++) {
> +               __xskq_cons_read_addr_unchecked(fq, cons++, &addr);
> +               addr &= pool->chunk_mask;
> +               idx = addr >> pp->area.niov_shift;
> +               if (unlikely(addr >= pool->addrs_cnt ||
> +                            !xsk_pp_claim(pp, idx, XSK_PP_USER))) {
> +                       fq->invalid_descs++;
> +                       continue;
> +               }
> +               netmems[n++] = net_iov_to_netmem(&pp->area.niovs[idx]);
> +       }
> +
> +       xskq_cons_release_n(fq, entries);
> +       __xskq_cons_release(fq);
> +       return n;
> +}
> +
> +static netmem_ref xsk_pp_alloc_netmems(struct page_pool *pool, gfp_t gfp)
> +{
> +       netmem_ref *netmems = pool->alloc.cache;
> +       struct xsk_pp *pp = pool->mp_priv;
> +       struct net_iov *niov;
> +       dma_addr_t dma;
> +       u32 i, n;
> +
> +       if (WARN_ON_ONCE(pool->alloc.count))

What has the driver writer or system done if this ever happens?

> +               return 0;
> +
> +       spin_lock_bh(&pp->lock);

For my understanding, what entities can call this function at the same
time? With the xsk_buff_pool we were protected by NAPI.


> +       n = xsk_pp_alloc_reuse(pp, netmems, PP_ALLOC_CACHE_REFILL);
> +       n += xsk_pp_alloc_fill(pp, netmems + n, PP_ALLOC_CACHE_REFILL - n);
> +       spin_unlock_bh(&pp->lock);
> +       if (!n)
> +               return 0;
> +
> +       for (i = 0; i < n; i++) {
> +               niov = netmem_to_net_iov(netmems[i]);
> +               dma = page_pool_get_dma_addr_netmem(netmems[i]);
> +               if (pp->dma_need_sync)
> +                       dma_sync_single_range_for_device(pool->p.dev, dma,
> +                                                        pool->p.offset,
> +                                                        pool->p.max_len,
> +                                                        pool->p.dma_dir);
> +               net_mp_niov_set_page_pool(pool, niov);
> +       }
> +
> +       pool->alloc.count = n - 1;
> +       return netmems[n - 1];
> +}
> +
> +static bool xsk_pp_release_netmem(struct page_pool *pool, netmem_ref netmem)
> +{
> +       struct net_iov *niov = netmem_to_net_iov(netmem);
> +       struct xsk_pp *pp = pool->mp_priv;
> +       u32 idx = net_iov_idx(niov);
> +
> +       spin_lock_bh(&pp->lock);
> +       net_mp_niov_clear_page_pool(niov);
> +       if (!WARN_ON_ONCE(pp->state[idx] != XSK_PP_POOL ||
> +                         pp->reuse_cnt == pp->area.num_niovs)) {
> +               pp->state[idx] = XSK_PP_REUSE;
> +               pp->reuse[pp->reuse_cnt++] = idx;
> +       }
> +       spin_unlock_bh(&pp->lock);
> +
> +       return false;
> +}
> +
> +static int xsk_pp_nl_fill(void *mp_priv, struct sk_buff *rsp,
> +                         struct netdev_rx_queue *rxq)
> +{
> +       return 0;
> +}
> +
> +const struct memory_provider_ops xsk_pp_ops = {
> +       .init           = xsk_pp_init,
> +       .destroy        = xsk_pp_destroy,
> +       .alloc_netmems  = xsk_pp_alloc_netmems,
> +       .release_netmem = xsk_pp_release_netmem,
> +       .nl_fill        = xsk_pp_nl_fill,
> +       .caps           = MP_CAP_READABLE,
> +};
> --
> 2.56.0
>
>