[RFC 0/9] optimise zcrx refs cache bouncing

[RFC 0/9] optimise zcrx refs cache bouncing

[Pavel Begunkov]

zcrx works well when user space and NAPI run on the same CPU but loses a
chunk of performance otherwise. It's caused by cache bounces from
1) zcrx "user" refs, which track whether buffers are given to the user
2) niov refs, as zcrx bumps them on recv(), and hence
   skb_attempt_defer_free() doesn't help.

In this patchset, zcrx steals received skbs, delays their destruction
similar to skb_attempt_defer_free(), and processes them in
io_pp_zc_alloc_netmems(). This moves all aforementioned refs
modifications for the hot path to the NAPI context.

For the networking side the most interesting bits are patches 1 and 8,
and patch 6 around the call to tcp_read_sock_steal_skb(). I'm looking
to get opinions on whether tcp_read_sock_steal_skb() is fine or what
kind of helpers / API would work better.

Tested with liburing/examples/{zcrx + send-zerocopy},
200Gbit/s NICs, rx_page=4KB

before: MB/s=18948
CPU    %usr    %sys %iowait    %irq   %soft  %idle
  0    4.92   63.68    0.00    1.64    2.84  26.91
  7    0.00    0.00    0.00    0.30   89.50  10.20

after: MB/s=21034
CPU    %usr    %sys %iowait    %irq   %soft  %idle
  0    5.59   50.18    0.00    2.26    2.73  39.24
  7    0.00    0.00    0.00    0.20   87.49  12.31

Helps in a similar way to 32KB rx page size, and also improves numbers
when NAPI and user space run on the same CPU.

kernel:
url: https://github.com/isilence/linux/tree/zcrx/skb-stealing
git: https://github.com/isilence/linux.git zcrx/skb-stealing

liburing (can be used any other version):
url: https://github.com/isilence/liburing/tree/zcrx/test-skb-steal
git: https://github.com/isilence/liburing.git zcrx/test-skb-steal

Pavel Begunkov (9):
  net: allow __tcp_read_sock actors to steal skbs
  net: add provider specific net_iov field
  io_uring/zcrx: don't save/restore count for frag skbs
  io_uring/zcrx: split frag handling loop
  io_uring/zcrx: split io_zcrx_recv_frag()
  io_uring/zcrx: implement skb stealing
  io_uring/zcrx: don't lock for single producer ptr ring
  io_uring/zcrx: steal niov refs
  io_uring/zcrx: add rq_lock cache of "user" niov refs

 include/linux/net.h  |   1 +
 include/net/netmem.h |   1 +
 include/net/tcp.h    |  13 +++
 io_uring/zcrx.c      | 210 ++++++++++++++++++++++++++++++++++---------
 io_uring/zcrx.h      |   4 +
 net/ipv4/tcp.c       |  11 +++
 6 files changed, 200 insertions(+), 40 deletions(-)

-- 
2.54.0

[RFC 1/9] net: allow __tcp_read_sock actors to steal skbs

[Pavel Begunkov]

Currently __tcp_read_sock() owns skbs and expects them to be present
when the actor function returns (modulo collapsing). For zcrx
optimisations I want to be able to take ownership of the skb in the
callback, add a helper doing that. It's only implemented for tcp, hence
keep "tcp" in the helper name. It could be later extended to other
protocols but would need some whitelisting mechanism.

Signed-off-by: Pavel Begunkov <asml.silence@gmail.com>
---
 include/linux/net.h |  1 +
 include/net/tcp.h   | 13 +++++++++++++
 net/ipv4/tcp.c      | 11 +++++++++++
 3 files changed, 25 insertions(+)

diff --git a/include/linux/net.h b/include/linux/net.h
index f268f395ce47..ed882aeac4a5 100644
--- a/include/linux/net.h
+++ b/include/linux/net.h
@@ -165,6 +165,7 @@ typedef struct {
 		void *data;
 	} arg;
 	int error;
+	bool stolen;
 } read_descriptor_t;
 
 struct vm_area_struct;
diff --git a/include/net/tcp.h b/include/net/tcp.h
index ecbadcb3a744..3d25707b73c3 100644
--- a/include/net/tcp.h
+++ b/include/net/tcp.h
@@ -3089,6 +3089,19 @@ static inline int tcp_recv_should_stop(struct sock *sk)
 	       signal_pending(current);
 }
 
+static inline bool tcp_read_sock_steal_skb(read_descriptor_t *desc,
+					   struct sk_buff *skb,
+					   struct sock *sk)
+{
+	if (skb_shared(skb))
+		return false;
+
+	desc->stolen = true;
+	__skb_unlink(skb, &sk->sk_receive_queue);
+	skb_orphan(skb);
+	return true;
+}
+
 INDIRECT_CALLABLE_DECLARE(union tcp_seq_and_ts_off
 			  tcp_v4_init_seq_and_ts_off(const struct net *net,
 						     const struct sk_buff *skb));
diff --git a/net/ipv4/tcp.c b/net/ipv4/tcp.c
index 432fa28e47d4..309a0e6b0173 100644
--- a/net/ipv4/tcp.c
+++ b/net/ipv4/tcp.c
@@ -1677,6 +1677,7 @@ static int __tcp_read_sock(struct sock *sk, read_descriptor_t *desc,
 		return -ENOTCONN;
 	while ((skb = tcp_recv_skb(sk, seq, &offset)) != NULL) {
 		if (offset < skb->len) {
+			u8 tcp_flags = TCP_SKB_CB(skb)->tcp_flags;
 			int used;
 			size_t len;
 
@@ -1689,6 +1690,7 @@ static int __tcp_read_sock(struct sock *sk, read_descriptor_t *desc,
 				if (!len)
 					break;
 			}
+			desc->stolen = false;
 			used = recv_actor(desc, skb, offset, len);
 			if (used <= 0) {
 				if (!copied)
@@ -1701,6 +1703,14 @@ static int __tcp_read_sock(struct sock *sk, read_descriptor_t *desc,
 			copied += used;
 			offset += used;
 
+			if (desc->stolen) {
+				if (tcp_flags & TCPHDR_FIN) {
+					++seq;
+					break;
+				}
+				goto next;
+			}
+
 			/* If recv_actor drops the lock (e.g. TCP splice
 			 * receive) the skb pointer might be invalid when
 			 * getting here: tcp_collapse might have deleted it
@@ -1721,6 +1731,7 @@ static int __tcp_read_sock(struct sock *sk, read_descriptor_t *desc,
 			break;
 		}
 		tcp_eat_recv_skb(sk, skb);
+next:
 		if (!desc->count)
 			break;
 		WRITE_ONCE(*copied_seq, seq);
-- 
2.54.0

[RFC 2/9] net: add provider specific net_iov field

[Pavel Begunkov]

Use a hole in struct net_iov to give some extra space to memory
providers like zcrx. Keeping some extra info in net_iov itself helps
with cache utilisation.

Signed-off-by: Pavel Begunkov <asml.silence@gmail.com>
---
 include/net/netmem.h | 1 +
 1 file changed, 1 insertion(+)

diff --git a/include/net/netmem.h b/include/net/netmem.h
index bccacd21b6c3..a564c510b484 100644
--- a/include/net/netmem.h
+++ b/include/net/netmem.h
@@ -95,6 +95,7 @@ enum net_iov_type {
 struct net_iov {
 	struct netmem_desc desc;
 	enum net_iov_type type;
+	unsigned int mp_private;
 	struct net_iov_area *owner;
 };
 
-- 
2.54.0

[RFC 3/9] io_uring/zcrx: don't save/restore count for frag skbs

[Pavel Begunkov]

We save and restore desc->count before recursing for frag skb
processing. Extract the handling into a separate function instead and
pass a flag.

Signed-off-by: Pavel Begunkov <asml.silence@gmail.com>
---
 io_uring/zcrx.c | 22 ++++++++++++++--------
 1 file changed, 14 insertions(+), 8 deletions(-)

diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c
index 0aa6455971d6..816a169b848e 100644
--- a/io_uring/zcrx.c
+++ b/io_uring/zcrx.c
@@ -1827,9 +1827,8 @@ static int io_zcrx_recv_frag(struct io_kiocb *req, struct io_zcrx_ifq *ifq,
 	return len;
 }
 
-static int
-io_zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
-		 unsigned int offset, size_t len)
+static int __zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
+			   unsigned int offset, size_t len)
 {
 	struct io_zcrx_args *args = desc->arg.data;
 	struct io_zcrx_ifq *ifq = args->ifq;
@@ -1907,11 +1906,8 @@ io_zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
 		if (offset < frag_end) {
 			unsigned copy = min(frag_end - offset, len);
 			unsigned frag_off = offset - start;
-			size_t count;
 
-			count = desc->count;
-			ret = io_zcrx_recv_skb(desc, frag_iter, frag_off, copy);
-			desc->count = count;
+			ret = __zcrx_recv_skb(desc, frag_iter, frag_off, copy);
 			if (ret < 0)
 				goto out;
 
@@ -1926,10 +1922,20 @@ io_zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
 out:
 	if (offset == start_off)
 		return ret;
-	desc->count -= (offset - start_off);
 	return offset - start_off;
 }
 
+static
+int io_zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
+			unsigned int offset, size_t len)
+{
+	int ret;
+
+	ret = __zcrx_recv_skb(desc, skb, offset, len);
+	desc->count -= max(0, ret);
+	return ret;
+}
+
 static int io_zcrx_tcp_recvmsg(struct io_kiocb *req, struct io_zcrx_ifq *ifq,
 				struct sock *sk, int flags,
 				unsigned issue_flags, unsigned int *outlen)
-- 
2.54.0

[RFC 4/9] io_uring/zcrx: split frag handling loop

[Pavel Begunkov]

A preparation patch splitting the frag array handling loop into two,
where first we skip frags below the requested offset. It makes further
changes more readable.

Signed-off-by: Pavel Begunkov <asml.silence@gmail.com>
---
 io_uring/zcrx.c | 32 +++++++++++++++++++-------------
 1 file changed, 19 insertions(+), 13 deletions(-)

diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c
index 816a169b848e..162e67287916 100644
--- a/io_uring/zcrx.c
+++ b/io_uring/zcrx.c
@@ -1877,23 +1877,29 @@ static int __zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
 		const skb_frag_t *frag = &shi->frags[i];
 		unsigned frag_end = start + skb_frag_size(frag);
 
+		if (offset < frag_end)
+			break;
+		start = frag_end;
+	}
+
+	for (; i < shi->nr_frags; i++) {
+		const skb_frag_t *frag = &shi->frags[i];
+		unsigned frag_end = start + skb_frag_size(frag);
+		unsigned copy = min(frag_end - offset, len);
+		unsigned frag_off = offset - start;
+
 		if (WARN_ON(start > offset + len))
 			return -EFAULT;
+		start = frag_end;
 
-		if (offset < frag_end) {
-			unsigned copy = min(frag_end - offset, len);
-			unsigned frag_off = offset - start;
-
-			ret = io_zcrx_recv_frag(req, ifq, frag, frag_off, copy);
-			if (ret < 0)
-				goto out;
+		ret = io_zcrx_recv_frag(req, ifq, frag, frag_off, copy);
+		if (ret < 0)
+			goto out;
 
-			offset += ret;
-			len -= ret;
-			if (len == 0 || ret != copy)
-				goto out;
-		}
-		start = frag_end;
+		offset += ret;
+		len -= ret;
+		if (len == 0 || ret != copy)
+			goto out;
 	}
 
 	skb_walk_frags(skb, frag_iter) {
-- 
2.54.0

[RFC 5/9] io_uring/zcrx: split io_zcrx_recv_frag()

[Pavel Begunkov]

In preparation for having more elaborate reference counting for niovs,
split normal pages handling (copy path) out of io_zcrx_recv_frag() and
inline it into callers. Also move refcounting out of it.

Signed-off-by: Pavel Begunkov <asml.silence@gmail.com>
---
 io_uring/zcrx.c | 44 +++++++++++++++++++++++---------------------
 1 file changed, 23 insertions(+), 21 deletions(-)

diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c
index 162e67287916..80aa68ab9968 100644
--- a/io_uring/zcrx.c
+++ b/io_uring/zcrx.c
@@ -1800,30 +1800,16 @@ static int io_zcrx_copy_frag(struct io_kiocb *req, struct io_zcrx_ifq *ifq,
 	return ret;
 }
 
-static int io_zcrx_recv_frag(struct io_kiocb *req, struct io_zcrx_ifq *ifq,
-			     const skb_frag_t *frag, int off, int len)
+static int zcrx_recv_niov(struct io_kiocb *req, struct io_zcrx_ifq *ifq,
+				struct net_iov *niov, int off, int len)
 {
-	struct net_iov *niov;
-	struct page_pool *pp;
-
-	if (unlikely(!skb_frag_is_net_iov(frag)))
-		return io_zcrx_copy_frag(req, ifq, frag, off, len);
-
-	niov = netmem_to_net_iov(frag->netmem);
-	pp = niov->desc.pp;
+	struct page_pool *pp = niov->desc.pp;
 
 	if (!pp || pp->mp_ops != &io_uring_pp_zc_ops || io_pp_to_ifq(pp) != ifq)
 		return -EFAULT;
 
-	if (!io_zcrx_queue_cqe(req, niov, ifq, off + skb_frag_off(frag), len))
+	if (!io_zcrx_queue_cqe(req, niov, ifq, off, len))
 		return -ENOSPC;
-
-	/*
-	 * Prevent it from being recycled while user is accessing it.
-	 * It has to be done before grabbing a user reference.
-	 */
-	page_pool_ref_netmem(net_iov_to_netmem(niov));
-	io_zcrx_get_niov_uref(niov);
 	return len;
 }
 
@@ -1892,9 +1878,25 @@ static int __zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
 			return -EFAULT;
 		start = frag_end;
 
-		ret = io_zcrx_recv_frag(req, ifq, frag, frag_off, copy);
-		if (ret < 0)
-			goto out;
+		if (unlikely(!skb_frag_is_net_iov(frag))) {
+			ret = io_zcrx_copy_frag(req, ifq, frag, frag_off, copy);
+			if (ret < 0)
+				goto out;
+		} else {
+			struct net_iov *niov = netmem_to_net_iov(frag->netmem);
+
+			ret = zcrx_recv_niov(req, ifq, niov,
+					     frag_off + skb_frag_off(frag),
+					     copy);
+			if (ret < 0)
+				goto out;
+			/*
+			 * Prevent it from being recycled while user is accessing it.
+			 * It has to be done before grabbing a user reference.
+			 */
+			page_pool_ref_netmem(net_iov_to_netmem(niov));
+			io_zcrx_get_niov_uref(niov);
+		}
 
 		offset += ret;
 		len -= ret;
-- 
2.54.0

[RFC 6/9] io_uring/zcrx: implement skb stealing

[Pavel Begunkov]

One of major hotspots for zcrx is handing buffers to the user space and
getting them back from the refill queue. For each niov we keep an atomic
reference counter, which is incremented from the syscall path when we
give the buffer to the user space, and decremented from NAPI when
processing the refill queue on page pool allocation. When user space and
NAPI run on different CPUs it causes cache bouncing.

Instead of bumping the ref counter for each frag on receive, try to steal
the entire skb and send it to NAPI for zcrx to process it, so that put
and gets happen on the same CPU. We trade a bunch of atomics with cache
bouncing with a single ptr_ring produce / consume. It achieves same goals
and replaces skb_attempt_defer_free() but also improves locality for zcrx
specific accounting.

It still uses atomics for "user" counting in this patch, but now it's
accessed by a single CPU only when the optimisation works. It also
improves temporal locality as well as we delay grabbing reference,
however we need to make sure that skbs are processed before the
corresponding RQEs.

Signed-off-by: Pavel Begunkov <asml.silence@gmail.com>
---
 io_uring/zcrx.c | 112 +++++++++++++++++++++++++++++++++++++++++-------
 io_uring/zcrx.h |   3 ++
 2 files changed, 99 insertions(+), 16 deletions(-)

diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c
index 80aa68ab9968..80a26b5798d3 100644
--- a/io_uring/zcrx.c
+++ b/io_uring/zcrx.c
@@ -40,6 +40,8 @@
 
 #define IO_DMA_ATTR (DMA_ATTR_SKIP_CPU_SYNC | DMA_ATTR_WEAK_ORDERING)
 
+static void zcrx_release_skbs(struct io_zcrx_ifq *ifq);
+
 static inline u64 zcrx_area_id_to_token(u32 area_id)
 {
 	return (u64)area_id << IORING_ZCRX_AREA_SHIFT;
@@ -677,6 +679,9 @@ static void io_zcrx_ifq_free(struct io_zcrx_ifq *ifq)
 		return;
 	if (WARN_ON_ONCE(ifq->master_ctx))
 		return;
+	if (WARN_ON_ONCE(!__ptr_ring_empty(&ifq->skb_ring)))
+		return;
+
 
 	for (i = 0; i < ifq->nr_areas; i++)
 		io_zcrx_free_area(ifq, ifq->areas[i]);
@@ -685,6 +690,7 @@ static void io_zcrx_ifq_free(struct io_zcrx_ifq *ifq)
 	if (ifq->dev)
 		put_device(ifq->dev);
 
+	ptr_ring_cleanup(&ifq->skb_ring, NULL);
 	io_free_rbuf_ring(ifq);
 	free_uid(ifq->user);
 	mutex_destroy(&ifq->pp_lock);
@@ -755,6 +761,9 @@ static void io_zcrx_scrub(struct io_zcrx_ifq *ifq)
 {
 	int i;
 
+	scoped_guard(spinlock_bh, &ifq->rq.lock)
+		zcrx_release_skbs(ifq);
+
 	guard(mutex)(&ifq->pp_lock);
 	for (i = 0; i < ifq->nr_areas; i++)
 		io_zcrx_scrub_area(ifq, ifq->areas[i]);
@@ -1039,6 +1048,9 @@ int io_register_zcrx(struct io_ring_ctx *ctx,
 	ifq = io_zcrx_ifq_alloc(ctx);
 	if (!ifq)
 		return -ENOMEM;
+	ret = ptr_ring_init(&ifq->skb_ring, 1024, GFP_KERNEL_ACCOUNT);
+	if (ret < 0)
+		goto ifq_free;
 
 	ifq->notif_data = notif.user_data;
 	ifq->allowed_notif_mask = notif.type_mask;
@@ -1188,9 +1200,10 @@ static inline u32 __zcrx_rq_entries(struct zcrx_rq *rq)
 	return min(entries, rq->nr_entries);
 }
 
-static inline u32 zcrx_rq_entries(struct zcrx_rq *rq)
+static inline u32 zcrx_rq_entries(struct zcrx_rq *rq, struct io_zcrx_ifq *ifq)
 {
 	rq->cached_tail = smp_load_acquire(&rq->ring->tail);
+	zcrx_release_skbs(ifq);
 	return __zcrx_rq_entries(rq);
 }
 
@@ -1209,6 +1222,7 @@ static inline void zcrx_rq_iter_init(struct zcrx_rq_iter *it,
 }
 
 static inline bool zcrx_rq_iter_next(struct zcrx_rq_iter *it,
+				     struct io_zcrx_ifq *ifq,
 				     struct zcrx_rq *rq,
 				     struct io_uring_zcrx_rqe **rqe)
 {
@@ -1217,6 +1231,14 @@ static inline bool zcrx_rq_iter_next(struct zcrx_rq_iter *it,
 		if (it->flushed)
 			return false;
 		rq->cached_tail = smp_load_acquire(&rq->ring->tail);
+
+		/*
+		 * skbs carry user niov references from the syscall path,
+		 * process them first before refilling will try to put
+		 * them back down.
+		 */
+		zcrx_release_skbs(ifq);
+
 		it->rqes_left = min_t(unsigned, __zcrx_rq_entries(rq),
 				      ZCRX_REFILL_CAP);
 		it->flushed = true;
@@ -1270,6 +1292,42 @@ static bool zcrx_put_refill_niov(struct net_iov *niov, struct page_pool *pp,
 	return true;
 }
 
+static void zcrx_user_ref_frags(struct io_zcrx_ifq *ifq, struct sk_buff *skb,
+				unsigned start, unsigned nr)
+{
+	struct skb_shared_info *shi = skb_shinfo(skb);
+	unsigned i;
+
+	nr = min_t(unsigned, nr, shi->nr_frags);
+	for (i = start; i < nr; i++) {
+		const skb_frag_t *frag = &shi->frags[i];
+		struct net_iov *niov = netmem_to_net_iov(frag->netmem);
+
+		/*
+		 * Prevent it from being recycled while user is accessing it.
+		 * It has to be done before grabbing a user reference.
+		 */
+		page_pool_ref_netmem(net_iov_to_netmem(niov));
+		io_zcrx_get_niov_uref(niov);
+	}
+}
+
+static void zcrx_release_skbs(struct io_zcrx_ifq *ifq)
+{
+	while (1) {
+		struct sk_buff *skb = __ptr_ring_consume(&ifq->skb_ring);
+
+		if (!skb)
+			break;
+
+		zcrx_user_ref_frags(ifq, skb, 0, -1U);
+		if (skb->fclone != SKB_FCLONE_UNAVAILABLE)
+			__kfree_skb(skb);
+		else
+			__napi_kfree_skb(skb, SKB_CONSUMED);
+	}
+}
+
 static unsigned io_zcrx_ring_refill(struct page_pool *pp,
 				    struct io_zcrx_ifq *ifq,
 				    netmem_ref *netmems, unsigned to_alloc)
@@ -1285,7 +1343,7 @@ static unsigned io_zcrx_ring_refill(struct page_pool *pp,
 
 	zcrx_rq_iter_init(&it, rq);
 
-	while (allocated < to_alloc - 1 && zcrx_rq_iter_next(&it, rq, &rqe)) {
+	while (allocated < to_alloc - 1 && zcrx_rq_iter_next(&it, ifq, rq, &rqe)) {
 		struct net_iov *next_niov;
 
 		if (!io_parse_rqe(rqe, ifq, &next_niov))
@@ -1489,7 +1547,7 @@ static unsigned zcrx_parse_rq(netmem_ref *netmem_array, unsigned nr,
 	unsigned int mask = rq->nr_entries - 1;
 	unsigned int i;
 
-	nr = min(nr, zcrx_rq_entries(rq));
+	nr = min(nr, zcrx_rq_entries(rq, zcrx));
 	for (i = 0; i < nr; i++) {
 		struct io_uring_zcrx_rqe *rqe = zcrx_next_rqe(rq, mask);
 		struct net_iov *niov;
@@ -1814,7 +1872,7 @@ static int zcrx_recv_niov(struct io_kiocb *req, struct io_zcrx_ifq *ifq,
 }
 
 static int __zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
-			   unsigned int offset, size_t len)
+			   unsigned int offset, size_t len, bool frag_skb)
 {
 	struct io_zcrx_args *args = desc->arg.data;
 	struct io_zcrx_ifq *ifq = args->ifq;
@@ -1822,6 +1880,8 @@ static int __zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
 	struct sk_buff *frag_iter;
 	unsigned start, start_off = offset;
 	struct skb_shared_info *shi;
+	unsigned first_frag;
+	bool can_steal;
 	int i, ret = 0;
 
 	len = min_t(size_t, len, desc->count);
@@ -1868,6 +1928,8 @@ static int __zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
 		start = frag_end;
 	}
 
+	first_frag = i;
+
 	for (; i < shi->nr_frags; i++) {
 		const skb_frag_t *frag = &shi->frags[i];
 		unsigned frag_end = start + skb_frag_size(frag);
@@ -1881,7 +1943,7 @@ static int __zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
 		if (unlikely(!skb_frag_is_net_iov(frag))) {
 			ret = io_zcrx_copy_frag(req, ifq, frag, frag_off, copy);
 			if (ret < 0)
-				goto out;
+				break;
 		} else {
 			struct net_iov *niov = netmem_to_net_iov(frag->netmem);
 
@@ -1889,20 +1951,38 @@ static int __zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
 					     frag_off + skb_frag_off(frag),
 					     copy);
 			if (ret < 0)
-				goto out;
-			/*
-			 * Prevent it from being recycled while user is accessing it.
-			 * It has to be done before grabbing a user reference.
-			 */
-			page_pool_ref_netmem(net_iov_to_netmem(niov));
-			io_zcrx_get_niov_uref(niov);
+				break;
 		}
 
 		offset += ret;
 		len -= ret;
-		if (len == 0 || ret != copy)
-			goto out;
+		if (len == 0 || ret != copy) {
+			i++;
+			len = 0;
+			break;
+		}
+	}
+
+	if (start != offset || ret < 0) {
+		if (!skb_frags_readable(skb))
+			zcrx_user_ref_frags(ifq, skb, first_frag, i);
+		goto out;
+	}
+
+	can_steal = !skb_frags_readable(skb) && !skb_has_frag_list(skb) &&
+		    start_off == 0 && !frag_skb;
+
+	if (can_steal && !__ptr_ring_full(&ifq->skb_ring) &&
+	    tcp_read_sock_steal_skb(desc, skb, args->sock->sk)) {
+		ret = ptr_ring_produce(&ifq->skb_ring, skb);
+		if (ret) {
+			zcrx_user_ref_frags(ifq, skb, first_frag, i);
+			__kfree_skb(skb);
+		}
+		goto out;
 	}
+	if (!skb_frags_readable(skb))
+		zcrx_user_ref_frags(ifq, skb, first_frag, i);
 
 	skb_walk_frags(skb, frag_iter) {
 		unsigned frag_end;
@@ -1915,7 +1995,7 @@ static int __zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
 			unsigned copy = min(frag_end - offset, len);
 			unsigned frag_off = offset - start;
 
-			ret = __zcrx_recv_skb(desc, frag_iter, frag_off, copy);
+			ret = __zcrx_recv_skb(desc, frag_iter, frag_off, copy, true);
 			if (ret < 0)
 				goto out;
 
@@ -1939,7 +2019,7 @@ int io_zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
 {
 	int ret;
 
-	ret = __zcrx_recv_skb(desc, skb, offset, len);
+	ret = __zcrx_recv_skb(desc, skb, offset, len, false);
 	desc->count -= max(0, ret);
 	return ret;
 }
diff --git a/io_uring/zcrx.h b/io_uring/zcrx.h
index 05598f08eda0..7fc12e53c8a1 100644
--- a/io_uring/zcrx.h
+++ b/io_uring/zcrx.h
@@ -5,6 +5,7 @@
 #include <linux/io_uring_types.h>
 #include <linux/dma-buf.h>
 #include <linux/socket.h>
+#include <linux/ptr_ring.h>
 #include <net/page_pool/types.h>
 #include <net/net_trackers.h>
 
@@ -69,6 +70,8 @@ struct io_zcrx_ifq {
 	struct zcrx_rq			rq ____cacheline_aligned_in_smp;
 	spinlock_t			alloc_lock ____cacheline_aligned_in_smp;
 
+	struct ptr_ring			skb_ring;
+
 	u32				if_rxq;
 	struct device			*dev;
 	struct net_device		*netdev;
-- 
2.54.0

[RFC 7/9] io_uring/zcrx: don't lock for single producer ptr ring

[Pavel Begunkov]

Normally, there is just one io_uring instance using zcrx and all
receiving happens under its lock. In this case we can avoid grabbing the
ptr ring lock on the production side.

Signed-off-by: Pavel Begunkov <asml.silence@gmail.com>
---
 io_uring/zcrx.c | 6 +++++-
 io_uring/zcrx.h | 1 +
 2 files changed, 6 insertions(+), 1 deletion(-)

diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c
index 80a26b5798d3..3d5d5c9fd9a5 100644
--- a/io_uring/zcrx.c
+++ b/io_uring/zcrx.c
@@ -825,6 +825,7 @@ static int zcrx_export(struct io_ring_ctx *ctx, struct io_zcrx_ifq *ifq,
 	if (!mem_is_zero(ce, sizeof(*ce)))
 		return -EINVAL;
 
+	ifq->shared = true;
 	refcount_inc(&ifq->refs);
 	refcount_inc(&ifq->user_refs);
 
@@ -1974,7 +1975,10 @@ static int __zcrx_recv_skb(read_descriptor_t *desc, struct sk_buff *skb,
 
 	if (can_steal && !__ptr_ring_full(&ifq->skb_ring) &&
 	    tcp_read_sock_steal_skb(desc, skb, args->sock->sk)) {
-		ret = ptr_ring_produce(&ifq->skb_ring, skb);
+		if (ifq->shared)
+			ret = ptr_ring_produce(&ifq->skb_ring, skb);
+		else
+			ret = __ptr_ring_produce(&ifq->skb_ring, skb);
 		if (ret) {
 			zcrx_user_ref_frags(ifq, skb, first_frag, i);
 			__kfree_skb(skb);
diff --git a/io_uring/zcrx.h b/io_uring/zcrx.h
index 7fc12e53c8a1..5ff4dabb0f68 100644
--- a/io_uring/zcrx.h
+++ b/io_uring/zcrx.h
@@ -71,6 +71,7 @@ struct io_zcrx_ifq {
 	spinlock_t			alloc_lock ____cacheline_aligned_in_smp;
 
 	struct ptr_ring			skb_ring;
+	bool				shared;
 
 	u32				if_rxq;
 	struct device			*dev;
-- 
2.54.0

[RFC 8/9] io_uring/zcrx: steal niov refs

[Pavel Begunkov]

In zcrx_release_skbs(), we reference all niovs of an skb and then
immediately put them down. Optimise it by stealing the frags.

Signed-off-by: Pavel Begunkov <asml.silence@gmail.com>
---
 io_uring/zcrx.c | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c
index 3d5d5c9fd9a5..23669471a8f0 100644
--- a/io_uring/zcrx.c
+++ b/io_uring/zcrx.c
@@ -1317,11 +1317,22 @@ static void zcrx_release_skbs(struct io_zcrx_ifq *ifq)
 {
 	while (1) {
 		struct sk_buff *skb = __ptr_ring_consume(&ifq->skb_ring);
+		struct skb_shared_info *shi;
+		unsigned i;
 
 		if (!skb)
 			break;
 
-		zcrx_user_ref_frags(ifq, skb, 0, -1U);
+		shi = skb_shinfo(skb);
+		for (i = 0; i < shi->nr_frags; i++) {
+			const skb_frag_t *frag = &shi->frags[i];
+			struct net_iov *niov = netmem_to_net_iov(frag->netmem);
+
+			/* Take niov references the skb holds */
+			io_zcrx_get_niov_uref(niov);
+		}
+		shi->nr_frags = 0;
+
 		if (skb->fclone != SKB_FCLONE_UNAVAILABLE)
 			__kfree_skb(skb);
 		else
-- 
2.54.0

[RFC 9/9] io_uring/zcrx: add rq_lock cache of "user" niov refs

[Pavel Begunkov]

Now the "user" refs are acquired and released by NAPI/page pool for the
optimised path, cache them on the NAPI side and protect it by rq.lock.
Store it in net_iov as we'd be touching the cache line by refilling soon
anyway.

Signed-off-by: Pavel Begunkov <asml.silence@gmail.com>
---
 io_uring/zcrx.c | 27 ++++++++++++++++++++++++---
 1 file changed, 24 insertions(+), 3 deletions(-)

diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c
index 23669471a8f0..04a80d1a2b3a 100644
--- a/io_uring/zcrx.c
+++ b/io_uring/zcrx.c
@@ -393,9 +393,20 @@ static inline atomic_t *io_get_user_counter(struct net_iov *niov)
 
 static bool io_zcrx_put_niov_uref(struct net_iov *niov, unsigned refs)
 {
-	atomic_t *uref = io_get_user_counter(niov);
+	unsigned *cached_ref = &niov->mp_private;
+	atomic_t *uref;
 	int old;
 
+	lockdep_assert_held(&io_zcrx_iov_to_area(niov)->ifq->rq.lock);
+
+	if (likely(*cached_ref >= refs)) {
+		*cached_ref -= refs;
+		return true;
+	}
+	refs -= *cached_ref;
+	*cached_ref = 0;
+
+	uref = io_get_user_counter(niov);
 	old = atomic_read(uref);
 	do {
 		if (unlikely(old < refs))
@@ -744,6 +755,16 @@ static void io_zcrx_scrub_area(struct io_zcrx_ifq *ifq, struct io_zcrx_area *are
 {
 	int i;
 
+	scoped_guard(spinlock_bh, &ifq->rq.lock) {
+		for (i = 0; i < area->nia.num_niovs; i++) {
+			struct net_iov *niov = &area->nia.niovs[i];
+			unsigned *ref = &niov->mp_private;
+
+			atomic_add(*ref, &area->user_refs[i]);
+			*ref = 0;
+		}
+	}
+
 	/* Reclaim back all buffers given to the user space. */
 	for (i = 0; i < area->nia.num_niovs; i++) {
 		struct net_iov *niov = &area->nia.niovs[i];
@@ -1327,9 +1348,9 @@ static void zcrx_release_skbs(struct io_zcrx_ifq *ifq)
 		for (i = 0; i < shi->nr_frags; i++) {
 			const skb_frag_t *frag = &shi->frags[i];
 			struct net_iov *niov = netmem_to_net_iov(frag->netmem);
+			unsigned *ref = &niov->mp_private;
 
-			/* Take niov references the skb holds */
-			io_zcrx_get_niov_uref(niov);
+			*ref += 1;
 		}
 		shi->nr_frags = 0;
 
-- 
2.54.0