[PATCH v8 00/14] Improve the scalability of NFSD's classic DRC

[PATCH v8 00/14] Improve the scalability of NFSD's classic DRC

[Chuck Lever]

The NFSD Duplicate Reply Cache evicts cache entries older than 120
seconds (RC_EXPIRE). On an NFS server with more than 64 GB of
physical memory, during a steady-state workload the 256k-entry limit
on DRC size, and not RC_EXPIRE, governs retention of DRC entries.
Cache entries are thus evicted before a retransmit on a connected
transport can hit them, since the the default retransmit timeout for
such transports is 3 x 60 seconds.

To remedy this situation, this series modifies the DRC eviction
algorithm to make an entry eligible for eviction once the RPC
transport has seen the client peer acknowledge that entry's reply.
This approach is used in several other common NFS server
implementations today.

Each network connection publishes two positions: the reply it has
just sent and the point up to which the peer has acknowledged. The
DRC then compares only those two values to decide on eviction. UDP
entries keep the current time-based eviction.

TCP and RDMA transports provide acknowledgement semantics that are
precise to the byte. On a TCP-with-TLS connection, byte-level
acknowledgement is not available to consumers. Therefore the ACK
position advances only while no TLS record is open or parked on a
short send, and is withheld while decrypted data waits in the
receive path.

---
Changes in v8:
- Reject an AUTH_TLS probe sent inside a TLS session. (sashiko)
- Count a request again when a procedure defers it after its lookup. (sashiko)
- TLS: withhold the position while the strparser copies a record in. (sashiko)
- TCP: take the mark from the out-of-order queue when it is not empty.
- TCP: raise the mark to snd_wl1 to cover a wholly lost retransmit.
- TCP: move svc_tcp_acked_pos()'s kernel-doc block onto that function.
- svcrdma: correct the comment on why the position is not gated.
- TCP: test the request count before taking the socket lock.
- TCP: skip the position refresh for a service with no reply hook.
- Scan the whole bucket for acknowledged entries, not just the oldest.
- Link to v7: https://patch.msgid.link/20261029-duplicate-reply-cache-v7-0-f3b0f3cf5bc7@kernel.org

Changes in v7:
- Count requests in flight per transport, from first read to lookup.
- Withhold the acked position while another request is in flight.
- TCP: hold the position until copied_seq passes the rcv_nxt mark.
- TCP: read snd_una and rcv_nxt under the socket lock.
- Add the svcsock_tcp_acked tracepoint.
- Drop a call whose transport closed before its DRC lookup.
- Explain why svcrdma publishes its position without a receive gate.
- Narrow the reclaimability claim in the cap removal.
- Correct the collision-window claim in the checksum removal.
- New patch: advance the acked position on TLS connections.
- Link to v6: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-0-db5e13fd9944@kernel.org

Changes in v6:
- Make the DRC size limit independent of page size (Neil)
- Drop same-connection retransmits (Neil)
- Link to v5: https://patch.msgid.link/20260918-duplicate-reply-cache-v5-0-b6aba9ebf2f4@kernel.org

Changes in v5:
- Actually make xpt_id 64 bits wide (per Neil's review).
- Move the DRC cap removal first (per Neil's review).
- Rewrite the cap removal's description to stand on its own.
- Drop the 4x visit bound from the prune walk (per Neil's review).
- Evict acknowledged entries only where the prune walk reaches them.
- Restore the shrinker's unlimited per-bucket scan.
- Link to v4: https://patch.msgid.link/20260914-duplicate-reply-cache-v4-0-bcf9d2628a94@kernel.org

Changes in v4:
- Track reply positions instead of ack cookies (per Neil's review).
- Make xpt_id a 64-bit counter and drop the xarray (per Neil's review).
- Drop implied ACK and the last-request timestamp (per Neil's review).
- Keep an unsent reply until RC_EXPIRE (per Neil's review).
- Pin a DRC entry against eviction until its reply has been sent.
- svcrdma: Serialize Send posting so positions match completion order.
- Drop the nfsd_drc_reply_acked tracepoint.
- Link to v3: https://patch.msgid.link/20260910-duplicate-reply-cache-v3-0-31532a4c7449@kernel.org

Changes in v3:
- Add the reply-acknowledged callback and its TCP and RDMA reporters
  ahead of implied ACK, which now defers to a pending report.
- Split the prune-loop restructure into its own patch.
- Move the eviction tracepoints ahead of the ack patches; the
  implied-ACK event now lands with implied ACK.
- Keep err local to the pmap_register block in svc_setup_socket()
  (per Jeff's review).
- Never move xpt_last_recv backwards when nfsd threads race.
- Fold the XPT_ORDERED patch into its consumer (per Jeff's review).
- State the re-execution exposure of implied-ACK eviction instead of
  calling the DRC advisory (per Jeff's review).
- New patch: remove the DRC request checksum and payload_misses stat.
- New patch: remove the 256k-entry cap on the DRC size.
- Link to v2: https://patch.msgid.link/20260828-duplicate-reply-cache-v2-0-25069e660a7b@kernel.org

Changes in v2:
- Print the DRC eviction tracepoints' age field as unsigned.
- Link to v1: https://patch.msgid.link/20260826-duplicate-reply-cache-v1-0-b1d51e1af5c7@kernel.org

---
Chuck Lever (14):
      SUNRPC: Reject an AUTH_TLS probe inside a TLS session
      NFSD: Make the DRC size limit independent of page size
      NFSD: Remove hard cap on duplicate reply cache size
      SUNRPC: Assign a unique identifier to each svc_xprt
      NFSD: Track transport in DRC entries
      NFSD: Prepare bucket pruning for additional eviction reasons
      NFSD: Add tracepoints for DRC entry eviction
      NFSD: Record DRC population in lookup tracepoints
      SUNRPC: Publish reply positions for upper-layer consumers
      SUNRPC: Publish TCP reply positions
      svcrdma: Publish RDMA reply positions
      NFSD: Evict acknowledged DRC entries
      NFSD: Remove DRC checksum and payload_misses stat
      SUNRPC: Advance the acked position on TLS connections

 .../ABI/testing/procfs-nfsd-reply_cache_stats      |  11 +-
 fs/nfsd/cache.h                                    |  16 +-
 fs/nfsd/netns.h                                    |   2 -
 fs/nfsd/nfscache.c                                 | 238 ++++++++++++---------
 fs/nfsd/nfsd.h                                     |   3 +
 fs/nfsd/nfssvc.c                                   |  11 +-
 fs/nfsd/stats.h                                    |   5 -
 fs/nfsd/trace.h                                    |  47 ++--
 include/linux/sunrpc/svc.h                         |  18 ++
 include/linux/sunrpc/svc_rdma.h                    |   5 +-
 include/linux/sunrpc/svc_xprt.h                    |  26 +++
 include/linux/sunrpc/svcsock.h                     |  16 ++
 include/trace/events/sunrpc.h                      |  62 +++++-
 net/sunrpc/svc.c                                   |   3 +
 net/sunrpc/svc_xprt.c                              |  73 ++++++-
 net/sunrpc/svcauth_unix.c                          |   6 +
 net/sunrpc/svcsock.c                               | 209 +++++++++++++++++-
 net/sunrpc/xprtrdma/svc_rdma_backchannel.c         |   2 +-
 net/sunrpc/xprtrdma/svc_rdma_sendto.c              |  26 ++-
 net/sunrpc/xprtrdma/svc_rdma_transport.c           |  15 ++
 20 files changed, 639 insertions(+), 155 deletions(-)
---
base-commit: 7931bbfa2e56dc27ddd7f1ef46f90fa0194a2868
change-id: 20260325-duplicate-reply-cache-f0fe7c7b740c

Best regards,
--  
Chuck Lever <cel@kernel.org>

[PATCH v8 01/14] SUNRPC: Reject an AUTH_TLS probe inside a TLS session

[Chuck Lever]

svcauth_tls_accept() accepts an AUTH_TLS probe on a connection that
already carries a TLS session. It sets XPT_HANDSHAKE, and
svc_tcp_handshake() then clears XPT_TLS_SESSION and calls
tls_server_hello_x509(). The first handshake_req stays hashed until
the socket is destroyed, so the second request fails with -EBUSY and
svc_tcp_handshake() leaves the connection open. The kTLS ULP is
still installed, but XPT_TLS_SESSION is clear, so every later test
of that flag treats the connection as cleartext.

RFC 9289 Section 4.1 requires the server to reject a probe sent
within an existing TLS session with AUTH_BADCRED. Deny the call
there, before the handshake is started.

Fixes: 74aaf96feaca ("SUNRPC: Teach server to recognize RPC_AUTH_TLS")
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 net/sunrpc/svcauth_unix.c | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/net/sunrpc/svcauth_unix.c b/net/sunrpc/svcauth_unix.c
index ad8a73653e42..fee7e20e2e9b 100644
--- a/net/sunrpc/svcauth_unix.c
+++ b/net/sunrpc/svcauth_unix.c
@@ -1144,6 +1144,12 @@ svcauth_tls_accept(struct svc_rqst *rqstp)
 		return SVC_DENIED;
 	}
 
+	/* RFC 9289 Section 4.1: a probe inside a TLS session is rejected */
+	if (test_bit(XPT_TLS_SESSION, &xprt->xpt_flags)) {
+		rqstp->rq_auth_stat = rpc_autherr_badcred;
+		return SVC_DENIED;
+	}
+
 	/* Signal that mapping to nobody uid/gid is required */
 	cred->cr_uid = INVALID_UID;
 	cred->cr_gid = INVALID_GID;

-- 
2.56.0

[PATCH v8 02/14] NFSD: Make the DRC size limit independent of page size

[Chuck Lever]

nfsd_cache_size_limit() takes the square root of the low-memory page
count and then shifts the result by the page size in KB. The shift
belongs under the root. A host with 64 KB pages has a page count 16
times smaller than a 4 KB host with the same memory, so its root is
4 times smaller and its shift 16 times larger, and the limit comes
out 4 times larger. With 8 GB of low memory the limit is 92672
entries on 4 KB pages and 370688 on 64 KB pages.

Take the root of the byte count instead. The result matches the
table above the function on every page size. Low memory on a 32-bit
host is under 1 GB, so the byte count fits in an unsigned long.

Fixes: 0338dd157282 ("nfsd: dynamically allocate DRC entries")
Suggested-by: NeilBrown <neil@brown.name>
Reviewed-by: NeilBrown <neil@brown.name>
Link: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-1-db5e13fd9944@kernel.org
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 fs/nfsd/nfscache.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/fs/nfsd/nfscache.c b/fs/nfsd/nfscache.c
index 80364b91331a..b43b276a92b3 100644
--- a/fs/nfsd/nfscache.c
+++ b/fs/nfsd/nfscache.c
@@ -73,7 +73,7 @@ nfsd_cache_size_limit(void)
 	unsigned int limit;
 	unsigned long low_pages = totalram_pages() - totalhigh_pages();
 
-	limit = (16 * int_sqrt(low_pages)) << (PAGE_SHIFT-10);
+	limit = int_sqrt(low_pages << PAGE_SHIFT);
 	return min_t(unsigned int, limit, 256*1024);
 }
 

-- 
2.56.0

[PATCH v8 03/14] NFSD: Remove hard cap on duplicate reply cache size

[Chuck Lever]

nfsd_cache_size_limit() scales the DRC entry limit with the square
root of low memory, then clamps the result at 256k entries. The
clamp binds on any host with more than 64 GB of low memory. Once
the cache reaches max_drc_entries, nfsd_prune_bucket_locked()
evicts oldest-first regardless of age, so on such a host the cap
governs retention rather than RC_EXPIRE. A server that completes
more than about 2200 calls per second fills 256k entries inside
the 120 second RC_EXPIRE window, and from then on entries are
evicted before they expire. Those are the entries a retransmit
could still hit.

Commit 0338dd157282 ("nfsd: dynamically allocate DRC entries")
added the clamp in 2013, when the formula's worst case of 1 KB
per entry made 256 MB a reasonable ceiling for the hosts of the
day. The formula was already sized to memory; the clamp froze it
at that generation of hardware.

Remove the cap and let the square-root formula govern sizing. It
scales sub-linearly with memory, and the hash table already uses
kvzalloc, so larger sizes need no physical contiguity. The limit is
unchanged at 64 GB and below. At 1 TB the formula yields 1048576
entries, four times the old cap. Even at 1 KB per entry, the worst
case the old comment assumed, a full cache is 1 GB, 0.1% of that
host's memory. Below the limit, the shrinker frees only entries
older than RC_EXPIRE, so the unexpired entries are not reclaimable,
and each net namespace sizes its own cache.

Reviewed-by: NeilBrown <neil@brown.name>
Link: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-2-db5e13fd9944@kernel.org
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 fs/nfsd/nfscache.c | 24 +++++++++++++-----------
 1 file changed, 13 insertions(+), 11 deletions(-)

diff --git a/fs/nfsd/nfscache.c b/fs/nfsd/nfscache.c
index b43b276a92b3..391cbca74836 100644
--- a/fs/nfsd/nfscache.c
+++ b/fs/nfsd/nfscache.c
@@ -47,8 +47,8 @@ static unsigned long nfsd_reply_cache_scan(struct shrinker *shrink,
 					   struct shrink_control *sc);
 
 /*
- * Put a cap on the size of the DRC based on the amount of available
- * low memory in the machine.
+ * Size the DRC by the amount of low memory in the machine. The
+ * limit scales with the square root of available memory:
  *
  *  64MB:    8192
  * 128MB:   11585
@@ -59,22 +59,24 @@ static unsigned long nfsd_reply_cache_scan(struct shrinker *shrink,
  *   4GB:   65536
  *   8GB:   92681
  *  16GB:  131072
+ *  32GB:  185363
+ *  64GB:  262144
+ * 128GB:  370727
+ * 256GB:  524288
+ * 512GB:  741455
+ *   1TB: 1048576
  *
- * ...with a hard cap of 256k entries. In the worst case, each entry will be
- * ~1k, so the above numbers should give a rough max of the amount of memory
- * used in k.
- *
- * XXX: these limits are per-container, so memory used will increase
- * linearly with number of containers.  Maybe that's OK.
+ * These limits are per-net-namespace, so memory used increases
+ * linearly with the number of namespaces. Below the limit, the
+ * shrinker frees only entries older than RC_EXPIRE, so unexpired
+ * entries are not reclaimable under memory pressure.
  */
 static unsigned int
 nfsd_cache_size_limit(void)
 {
-	unsigned int limit;
 	unsigned long low_pages = totalram_pages() - totalhigh_pages();
 
-	limit = int_sqrt(low_pages << PAGE_SHIFT);
-	return min_t(unsigned int, limit, 256*1024);
+	return int_sqrt(low_pages << PAGE_SHIFT);
 }
 
 /*

-- 
2.56.0

[PATCH v8 04/14] SUNRPC: Assign a unique identifier to each svc_xprt

[Chuck Lever]

A consumer that associates state with a transport without holding a
reference cannot detect ABA collisions: once a transport is freed,
SLUB may hand out a new svc_xprt at the same address.

Assign each transport a 64-bit identifier from a global sequential
counter in svc_xprt_init(). The counter does not wrap, so two
transports never share an identifier and a consumer needs no
release when the transport is freed. The NFSD duplicate reply cache
is the first consumer.

Reviewed-by: NeilBrown <neil@brown.name>
Link: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-3-db5e13fd9944@kernel.org
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 include/linux/sunrpc/svc_xprt.h |  1 +
 include/trace/events/sunrpc.h   |  8 ++++++--
 net/sunrpc/svc_xprt.c           | 17 ++++++++++++++---
 3 files changed, 21 insertions(+), 5 deletions(-)

diff --git a/include/linux/sunrpc/svc_xprt.h b/include/linux/sunrpc/svc_xprt.h
index 2af222f3ea2c..52082d8ea6dd 100644
--- a/include/linux/sunrpc/svc_xprt.h
+++ b/include/linux/sunrpc/svc_xprt.h
@@ -56,6 +56,7 @@ struct svc_xprt {
 	struct svc_xprt_class	*xpt_class;
 	const struct svc_xprt_ops *xpt_ops;
 	struct kref		xpt_ref;
+	u64			xpt_id;
 	ktime_t			xpt_qtime;
 	struct list_head	xpt_list;
 	struct lwq_node		xpt_ready;
diff --git a/include/trace/events/sunrpc.h b/include/trace/events/sunrpc.h
index ff855197880d..77efe63d2c14 100644
--- a/include/trace/events/sunrpc.h
+++ b/include/trace/events/sunrpc.h
@@ -1986,7 +1986,8 @@ TRACE_EVENT(svc_xprt_create_err,
 		__sockaddr(server, (x)->xpt_locallen) \
 		__sockaddr(client, (x)->xpt_remotelen) \
 		__field(unsigned long, flags) \
-		__field(unsigned int, netns_ino)
+		__field(unsigned int, netns_ino) \
+		__field(u64, xpt_id)
 
 #define SVC_XPRT_ENDPOINT_ASSIGNMENTS(x) \
 		do { \
@@ -1996,13 +1997,15 @@ TRACE_EVENT(svc_xprt_create_err,
 					  (x)->xpt_remotelen); \
 			__entry->flags = (x)->xpt_flags; \
 			__entry->netns_ino = (x)->xpt_net->ns.inum; \
+			__entry->xpt_id = (x)->xpt_id; \
 		} while (0)
 
 #define SVC_XPRT_ENDPOINT_FORMAT \
-		"server=%pISpc client=%pISpc flags=%s"
+		"server=%pISpc client=%pISpc xpt_id=%llu flags=%s"
 
 #define SVC_XPRT_ENDPOINT_VARARGS \
 		__get_sockaddr(server), __get_sockaddr(client), \
+		__entry->xpt_id, \
 		show_svc_xprt_flags(__entry->flags)
 
 TRACE_EVENT(svc_xprt_enqueue,
@@ -2024,6 +2027,7 @@ TRACE_EVENT(svc_xprt_enqueue,
 				  xprt->xpt_remotelen);
 		__entry->flags = flags;
 		__entry->netns_ino = xprt->xpt_net->ns.inum;
+		__entry->xpt_id = xprt->xpt_id;
 	),
 
 	TP_printk(SVC_XPRT_ENDPOINT_FORMAT, SVC_XPRT_ENDPOINT_VARARGS)
diff --git a/net/sunrpc/svc_xprt.c b/net/sunrpc/svc_xprt.c
index bcc867592f2d..15c14d75ec12 100644
--- a/net/sunrpc/svc_xprt.c
+++ b/net/sunrpc/svc_xprt.c
@@ -26,6 +26,8 @@
 static unsigned int svc_rpc_per_connection_limit __read_mostly;
 module_param(svc_rpc_per_connection_limit, uint, 0644);
 
+static atomic64_t svc_xprt_id_seq;
+
 
 static struct svc_deferred_req *svc_deferred_dequeue(struct svc_xprt *xprt);
 static int svc_deferred_recv(struct svc_rqst *rqstp);
@@ -191,9 +193,17 @@ void svc_xprt_put(struct svc_xprt *xprt)
 }
 EXPORT_SYMBOL_GPL(svc_xprt_put);
 
-/*
- * Called by transport drivers to initialize the transport independent
- * portion of the transport instance.
+/**
+ * svc_xprt_init - initialize transport-independent portion of a transport
+ * @net: network namespace in which the transport operates
+ * @xcl: transport class providing operations and metadata
+ * @xprt: svc_xprt to initialize
+ * @serv: RPC service that owns this transport
+ *
+ * @xprt->xpt_id is never reused: the 64-bit counter behind it does
+ * not wrap.
+ *
+ * Context: Process context. May sleep.
  */
 void svc_xprt_init(struct net *net, struct svc_xprt_class *xcl,
 		   struct svc_xprt *xprt, struct svc_serv *serv)
@@ -211,6 +221,7 @@ void svc_xprt_init(struct net *net, struct svc_xprt_class *xcl,
 	set_bit(XPT_BUSY, &xprt->xpt_flags);
 	xprt->xpt_net = get_net_track(net, &xprt->ns_tracker, GFP_ATOMIC);
 	strcpy(xprt->xpt_remotebuf, "uninitialized");
+	xprt->xpt_id = atomic64_inc_return(&svc_xprt_id_seq);
 }
 EXPORT_SYMBOL_GPL(svc_xprt_init);
 

-- 
2.56.0

[PATCH v8 05/14] NFSD: Track transport in DRC entries

[Chuck Lever]

A cached reply carries no record of the transport its request arrived
on. Transport-aware eviction needs that association.

Record the xprt id in each entry. struct nfsd_cacherep grows from 128
to 136 bytes. No behavior change.

Reviewed-by: NeilBrown <neil@brown.name>
Link: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-4-db5e13fd9944@kernel.org
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 fs/nfsd/cache.h    | 1 +
 fs/nfsd/nfscache.c | 1 +
 2 files changed, 2 insertions(+)

diff --git a/fs/nfsd/cache.h b/fs/nfsd/cache.h
index 3bc4856e34b8..589ffe4762d5 100644
--- a/fs/nfsd/cache.h
+++ b/fs/nfsd/cache.h
@@ -37,6 +37,7 @@ struct nfsd_cacherep {
 	unsigned char		c_state,	/* unused, inprog, done */
 				c_type,		/* status, buffer */
 				c_secure : 1;	/* req came from port < 1024 */
+	u64			c_xprt;		/* svc_xprt that carried req */
 	unsigned long		c_timestamp;
 	union {
 		struct kvec	u_vec;
diff --git a/fs/nfsd/nfscache.c b/fs/nfsd/nfscache.c
index 391cbca74836..01ad6a9ebe1f 100644
--- a/fs/nfsd/nfscache.c
+++ b/fs/nfsd/nfscache.c
@@ -111,6 +111,7 @@ nfsd_cacherep_alloc(struct svc_rqst *rqstp, __wsum csum,
 		rp->c_key.k_vers = rqstp->rq_vers;
 		rp->c_key.k_len = rqstp->rq_arg.len;
 		rp->c_key.k_csum = csum;
+		rp->c_xprt = rqstp->rq_xprt->xpt_id;
 	}
 	return rp;
 }

-- 
2.56.0

[PATCH v8 06/14] NFSD: Prepare bucket pruning for additional eviction reasons

[Chuck Lever]

nfsd_prune_bucket_locked() combines its two eviction reasons,
memory pressure and RC_EXPIRE, in a single test and stops the walk
at the first entry the test does not evict. A reason that needs
its own handling, a tracepoint per reason or a test that must run
ahead of the stop, has nowhere to go.

Test each reason on its own and jump to a shared eviction label.
The walk still stops at the first entry no reason applies to.

Reviewed-by: NeilBrown <neil@brown.name>
Link: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-5-db5e13fd9944@kernel.org
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 fs/nfsd/nfscache.c | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/fs/nfsd/nfscache.c b/fs/nfsd/nfscache.c
index 01ad6a9ebe1f..988b955590cc 100644
--- a/fs/nfsd/nfscache.c
+++ b/fs/nfsd/nfscache.c
@@ -274,10 +274,13 @@ nfsd_prune_bucket_locked(struct nfsd_net *nn, struct nfsd_drc_bucket *b,
 
 	/* The bucket LRU is ordered oldest-first. */
 	list_for_each_entry_safe(rp, tmp, &b->lru_head, c_lru) {
-		if (atomic_read(&nn->num_drc_entries) <= nn->max_drc_entries &&
-		    time_before(expiry, rp->c_timestamp))
-			break;
+		if (atomic_read(&nn->num_drc_entries) > nn->max_drc_entries)
+			goto evict;
+		if (time_before_eq(rp->c_timestamp, expiry))
+			goto evict;
+		break;
 
+evict:
 		nfsd_cacherep_unlink_locked(nn, b, rp);
 		list_add(&rp->c_lru, dispose);
 

-- 
2.56.0

[PATCH v8 07/14] NFSD: Add tracepoints for DRC entry eviction

[Chuck Lever]

The DRC pruning path evicts entries for memory pressure and expiry,
and neither emits a trace event, so there is no way to see which
reason is retiring entries or how old they are when it happens. A
further eviction reason needs the same visibility.

Add an event class that records the cache population, the XID, the
transport, and the entry's age, and define an event for each
eviction reason.

Reviewed-by: NeilBrown <neil@brown.name>
Link: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-6-db5e13fd9944@kernel.org
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 fs/nfsd/nfscache.c |  8 ++++++--
 fs/nfsd/trace.h    | 37 +++++++++++++++++++++++++++++++++++++
 2 files changed, 43 insertions(+), 2 deletions(-)

diff --git a/fs/nfsd/nfscache.c b/fs/nfsd/nfscache.c
index 988b955590cc..680025334f1f 100644
--- a/fs/nfsd/nfscache.c
+++ b/fs/nfsd/nfscache.c
@@ -274,10 +274,14 @@ nfsd_prune_bucket_locked(struct nfsd_net *nn, struct nfsd_drc_bucket *b,
 
 	/* The bucket LRU is ordered oldest-first. */
 	list_for_each_entry_safe(rp, tmp, &b->lru_head, c_lru) {
-		if (atomic_read(&nn->num_drc_entries) > nn->max_drc_entries)
+		if (atomic_read(&nn->num_drc_entries) > nn->max_drc_entries) {
+			trace_nfsd_drc_evict_pressure(nn, rp);
 			goto evict;
-		if (time_before_eq(rp->c_timestamp, expiry))
+		}
+		if (time_before_eq(rp->c_timestamp, expiry)) {
+			trace_nfsd_drc_evict_expired(nn, rp);
 			goto evict;
+		}
 		break;
 
 evict:
diff --git a/fs/nfsd/trace.h b/fs/nfsd/trace.h
index e58b753c1ebc..4a3623b3899a 100644
--- a/fs/nfsd/trace.h
+++ b/fs/nfsd/trace.h
@@ -1709,6 +1709,43 @@ TRACE_EVENT(nfsd_drc_mismatch,
 		__entry->ingress)
 );
 
+DECLARE_EVENT_CLASS(nfsd_drc_entry_class,
+	TP_PROTO(
+		const struct nfsd_net *nn,
+		const struct nfsd_cacherep *rp
+	),
+	TP_ARGS(nn, rp),
+	TP_STRUCT__entry(
+		__field(unsigned long long, boot_time)
+		__field(unsigned int, num_drc_entries)
+		__field(u32, xid)
+		__field(u64, xprt)
+		__field(unsigned long, age)
+	),
+	TP_fast_assign(
+		__entry->boot_time = nn->boot_time;
+		__entry->num_drc_entries = atomic_read(&nn->num_drc_entries);
+		__entry->xid = be32_to_cpu(rp->c_key.k_xid);
+		__entry->xprt = rp->c_xprt;
+		__entry->age = time_is_after_jiffies(rp->c_timestamp) ?
+				1 : jiffies - rp->c_timestamp;
+	),
+	TP_printk("boot_time=%16llx entries=%u xid=0x%08x xprt=%llu age=%lu",
+		__entry->boot_time, __entry->num_drc_entries,
+		__entry->xid, __entry->xprt, __entry->age)
+);
+
+#define DEFINE_NFSD_DRC_ENTRY_EVENT(name)			\
+DEFINE_EVENT(nfsd_drc_entry_class, nfsd_drc_##name,		\
+	TP_PROTO(						\
+		const struct nfsd_net *nn,			\
+		const struct nfsd_cacherep *rp			\
+	),							\
+	TP_ARGS(nn, rp))
+
+DEFINE_NFSD_DRC_ENTRY_EVENT(evict_pressure);
+DEFINE_NFSD_DRC_ENTRY_EVENT(evict_expired);
+
 TRACE_EVENT(nfsd_cb_args,
 	TP_PROTO(
 		const struct nfs4_client *clp,

-- 
2.56.0

[PATCH v8 08/14] NFSD: Record DRC population in lookup tracepoints

[Chuck Lever]

nfsd_drc_found reports the outcome of a lookup but not how full the
cache was at the time, so a trace cannot show whether retransmits are
being caught while the cache runs near its cap.

Reviewed-by: NeilBrown <neil@brown.name>
Link: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-7-db5e13fd9944@kernel.org
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 fs/nfsd/nfscache.c |  4 +++-
 fs/nfsd/trace.h    | 11 +++++++----
 2 files changed, 10 insertions(+), 5 deletions(-)

diff --git a/fs/nfsd/nfscache.c b/fs/nfsd/nfscache.c
index 680025334f1f..453ff718ff6c 100644
--- a/fs/nfsd/nfscache.c
+++ b/fs/nfsd/nfscache.c
@@ -486,6 +486,7 @@ int nfsd_cache_lookup(struct svc_rqst *rqstp, unsigned int start,
 	struct nfsd_drc_bucket	*b;
 	int type = ntli->ntli_cachetype;
 	LIST_HEAD(dispose);
+	unsigned int entries;
 	int rtn = RC_DOIT;
 
 	if (type == RC_NOCACHE) {
@@ -555,7 +556,8 @@ int nfsd_cache_lookup(struct svc_rqst *rqstp, unsigned int start,
 	}
 
 out_trace:
-	trace_nfsd_drc_found(nn, rqstp, rtn);
+	entries = atomic_read(&nn->num_drc_entries);
+	trace_nfsd_drc_found(nn, entries, rqstp, rtn);
 out_unlock:
 	spin_unlock(&b->cache_lock);
 out:
diff --git a/fs/nfsd/trace.h b/fs/nfsd/trace.h
index 4a3623b3899a..41a9407e3efa 100644
--- a/fs/nfsd/trace.h
+++ b/fs/nfsd/trace.h
@@ -1665,23 +1665,26 @@ TRACE_DEFINE_ENUM(RC_DOIT);
 TRACE_EVENT(nfsd_drc_found,
 	TP_PROTO(
 		const struct nfsd_net *nn,
+		unsigned int num_drc_entries,
 		const struct svc_rqst *rqstp,
 		int result
 	),
-	TP_ARGS(nn, rqstp, result),
+	TP_ARGS(nn, num_drc_entries, rqstp, result),
 	TP_STRUCT__entry(
 		__field(unsigned long long, boot_time)
+		__field(unsigned int, num_drc_entries)
 		__field(unsigned long, result)
 		__field(u32, xid)
 	),
 	TP_fast_assign(
 		__entry->boot_time = nn->boot_time;
+		__entry->num_drc_entries = num_drc_entries;
 		__entry->result = result;
 		__entry->xid = be32_to_cpu(rqstp->rq_xid);
 	),
-	TP_printk("boot_time=%16llx xid=0x%08x result=%s",
-		__entry->boot_time, __entry->xid,
-		show_drc_retval(__entry->result))
+	TP_printk("boot_time=%16llx entries=%u xid=0x%08x result=%s",
+		__entry->boot_time, __entry->num_drc_entries,
+		__entry->xid, show_drc_retval(__entry->result))
 
 );
 

-- 
2.56.0

[PATCH v8 09/14] SUNRPC: Publish reply positions for upper-layer consumers

[Chuck Lever]

An RPC server's duplicate reply cache has to hold a cached reply
until the client has received it. SUNRPC gives an upper layer no way
to learn that, so a cache entry lives for the full retention
interval even when the reply was delivered at once.

Add two transport-neutral values that an upper layer can compare:
the position of each reply on its transport, recorded in the svc_rqst
by xpo_sendto(), and the position up to which the peer has
acknowledged, recorded on the svc_xprt. The units are private to the
transport (bytes for a stream, Send sequence for RDMA). A consumer
learns only whether one is less than or equal to the other. A reply
that xpo_sendto() does not hand to the wire keeps position zero and
is never reported as acknowledged.

The acknowledged position is an atomic64_t. A transport publishes it
outside any lock the consumer holds, and a torn 64-bit read on a
32-bit host could run ahead of the peer. A transport raises the
position through svc_xprt_advance_acked(), a plain read and store
that computes a maximum only while one context per transport
writes. A cmpxchg loop would make the helper safe from any context
at the cost of a locked instruction on every acknowledgment.

An acknowledgment proves the peer holds the reply, so no retransmit
of that call follows. A retransmit the peer sent before the reply
reached it precedes the acknowledgment in the byte stream and can
still be ahead of its lookup when the transport refreshes the
position: unread in the socket, or read and parked in
svc_authenticate(), which can sleep for seconds in the ip_map upcall
or defer the request. A consumer that evicts on the refreshed
position sends the retransmit through the procedure again.

Count a transport's requests from first read until the consumer
reports the lookup with svc_xprt_recv_done(), and carry the count
across a deferral. A revisited request repeats its lookup, so a
procedure that defers after the first lookup was reported puts the
request back under the count. Publish the position through
svc_xprt_acked_pos(), so the transport can withhold the position
while any request but the caller's is counted. A request that never
reaches a lookup is uncounted when it is released.

Add a per-service hook that runs once per request when its reply
phase ends, whether svc_send() sent the reply or svc_process()
dropped it, so an upper layer can record the reply's position
against its own state either way. On the drop path the position is
zero.

Link: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-8-db5e13fd9944@kernel.org
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 include/linux/sunrpc/svc.h      | 18 +++++++++++++
 include/linux/sunrpc/svc_xprt.h | 25 ++++++++++++++++++
 net/sunrpc/svc.c                |  3 +++
 net/sunrpc/svc_xprt.c           | 56 +++++++++++++++++++++++++++++++++++++++++
 4 files changed, 102 insertions(+)

diff --git a/include/linux/sunrpc/svc.h b/include/linux/sunrpc/svc.h
index b1dff5c8aa89..fc83b24934b0 100644
--- a/include/linux/sunrpc/svc.h
+++ b/include/linux/sunrpc/svc.h
@@ -59,6 +59,8 @@ enum {
 };
 
 
+struct svc_rqst;
+
 /*
  * RPC service.
  *
@@ -101,6 +103,13 @@ struct svc_serv {
 						 * connection */
 	bool			sv_bc_enabled;	/* service uses backchannel */
 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
+
+	/*
+	 * Called once per request after its reply phase, whether
+	 * xpo_sendto() succeeded, failed, or was never reached because
+	 * svc_process() dropped the reply. See rq_reply_pos.
+	 */
+	void			(*sv_reply_sent)(struct svc_rqst *rqstp);
 };
 
 /* This is used by pool_stats to find and lock an svc */
@@ -272,6 +281,13 @@ struct svc_rqst {
 	unsigned int		bc_to_retries;
 	unsigned int		rq_status_counter; /* RPC processing counter */
 	void			*rq_private;	/* For use by the service thread */
+
+	/*
+	 * Transport position of this request's reply, set by
+	 * xpo_sendto(); zero when the reply did not reach the wire.
+	 * Comparable only with the xpt_acked_pos of the same svc_xprt.
+	 */
+	u64			rq_reply_pos;
 };
 
 /* bits for rq_flags */
@@ -282,6 +298,7 @@ enum {
 	RQ_DROPME,		/* drop current reply */
 	RQ_VICTIM,		/* Have agreed to shut down */
 	RQ_DATA,		/* request has data */
+	RQ_RECV_PENDING,	/* counted in xpt_recv_pending */
 };
 
 #define SVC_NET(rqst) (rqst->rq_xprt ? rqst->rq_xprt->xpt_net : rqst->rq_bc_net)
@@ -364,6 +381,7 @@ static inline void svc_thread_init_status(struct svc_rqst *rqstp, int err)
 struct svc_deferred_req {
 	u32			prot;	/* protocol (UDP or TCP) */
 	bool			secure;	/* RQ_SECURE of the original request */
+	bool			recv_pending; /* RQ_RECV_PENDING, carried over */
 	struct svc_xprt		*xprt;
 	struct sockaddr_storage	addr;	/* where reply must go */
 	size_t			addrlen;
diff --git a/include/linux/sunrpc/svc_xprt.h b/include/linux/sunrpc/svc_xprt.h
index 52082d8ea6dd..aea98ddbfe52 100644
--- a/include/linux/sunrpc/svc_xprt.h
+++ b/include/linux/sunrpc/svc_xprt.h
@@ -28,6 +28,7 @@ struct svc_xprt_ops {
 	void		(*xpo_free)(struct svc_xprt *);
 	void		(*xpo_kill_temp_xprt)(struct svc_xprt *);
 	void		(*xpo_handshake)(struct svc_xprt *xprt);
+	u64		(*xpo_acked_pos)(struct svc_rqst *rqstp);
 };
 
 struct svc_xprt_class {
@@ -66,6 +67,10 @@ struct svc_xprt {
 	atomic_t		xpt_reserved;	/* outq space rsvd, UDP only */
 	atomic_t		xpt_nr_rqsts;	/* Number of requests */
 	struct mutex		xpt_mutex;	/* to serialize sending data */
+	atomic64_t		xpt_acked_pos;	/* replies up to this position
+						 * are acknowledged; 0 = none */
+	/* requests read but not yet past svc_xprt_recv_done() */
+	atomic_t		xpt_recv_pending;
 	spinlock_t		xpt_lock;	/* protects sk_deferred
 						 * and xpt_auth_cache */
 	void			*xpt_auth_cache;/* auth cache */
@@ -161,6 +166,24 @@ static inline bool svc_xprt_is_dead(const struct svc_xprt *xprt)
 		(test_bit(XPT_CLOSE, &xprt->xpt_flags) != 0);
 }
 
+/**
+ * svc_xprt_advance_acked - Raise a transport's acknowledged position
+ * @xprt: transport that sent the replies
+ * @pos: position the peer has acknowledged, in @xprt's own units
+ *
+ * Context: Any context. Callers must serialize against other writers
+ *	of @xprt's xpt_acked_pos.
+ * Return: %true if the position was raised, %false if @pos does not
+ *	exceed the recorded position.
+ */
+static inline bool svc_xprt_advance_acked(struct svc_xprt *xprt, u64 pos)
+{
+	if (pos <= atomic64_read(&xprt->xpt_acked_pos))
+		return false;
+	atomic64_set(&xprt->xpt_acked_pos, pos);
+	return true;
+}
+
 int	svc_reg_xprt_class(struct svc_xprt_class *);
 void	svc_unreg_xprt_class(struct svc_xprt_class *);
 void	svc_xprt_init(struct net *, struct svc_xprt_class *, struct svc_xprt *,
@@ -175,6 +198,8 @@ int	svc_xprt_create(struct svc_serv *serv, const char *xprt_name,
 void	svc_xprt_destroy_all(struct svc_serv *serv, struct net *net,
 			     bool unregister);
 void	svc_xprt_received(struct svc_xprt *xprt);
+void	svc_xprt_recv_done(struct svc_rqst *rqstp);
+u64	svc_xprt_acked_pos(struct svc_rqst *rqstp);
 void	svc_xprt_enqueue(struct svc_xprt *xprt);
 void	svc_xprt_put(struct svc_xprt *xprt);
 void	svc_xprt_copy_addrs(struct svc_rqst *rqstp, struct svc_xprt *xprt);
diff --git a/net/sunrpc/svc.c b/net/sunrpc/svc.c
index f261a4f47276..0dcb1085ecb1 100644
--- a/net/sunrpc/svc.c
+++ b/net/sunrpc/svc.c
@@ -1503,6 +1503,7 @@ svc_process_common(struct svc_rqst *rqstp)
 
 	/* Reset the accept_stat for the RPC */
 	rqstp->rq_accept_statp = NULL;
+	rqstp->rq_reply_pos = 0;
 
 	/* Will be turned off only when NFSv4 Sessions are used */
 	set_bit(RQ_USEDEFERRAL, &rqstp->rq_flags);
@@ -1748,6 +1749,8 @@ void svc_process(struct svc_rqst *rqstp)
 		goto out_baddir;
 
 	if (!svc_process_common(rqstp)) {
+		if (rqstp->rq_server->sv_reply_sent)
+			rqstp->rq_server->sv_reply_sent(rqstp);
 		svc_release_rqst(rqstp);
 		goto out_drop;
 	}
diff --git a/net/sunrpc/svc_xprt.c b/net/sunrpc/svc_xprt.c
index 15c14d75ec12..4baeb82d02c0 100644
--- a/net/sunrpc/svc_xprt.c
+++ b/net/sunrpc/svc_xprt.c
@@ -593,14 +593,57 @@ static void free_deferred(struct svc_xprt *xprt, struct svc_deferred_req *dr)
 	if (!dr)
 		return;
 
+	if (dr->recv_pending)
+		atomic_dec(&xprt->xpt_recv_pending);
 	xprt->xpt_ops->xpo_release_ctxt(xprt, dr->xprt_ctxt);
 	kfree(dr);
 }
 
+/**
+ * svc_xprt_recv_done - a request has been looked up in the reply cache
+ * @rqstp: request that has been looked up
+ *
+ * A transport counts each request from its read until this call, and
+ * svc_xprt_acked_pos() withholds the acknowledged position while the
+ * count is non-zero: a retransmit the peer sent before the
+ * acknowledgment may be among the requests counted. Call this once
+ * the request has been looked up, before the reply cache acts on the
+ * position. A request that never reaches a lookup is uncounted when
+ * it is released.
+ *
+ * Context: any context.
+ */
+void svc_xprt_recv_done(struct svc_rqst *rqstp)
+{
+	if (test_and_clear_bit(RQ_RECV_PENDING, &rqstp->rq_flags))
+		atomic_dec(&rqstp->rq_xprt->xpt_recv_pending);
+}
+EXPORT_SYMBOL_GPL(svc_xprt_recv_done);
+
+/**
+ * svc_xprt_acked_pos - acknowledged position the server may act on
+ * @rqstp: the request on whose behalf the position is read
+ *
+ * Context: process context. May take and release the socket lock.
+ * Return: the position up to which the peer has acknowledged replies
+ * and no earlier request can still be ahead of its lookup, or 0 when
+ * the transport publishes no position or cannot yet vouch for it.
+ */
+u64 svc_xprt_acked_pos(struct svc_rqst *rqstp)
+{
+	struct svc_xprt *xprt = rqstp->rq_xprt;
+
+	if (!xprt->xpt_ops->xpo_acked_pos)
+		return 0;
+	return xprt->xpt_ops->xpo_acked_pos(rqstp);
+}
+EXPORT_SYMBOL_GPL(svc_xprt_acked_pos);
+
 static void svc_xprt_release(struct svc_rqst *rqstp)
 {
 	struct svc_xprt	*xprt = rqstp->rq_xprt;
 
+	svc_xprt_recv_done(rqstp);
 	xprt->xpt_ops->xpo_release_ctxt(xprt, rqstp->rq_xprt_ctxt);
 	rqstp->rq_xprt_ctxt = NULL;
 
@@ -1037,6 +1080,8 @@ void svc_send(struct svc_rqst *rqstp)
 	trace_svc_stats_latency(rqstp);
 
 	status = xprt->xpt_ops->xpo_sendto(rqstp);
+	if (xprt->xpt_server->sv_reply_sent)
+		xprt->xpt_server->sv_reply_sent(rqstp);
 
 	trace_svc_send(rqstp, status);
 }
@@ -1352,6 +1397,13 @@ static struct cache_deferred_req *svc_defer(struct cache_req *req)
 	}
 	dr->xprt_ctxt = rqstp->rq_xprt_ctxt;
 	rqstp->rq_xprt_ctxt = NULL;
+	/*
+	 * The revisit repeats the lookup, so a request a procedure
+	 * defers after its lookup was reported is counted again.
+	 */
+	if (!test_and_clear_bit(RQ_RECV_PENDING, &rqstp->rq_flags))
+		atomic_inc(&rqstp->rq_xprt->xpt_recv_pending);
+	dr->recv_pending = true;
 	trace_svc_defer(rqstp);
 	svc_xprt_get(rqstp->rq_xprt);
 	dr->xprt = rqstp->rq_xprt;
@@ -1386,6 +1438,10 @@ static noinline int svc_deferred_recv(struct svc_rqst *rqstp)
 	 * it handled last. Restore the value computed at receive time.
 	 */
 	assign_bit(RQ_SECURE, &rqstp->rq_flags, dr->secure);
+	if (dr->recv_pending) {
+		set_bit(RQ_RECV_PENDING, &rqstp->rq_flags);
+		dr->recv_pending = false;
+	}
 
 	dr->xprt_ctxt = NULL;
 	svc_xprt_received(rqstp->rq_xprt);

-- 
2.56.0

[PATCH v8 10/14] SUNRPC: Publish TCP reply positions

[Chuck Lever]

Add the svcsock side of reply-position reporting. A reply's position
is the running count of bytes handed to the socket, kept per socket
under xpt_mutex. A TCP sequence number cannot serve as the position:
the 32-bit space wraps in a few hundred milliseconds at 40 GbE, and
a duplicate reply cache entry lives for two minutes.

The acknowledged position is derived from the socket's unacknowledged
byte count, write_seq minus snd_una, after each successful send.
Refreshing the position only in the send path means a consumer sees
the acknowledgment state as of the previous reply on that connection,
stale by one round trip but never ahead of the peer. A failed or
refused send leaves rq_reply_pos at zero, so the reply is never
reported as acknowledged.

A call the peer sent before it acknowledged a reply precedes the
acknowledgment in the stream, so the socket has received that call by
the time the acknowledgment raises the position. Record rcv_nxt as a
mark whenever the position rises, and withhold the position from
svc_tcp_acked_pos() until copied_seq has reached the mark and no
request but the caller's is counted in xpt_recv_pending. A record in
progress is counted from its first octet, so a partially read
retransmit withholds the position too. The socket lock keeps snd_una,
rcv_nxt, and the out-of-order queue from moving between the reads
and orders them against the reader. A zero return is always safe, so
the count is tested before the lock as well: a cacheable call takes
the lock only after the first position is published and only while
no other request is counted. On a busy connection the position lags
a refresh by the time the requests ahead of the mark take to reach
their lookups. A service that installs no sv_reply_sent hook never
reads the position, so svc_tcp_sendto() skips the refresh for it.

When a segment of the retransmit is lost on the wire, the
acknowledgment the peer sent after that segment arrives first. TCP
processes the acknowledgment, so snd_una covers the reply while
rcv_nxt stops short of the retransmit. The segments behind the hole
sit on the out-of-order queue. When every segment of the retransmit
is lost, the acknowledgment arrives alone: tcp_data_queue() frees a
segment that carries no data without queuing it, and only snd_wl1,
which tcp_ack_update_window() sets to the sequence number of each
segment that advances snd_una, records how far the peer has sent.
The mark is therefore the furthest of rcv_nxt, the end of the
out-of-order queue, and snd_wl1. copied_seq cannot reach that mark
until TCP has delivered the lost segments and the record behind them
has been read.

Under kTLS the socket counts ciphertext while svcsock counts
plaintext, and tls_sw_sendmsg() can return the full plaintext count
with part of a record still waiting for socket write space. The
unacknowledged count then understates what the peer has yet to
receive. Whether a record is pending is state private to net/tls, so
a TLS session publishes no acknowledged position and its replies
are retained for the full interval.

The svcsock_tcp_send tracepoint fires before the position is
updated and logs only the byte count, so a trace cannot show whether
xpt_acked_pos ever claimed more than the client had acknowledged.
Add svcsock_tcp_acked, recording write_seq, snd_una, and sk_send_pos,
the candidate position they yield, and the position published after
the monotonic clamp. The clamp can hold the published position above
the candidate, so the record carries both. xpt_id separates samples
from successive connections that share a remote address.

Link: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-9-db5e13fd9944@kernel.org
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 include/linux/sunrpc/svcsock.h |   7 +++
 include/trace/events/sunrpc.h  |  39 +++++++++++++
 net/sunrpc/svcsock.c           | 126 ++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 171 insertions(+), 1 deletion(-)

diff --git a/include/linux/sunrpc/svcsock.h b/include/linux/sunrpc/svcsock.h
index 372a00882ca6..0fa62f9c6465 100644
--- a/include/linux/sunrpc/svcsock.h
+++ b/include/linux/sunrpc/svcsock.h
@@ -38,9 +38,16 @@ struct svc_sock {
 	/* Total length of the data (not including fragment headers)
 	 * received so far in the fragments making up this rpc: */
 	u32			sk_datalen;
+	/* The record in progress is counted in xpt_recv_pending */
+	bool			sk_recv_counted;
 
 	struct page_frag_cache  sk_frag_cache;
 
+	/* reply bytes handed to the socket; protected by xpt_mutex */
+	u64			sk_send_pos;
+	/* rcv_nxt when xpt_acked_pos was last raised; socket lock */
+	u32			sk_acked_rcv;
+
 	struct completion	sk_handshake_done;
 
 	/* received data */
diff --git a/include/trace/events/sunrpc.h b/include/trace/events/sunrpc.h
index 77efe63d2c14..a5d914376627 100644
--- a/include/trace/events/sunrpc.h
+++ b/include/trace/events/sunrpc.h
@@ -2349,6 +2349,45 @@ TRACE_EVENT(svcsock_tcp_recv_short,
 	)
 );
 
+TRACE_EVENT(svcsock_tcp_acked,
+	TP_PROTO(
+		const struct svc_xprt *xprt,
+		u64 send_pos,
+		u32 write_seq,
+		u32 snd_una,
+		u64 candidate,
+		u64 acked_pos
+	),
+
+	TP_ARGS(xprt, send_pos, write_seq, snd_una, candidate, acked_pos),
+
+	TP_STRUCT__entry(
+		__field(u64, xpt_id)
+		__field(u64, send_pos)
+		__field(u64, candidate)
+		__field(u64, acked_pos)
+		__field(u32, write_seq)
+		__field(u32, snd_una)
+		__string(addr, xprt->xpt_remotebuf)
+	),
+
+	TP_fast_assign(
+		__entry->xpt_id = xprt->xpt_id;
+		__entry->send_pos = send_pos;
+		__entry->candidate = candidate;
+		__entry->acked_pos = acked_pos;
+		__entry->write_seq = write_seq;
+		__entry->snd_una = snd_una;
+		__assign_str(addr);
+	),
+
+	TP_printk("addr=%s xpt_id=%llu send_pos=%llu write_seq=%u snd_una=%u candidate=%llu acked_pos=%llu",
+		__get_str(addr), __entry->xpt_id, __entry->send_pos,
+		__entry->write_seq, __entry->snd_una, __entry->candidate,
+		__entry->acked_pos
+	)
+);
+
 TRACE_EVENT(svcsock_tcp_state,
 	TP_PROTO(
 		const struct svc_xprt *xprt,
diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index fd2f62652b93..4eaa8eb4d537 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -1120,6 +1120,15 @@ static int svc_tcp_recv_actor(read_descriptor_t *desc, struct sk_buff *skb,
 
 	len = min(len, desc->count);
 
+	/*
+	 * sk_tcplen cannot gate the count: an empty non-final fragment
+	 * returns sk_tcplen and sk_datalen to zero mid-record.
+	 */
+	if (!svsk->sk_recv_counted) {
+		atomic_inc(&svsk->sk_xprt.xpt_recv_pending);
+		svsk->sk_recv_counted = true;
+	}
+
 	if (svsk->sk_tcplen < sizeof(rpc_fraghdr)) {
 		want = sizeof(rpc_fraghdr) - svsk->sk_tcplen;
 		n = min(want, len);
@@ -1296,11 +1305,16 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 
 	p = (__be32 *)rqstp->rq_arg.head[0].iov_base;
 	calldir = p[1];
-	if (calldir)
+	if (calldir) {
+		atomic_dec(&svsk->sk_xprt.xpt_recv_pending);
 		len = receive_cb_reply(svsk, rqstp);
+	} else {
+		set_bit(RQ_RECV_PENDING, &rqstp->rq_flags);
+	}
 
 	/* Reset TCP read info */
 	svsk->sk_datalen = 0;
+	svsk->sk_recv_counted = false;
 	svc_tcp_fragment_received(svsk);
 
 	if (len < 0)
@@ -1335,6 +1349,10 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 	 */
 	svsk->sk_datalen = 0;
 err_delete:
+	if (svsk->sk_recv_counted) {
+		atomic_dec(&svsk->sk_xprt.xpt_recv_pending);
+		svsk->sk_recv_counted = false;
+	}
 	trace_svcsock_tcp_recv_err(&svsk->sk_xprt, len);
 	svc_xprt_deferred_close(&svsk->sk_xprt);
 err_noclose:
@@ -1382,6 +1400,107 @@ static int svc_tcp_sendmsg(struct svc_sock *svsk, struct svc_rqst *rqstp,
 	return ret;
 }
 
+/*
+ * Bytes the socket has not seen acknowledged all belong to the most
+ * recent replies, so sk_send_pos less that count is the acknowledged
+ * position.
+ *
+ * A call the peer sent before it acknowledged a reply precedes the
+ * acknowledgment in the stream. When a segment of the call is lost on
+ * the wire, rcv_nxt stops short of the call and the segments behind
+ * the hole sit on the out-of-order queue. When every segment of the
+ * call is lost, the acknowledgment arrives alone and TCP keeps its
+ * sequence number in snd_wl1. The furthest of rcv_nxt, the end of the
+ * out-of-order queue, and snd_wl1 covers the call. Record that mark
+ * with the position; svc_tcp_acked_pos() releases the position once
+ * everything up to the mark has been read. The socket lock keeps
+ * write_seq, snd_una, rcv_nxt, snd_wl1, and the out-of-order queue
+ * from moving between the reads and orders them against
+ * svc_tcp_acked_pos().
+ *
+ * Under kTLS, tls_sw_sendmsg() can return the full plaintext count
+ * with part of a record still waiting for socket write space, leaving
+ * write_seq short of the reply. Whether a record is pending is private
+ * to net/tls, so a TLS session publishes no acknowledged position.
+ * tlshd installs the ULP before XPT_TLS_SESSION reports the session,
+ * so neither does a reply sent while XPT_HANDSHAKE is set. Test
+ * XPT_HANDSHAKE first: it is cleared after XPT_TLS_SESSION is set.
+ */
+static void svc_tcp_update_acked(struct svc_sock *svsk)
+{
+	struct sock *sk = svsk->sk_sk;
+	struct tcp_sock *tp = tcp_sk(sk);
+	u64 unacked, acked = 0, acked_pos;
+	u32 write_seq, snd_una, mark;
+
+	if (test_bit(XPT_HANDSHAKE, &svsk->sk_xprt.xpt_flags) ||
+	    test_bit(XPT_TLS_SESSION, &svsk->sk_xprt.xpt_flags))
+		return;
+	lock_sock(sk);
+	write_seq = READ_ONCE(tp->write_seq);
+	snd_una = READ_ONCE(tp->snd_una);
+	unacked = write_seq - snd_una;
+	if (unacked < svsk->sk_send_pos)
+		acked = svsk->sk_send_pos - unacked;
+	if (svc_xprt_advance_acked(&svsk->sk_xprt, acked)) {
+		if (RB_EMPTY_ROOT(&tp->out_of_order_queue))
+			mark = READ_ONCE(tp->rcv_nxt);
+		else
+			mark = TCP_SKB_CB(tp->ooo_last_skb)->end_seq;
+		if (after(tp->snd_wl1, mark))
+			mark = tp->snd_wl1;
+		svsk->sk_acked_rcv = mark;
+	}
+	acked_pos = atomic64_read(&svsk->sk_xprt.xpt_acked_pos);
+	release_sock(sk);
+	trace_svcsock_tcp_acked(&svsk->sk_xprt, svsk->sk_send_pos, write_seq,
+				snd_una, acked, acked_pos);
+}
+
+static int svc_tcp_recv_pending(const struct svc_rqst *rqstp)
+{
+	return atomic_read(&rqstp->rq_xprt->xpt_recv_pending) -
+	       !!test_bit(RQ_RECV_PENDING, &rqstp->rq_flags);
+}
+
+/**
+ * svc_tcp_acked_pos - acknowledged position the reply cache may act on
+ * @rqstp: the request on whose behalf the position is read
+ *
+ * A retransmit the peer sent before the acknowledgment can still be
+ * on its way through svcsock: unread in the socket, or read and not
+ * yet looked up. Withhold the position until the socket has been read
+ * up to the mark taken with it and no request but @rqstp is counted
+ * between its read and its lookup. A record in progress is counted
+ * from its first octet, so a partially read retransmit withholds the
+ * position too.
+ *
+ * Context: process context. May take and release the socket lock.
+ * Return: xpt_acked_pos, or 0 while it cannot be acted on.
+ */
+static u64 svc_tcp_acked_pos(struct svc_rqst *rqstp)
+{
+	struct svc_xprt *xprt = rqstp->rq_xprt;
+	struct svc_sock *svsk = container_of(xprt, struct svc_sock, sk_xprt);
+	struct sock *sk = svsk->sk_sk;
+	u64 pos;
+
+	/* The position only rises, so a zero read is final. */
+	if (!atomic64_read(&xprt->xpt_acked_pos))
+		return 0;
+	/* A zero return is always safe, so test the count unlocked first. */
+	if (svc_tcp_recv_pending(rqstp))
+		return 0;
+
+	lock_sock(sk);
+	pos = atomic64_read(&xprt->xpt_acked_pos);
+	if (svc_tcp_recv_pending(rqstp) ||
+	    before(READ_ONCE(tcp_sk(sk)->copied_seq), svsk->sk_acked_rcv))
+		pos = 0;
+	release_sock(sk);
+	return pos;
+}
+
 /**
  * svc_tcp_sendto - Send out a reply on a TCP socket
  * @rqstp: completed svc_rqst
@@ -1410,6 +1529,10 @@ static int svc_tcp_sendto(struct svc_rqst *rqstp)
 	trace_svcsock_tcp_send(xprt, sent);
 	if (sent < 0 || sent != (xdr->len + sizeof(marker)))
 		goto out_close;
+	svsk->sk_send_pos += sent;
+	rqstp->rq_reply_pos = svsk->sk_send_pos;
+	if (rqstp->rq_server->sv_reply_sent)
+		svc_tcp_update_acked(svsk);
 	mutex_unlock(&xprt->xpt_mutex);
 	return sent;
 
@@ -1446,6 +1569,7 @@ static const struct svc_xprt_ops svc_tcp_ops = {
 	.xpo_accept = svc_tcp_accept,
 	.xpo_kill_temp_xprt = svc_tcp_kill_temp_xprt,
 	.xpo_handshake = svc_tcp_handshake,
+	.xpo_acked_pos = svc_tcp_acked_pos,
 };
 
 static struct svc_xprt_class svc_tcp_class = {

-- 
2.56.0

[PATCH v8 11/14] svcrdma: Publish RDMA reply positions

[Chuck Lever]

Add the svcrdma side of reply-position reporting. A reply's position
is the sequence number of its Send on the transport, and the Send
completion for that reply publishes the same number as the
transport's acknowledged position. A successful Send completion means
the peer's HCA has acknowledged the message, so a reply whose Send
has completed is in the client's receive buffer.

Completions on one queue pair arrive in posting order. Several nfsd
threads post on one queue pair with no serialization above the
provider, so the order in which Sends are queued is decided inside
the provider. Assign the sequence number and post the Send under one
lock so positions match queuing order. ib_post_send() does not
sleep; the Send Queue wait stays outside the lock.

Only a posted RPC Reply reports a position. The RDMA_ERROR message
and backchannel calls post with no position pointer, and a Send that
fails to post leaves rq_reply_pos at zero.

svcrdma publishes the position as the Send completion raised it,
with neither the request count nor the receive mark that svcsock
keeps. xprtrdma does not retransmit on a live connection:
xprt_rdma_timer() disconnects on an RPC timeout, and the retransmit
arrives on a new transport, whose lookups do not evict the old
transport's entries. RFC 8166 does not forbid a retransmit on the
same connection. A client that sends one needs the mark as well as
the count, and svcrdma has no rcv_nxt to record. Send and Receive
completions arrive on separate queues with no common lock, so the
mark needs an ordering between the two completion handlers that
svcrdma does not provide. The gate waits for a client that
retransmits on a live connection.

Link: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-10-db5e13fd9944@kernel.org
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 include/linux/sunrpc/svc_rdma.h            |  5 ++++-
 net/sunrpc/xprtrdma/svc_rdma_backchannel.c |  2 +-
 net/sunrpc/xprtrdma/svc_rdma_sendto.c      | 26 +++++++++++++++++++++++---
 net/sunrpc/xprtrdma/svc_rdma_transport.c   | 15 +++++++++++++++
 4 files changed, 43 insertions(+), 5 deletions(-)

diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 60966f92e8c8..71bd5bcc5ce2 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -96,6 +96,8 @@ struct svcxprt_rdma {
 
 	spinlock_t	     sc_send_lock;
 	struct llist_head    sc_send_ctxts;
+	spinlock_t	     sc_post_lock;	/* orders Send posting */
+	u64		     sc_post_seq;	/* Sends posted so far */
 	spinlock_t	     sc_rw_ctxt_lock;
 	struct llist_head    sc_rw_ctxts;
 
@@ -244,6 +246,7 @@ struct svc_rdma_send_ctxt {
 	struct ib_send_wr	*sc_wr_chain;
 	int			sc_sqecount;
 	struct ib_cqe		sc_cqe;
+	u64			sc_pos;
 	struct xdr_buf		sc_hdrbuf;
 	struct xdr_stream	sc_stream;
 
@@ -309,7 +312,7 @@ extern struct svc_rdma_send_ctxt *
 extern void svc_rdma_send_ctxt_put(struct svcxprt_rdma *rdma,
 				   struct svc_rdma_send_ctxt *ctxt);
 extern int svc_rdma_post_send(struct svcxprt_rdma *rdma,
-			      struct svc_rdma_send_ctxt *ctxt);
+			      struct svc_rdma_send_ctxt *ctxt, u64 *pos);
 extern int svc_rdma_map_reply_msg(struct svcxprt_rdma *rdma,
 				  struct svc_rdma_send_ctxt *sctxt,
 				  const struct svc_rdma_pcl *write_pcl,
diff --git a/net/sunrpc/xprtrdma/svc_rdma_backchannel.c b/net/sunrpc/xprtrdma/svc_rdma_backchannel.c
index e0768d1ee556..20b7dfc20d2b 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_backchannel.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_backchannel.c
@@ -94,7 +94,7 @@ static int svc_rdma_bc_sendto(struct svcxprt_rdma *rdma,
 	sctxt->sc_page_count = 1;
 
 	sctxt->sc_send_wr.opcode = IB_WR_SEND;
-	return svc_rdma_post_send(rdma, sctxt);
+	return svc_rdma_post_send(rdma, sctxt, NULL);
 }
 
 /* Server-side transport endpoint wants a whole page for its send
diff --git a/net/sunrpc/xprtrdma/svc_rdma_sendto.c b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
index b9eee2150ea1..07df06480294 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_sendto.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
@@ -222,6 +222,7 @@ struct svc_rdma_send_ctxt *svc_rdma_send_ctxt_get(struct svcxprt_rdma *rdma)
 	ctxt->sc_page_count = 0;
 	ctxt->sc_wr_chain = &ctxt->sc_send_wr;
 	ctxt->sc_sqecount = 1;
+	ctxt->sc_pos = 0;
 
 	return ctxt;
 
@@ -491,6 +492,7 @@ static void svc_rdma_wc_send(struct ib_cq *cq, struct ib_wc *wc)
 		goto flushed;
 
 	trace_svcrdma_wc_send(&ctxt->sc_cid);
+	svc_xprt_advance_acked(&rdma->sc_xprt, ctxt->sc_pos);
 	svc_rdma_send_ctxt_put(rdma, ctxt);
 	return;
 
@@ -507,24 +509,30 @@ static void svc_rdma_wc_send(struct ib_cq *cq, struct ib_wc *wc)
  * svc_rdma_post_send - Post a WR chain to the Send Queue
  * @rdma: transport context
  * @ctxt: WR chain to post
+ * @pos: OUT: position of this Send on @rdma, or NULL
  *
  * Copy fields in @ctxt to stack variables in order to guarantee
  * that these values remain available after the ib_post_send() call.
  * In some error flow cases, svc_rdma_wc_send() releases @ctxt.
  *
+ * @pos is written only when the chain was posted. Positions on one
+ * transport increase in posting order, and the Send completion of
+ * @ctxt publishes its position in xpt_acked_pos.
+ *
  * Return values:
  *   %0: The transport owns @ctxt; a Send completion or the
  *       transport destructor releases it
  *   %-ENOTCONN: The connection was lost
  */
 int svc_rdma_post_send(struct svcxprt_rdma *rdma,
-		       struct svc_rdma_send_ctxt *ctxt)
+		       struct svc_rdma_send_ctxt *ctxt, u64 *pos)
 {
 	struct ib_send_wr *first_wr = ctxt->sc_wr_chain;
 	struct ib_send_wr *send_wr = &ctxt->sc_send_wr;
 	const struct ib_send_wr *bad_wr = first_wr;
 	struct rpc_rdma_cid cid = ctxt->sc_cid;
 	int ret, sqecount = ctxt->sc_sqecount;
+	u64 seq;
 
 	might_sleep();
 
@@ -539,7 +547,17 @@ int svc_rdma_post_send(struct svcxprt_rdma *rdma,
 		return ret;
 
 	trace_svcrdma_post_send(ctxt);
+
+	/*
+	 * Assign the position and post under one lock so positions
+	 * match the order the provider queues the Sends, and thus
+	 * completion order.
+	 */
+	spin_lock(&rdma->sc_post_lock);
+	seq = ++rdma->sc_post_seq;
+	ctxt->sc_pos = seq;
 	ret = ib_post_send(rdma->sc_qp, first_wr, &bad_wr);
+	spin_unlock(&rdma->sc_post_lock);
 	if (ret) {
 		ret = svc_rdma_post_send_err(rdma, &cid, bad_wr,
 					     first_wr, sqecount, ret);
@@ -552,6 +570,8 @@ int svc_rdma_post_send(struct svcxprt_rdma *rdma,
 		llist_add(&ctxt->sc_node, &rdma->sc_send_stranded_ctxts);
 		return 0;
 	}
+	if (pos)
+		*pos = seq;
 	return 0;
 }
 
@@ -1082,7 +1102,7 @@ static int svc_rdma_send_reply_msg(struct svcxprt_rdma *rdma,
 		send_wr->opcode = IB_WR_SEND;
 	}
 
-	return svc_rdma_post_send(rdma, sctxt);
+	return svc_rdma_post_send(rdma, sctxt, &rqstp->rq_reply_pos);
 }
 
 /**
@@ -1152,7 +1172,7 @@ void svc_rdma_send_error_msg(struct svcxprt_rdma *rdma,
 	 */
 	sctxt->sc_wr_chain = &sctxt->sc_send_wr;
 	sctxt->sc_sqecount = 1;
-	if (svc_rdma_post_send(rdma, sctxt))
+	if (svc_rdma_post_send(rdma, sctxt, NULL))
 		goto put_ctxt;
 	return;
 
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index df6c3501f97c..fd007296caf0 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -77,6 +77,7 @@ static void svc_rdma_detach(struct svc_xprt *xprt);
 static void svc_rdma_free(struct svc_xprt *xprt);
 static int svc_rdma_has_wspace(struct svc_xprt *xprt);
 static void svc_rdma_kill_temp_xprt(struct svc_xprt *);
+static u64 svc_rdma_acked_pos(struct svc_rqst *rqstp);
 
 static const struct svc_xprt_ops svc_rdma_ops = {
 	.xpo_create = svc_rdma_create,
@@ -89,6 +90,7 @@ static const struct svc_xprt_ops svc_rdma_ops = {
 	.xpo_has_wspace = svc_rdma_has_wspace,
 	.xpo_accept = svc_rdma_accept,
 	.xpo_kill_temp_xprt = svc_rdma_kill_temp_xprt,
+	.xpo_acked_pos = svc_rdma_acked_pos,
 };
 
 struct svc_xprt_class svc_rdma_class = {
@@ -209,6 +211,7 @@ static struct svcxprt_rdma *svc_rdma_create_xprt(struct svc_serv *serv,
 	lockdep_set_class(&cma_xprt->sc_send_lock, &svcrdma_sctx_lock);
 	spin_lock_init(&cma_xprt->sc_rw_ctxt_lock);
 	lockdep_set_class(&cma_xprt->sc_rw_ctxt_lock, &svcrdma_rwctx_lock);
+	spin_lock_init(&cma_xprt->sc_post_lock);
 
 	/*
 	 * Note that this implies that the underlying transport support
@@ -724,6 +727,18 @@ static int svc_rdma_has_wspace(struct svc_xprt *xprt)
 	return 1;
 }
 
+/*
+ * Publish the position as the Send completion raised it. The
+ * xpt_recv_pending count covers only a deferred request. A
+ * retransmit still in flight needs a receive mark as well, and
+ * svcrdma keeps none: xprtrdma does not retransmit on a live
+ * connection.
+ */
+static u64 svc_rdma_acked_pos(struct svc_rqst *rqstp)
+{
+	return atomic64_read(&rqstp->rq_xprt->xpt_acked_pos);
+}
+
 static void svc_rdma_kill_temp_xprt(struct svc_xprt *xprt)
 {
 }

-- 
2.56.0

[PATCH v8 12/14] NFSD: Evict acknowledged DRC entries

[Chuck Lever]

A duplicate reply cache entry exists to answer a retransmit. Once
the client has received the reply, no retransmit of that XID can
follow, yet the entry stays in the cache until RC_EXPIRE elapses or
memory pressure forces it out. On a busy server the cache is mostly
replies the client already has.

Record each reply's transport position in its cache entry from the
sv_reply_sent hook, and compare it with the transport's acknowledged
position when a later request prunes the same bucket. An entry whose
position the peer has acknowledged is evicted before the age and
pressure checks run. Only entries that arrived on the current
request's transport are compared: the prune holds a reference on
that transport alone, so its acknowledged position is the only one
safe to read. The shrinker passes no transport and never evicts on
this basis. Entries from other connections interleave with this
one's in every bucket, and none of them can be acknowledged from
this request, so a walk that stopped at the first unexpired entry
would reach this connection's acknowledged entries only when they
are the oldest in the bucket. Past the first unexpired entry the
walk therefore continues, testing only for acknowledgment, to the
end of the bucket. The hash table is sized for a bucket of about
TARGET_BUCKET_SIZE entries, so the scan is short, and the shrinker,
which passes no transport, still stops at the first unexpired
entry.

Read the acknowledged position through svc_xprt_acked_pos() before
taking cache_lock, since the transport may sleep, and report the
lookup with svc_xprt_recv_done() under the lock, after the insert. A
prune by another thread on the same bucket runs under the same lock,
so the prune either sees this request still counted or runs after
the request has found its entry. A position the transport vouched
for before the lock stays safe under the lock: a retransmit arriving
later follows the acknowledgment.

A transport acknowledgment means the peer's transport holds the
reply, not that the RPC layer above it has read the reply. Two
retransmits can therefore find the entry already gone. A client that
discards unread socket data when it resets a connection retransmits
the call over the new connection. A client whose RPC timer fires
after the reply was acknowledged but before its RPC layer read it
retransmits the call on the same connection. The same-connection
retransmit executes the call again, but TCP delivers the first reply
ahead of the second, so the client completes the call with the first
reply and discards the second as an unknown XID. The re-execution is
visible only to a third party that reused the name in between. A
call still in progress when its retransmit arrives has sent no
reply, so its entry is not acknowledged and the retransmit is
dropped as before. Retention therefore ends at delivery rather than
at consumption. Only NFSv4.1 sessions close that gap.

A call read before its connection closed can still be stalled ahead
of its lookup while the retransmit on the new connection runs and
the retransmit's entry is acknowledged and evicted. The original
call then misses and runs the procedure again. No reply to the
original can reach the client, so drop a cacheable call at lookup
when its transport is dead. A call on a transport whose close NFSD
has not yet seen still runs. The per-transport accounting does not
cover that call.

Between nfsd_cache_update() and the hook the entry is pinned against
eviction, and the nfsd thread keeps its pointer in thread-local
state. Every nfsd_dispatch() path that leaves an entry in RC_DONE
returns 1, so svc_process() runs the hook whether it sends the reply
or drops it after svc_authorise() fails, and the pin is always
cleared. The drop and encode-error paths update with RC_NOCACHE,
which frees the entry without pinning it.

A reply that never reached the wire, or that went out over UDP,
keeps position zero and is retained for the full RC_EXPIRE interval.

Link: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-11-db5e13fd9944@kernel.org
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 fs/nfsd/cache.h    |   6 +++-
 fs/nfsd/nfscache.c | 104 ++++++++++++++++++++++++++++++++++++++++++++++++++---
 fs/nfsd/nfsd.h     |   3 ++
 fs/nfsd/nfssvc.c   |   1 +
 fs/nfsd/trace.h    |   1 +
 5 files changed, 109 insertions(+), 6 deletions(-)

diff --git a/fs/nfsd/cache.h b/fs/nfsd/cache.h
index 589ffe4762d5..894e0b61cbfc 100644
--- a/fs/nfsd/cache.h
+++ b/fs/nfsd/cache.h
@@ -36,8 +36,11 @@ struct nfsd_cacherep {
 	struct list_head	c_lru;
 	unsigned char		c_state,	/* unused, inprog, done */
 				c_type,		/* status, buffer */
-				c_secure : 1;	/* req came from port < 1024 */
+				c_secure : 1,	/* req came from port < 1024 */
+				c_inflight : 1;	/* reply cached but not yet sent */
 	u64			c_xprt;		/* svc_xprt that carried req */
+	u64			c_pos;		/* reply's position on c_xprt;
+						 * 0 = never acknowledged */
 	unsigned long		c_timestamp;
 	union {
 		struct kvec	u_vec;
@@ -88,6 +91,7 @@ int	nfsd_cache_lookup(struct svc_rqst *rqstp, unsigned int start,
 			  unsigned int len, struct nfsd_cacherep **cacherep);
 void	nfsd_cache_update(struct svc_rqst *rqstp, struct nfsd_cacherep *rp,
 			  int cachetype, __be32 *statp);
+void	nfsd_cache_reply_sent(struct svc_rqst *rqstp);
 int	nfsd_reply_cache_stats_show(struct seq_file *m, void *v);
 
 #endif /* NFSCACHE_H */
diff --git a/fs/nfsd/nfscache.c b/fs/nfsd/nfscache.c
index 453ff718ff6c..d7120854dfad 100644
--- a/fs/nfsd/nfscache.c
+++ b/fs/nfsd/nfscache.c
@@ -112,6 +112,8 @@ nfsd_cacherep_alloc(struct svc_rqst *rqstp, __wsum csum,
 		rp->c_key.k_len = rqstp->rq_arg.len;
 		rp->c_key.k_csum = csum;
 		rp->c_xprt = rqstp->rq_xprt->xpt_id;
+		rp->c_inflight = 0;
+		rp->c_pos = 0;
 	}
 	return rp;
 }
@@ -258,22 +260,53 @@ nfsd_cache_bucket_find(__be32 xid, struct nfsd_net *nn)
 	return &nn->drc_hashtbl[hash];
 }
 
+static bool
+nfsd_cacherep_acked(const struct nfsd_cacherep *rp,
+		    const struct svc_xprt *xprt, u64 acked)
+{
+	return rp->c_state == RC_DONE && rp->c_pos && acked &&
+	       rp->c_xprt == xprt->xpt_id && rp->c_pos <= acked;
+}
+
 /*
  * Remove and return no more than @max expired entries in bucket @b.
  * If @max is zero, do not limit the number of removed entries.
+ *
+ * @xprt is the transport of the current request, or NULL when the
+ * caller holds no transport, and @acked is the position
+ * svc_xprt_acked_pos() returned for it, or zero. Only entries that
+ * arrived on @xprt can be evicted as acknowledged: the caller holds
+ * a reference on that transport alone, so it is the only position
+ * safe to read. Entries behind the first unexpired one are younger
+ * still, so from there the walk evicts only acknowledged entries,
+ * which can sit anywhere in the bucket behind other transports'
+ * entries.
+ *
+ * An in-flight entry is never evicted: its svc_rqst still points to
+ * it and nfsd_cache_reply_sent() will write to it.
  */
 static void
 nfsd_prune_bucket_locked(struct nfsd_net *nn, struct nfsd_drc_bucket *b,
-			 unsigned int max, struct list_head *dispose)
+			 unsigned int max, struct list_head *dispose,
+			 struct svc_xprt *xprt, u64 acked)
 {
 	unsigned long expiry = jiffies - RC_EXPIRE;
 	struct nfsd_cacherep *rp, *tmp;
 	unsigned int freed = 0;
+	bool acked_only = false;
 
 	lockdep_assert_held(&b->cache_lock);
 
 	/* The bucket LRU is ordered oldest-first. */
 	list_for_each_entry_safe(rp, tmp, &b->lru_head, c_lru) {
+		if (rp->c_inflight)
+			continue;
+		if (nfsd_cacherep_acked(rp, xprt, acked)) {
+			trace_nfsd_drc_evict_acked(nn, rp);
+			goto evict;
+		}
+		if (acked_only)
+			continue;
 		if (atomic_read(&nn->num_drc_entries) > nn->max_drc_entries) {
 			trace_nfsd_drc_evict_pressure(nn, rp);
 			goto evict;
@@ -282,7 +315,10 @@ nfsd_prune_bucket_locked(struct nfsd_net *nn, struct nfsd_drc_bucket *b,
 			trace_nfsd_drc_evict_expired(nn, rp);
 			goto evict;
 		}
-		break;
+		if (!xprt)
+			break;
+		acked_only = true;
+		continue;
 
 evict:
 		nfsd_cacherep_unlink_locked(nn, b, rp);
@@ -337,7 +373,7 @@ nfsd_reply_cache_scan(struct shrinker *shrink, struct shrink_control *sc)
 			continue;
 
 		spin_lock(&b->cache_lock);
-		nfsd_prune_bucket_locked(nn, b, 0, &dispose);
+		nfsd_prune_bucket_locked(nn, b, 0, &dispose, NULL, 0);
 		spin_unlock(&b->cache_lock);
 
 		freed += nfsd_cacherep_dispose(&dispose);
@@ -488,8 +524,10 @@ int nfsd_cache_lookup(struct svc_rqst *rqstp, unsigned int start,
 	LIST_HEAD(dispose);
 	unsigned int entries;
 	int rtn = RC_DOIT;
+	u64 acked;
 
 	if (type == RC_NOCACHE) {
+		svc_xprt_recv_done(rqstp);
 		nfsd_stats_rc_nocache_inc(nn);
 		goto out;
 	}
@@ -501,17 +539,40 @@ int nfsd_cache_lookup(struct svc_rqst *rqstp, unsigned int start,
 	 * preallocate an entry.
 	 */
 	rp = nfsd_cacherep_alloc(rqstp, csum, nn);
-	if (!rp)
+	if (!rp) {
+		svc_xprt_recv_done(rqstp);
 		goto out;
+	}
+
+	/*
+	 * Read before taking cache_lock: the transport may sleep. A
+	 * position vouched for now stays safe to act on under the lock,
+	 * since a retransmit that arrives later follows the
+	 * acknowledgment.
+	 */
+	acked = svc_xprt_acked_pos(rqstp);
+
+	/*
+	 * No reply can reach the client, and the retransmit on a new
+	 * transport may already have run and had its entry evicted. The
+	 * test follows both sleeps above so a close during either is seen.
+	 */
+	if (svc_xprt_is_dead(rqstp->rq_xprt)) {
+		nfsd_cacherep_free(rp);
+		svc_xprt_recv_done(rqstp);
+		rtn = RC_DROPIT;
+		goto out;
+	}
 
 	b = nfsd_cache_bucket_find(rqstp->rq_xid, nn);
 	spin_lock(&b->cache_lock);
 	found = nfsd_cache_insert(b, rp, nn);
+	svc_xprt_recv_done(rqstp);
 	if (found != rp)
 		goto found_entry;
 	*cacherep = rp;
 	rp->c_state = RC_INPROG;
-	nfsd_prune_bucket_locked(nn, b, 3, &dispose);
+	nfsd_prune_bucket_locked(nn, b, 3, &dispose, rqstp->rq_xprt, acked);
 	spin_unlock(&b->cache_lock);
 
 	nfsd_cacherep_dispose(&dispose);
@@ -589,6 +650,7 @@ void nfsd_cache_update(struct svc_rqst *rqstp, struct nfsd_cacherep *rp,
 		       int cachetype, __be32 *statp)
 {
 	struct nfsd_net *nn = net_generic(SVC_NET(rqstp), nfsd_net_id);
+	struct nfsd_thread_local_info *ntli = rqstp->rq_private;
 	struct kvec	*resv = &rqstp->rq_res.head[0], *cachv;
 	struct nfsd_drc_bucket *b;
 	int		len;
@@ -635,10 +697,42 @@ void nfsd_cache_update(struct svc_rqst *rqstp, struct nfsd_cacherep *rp,
 	rp->c_secure = test_bit(RQ_SECURE, &rqstp->rq_flags);
 	rp->c_type = cachetype;
 	rp->c_state = RC_DONE;
+	rp->c_inflight = 1;
+	ntli->ntli_cacherep = rp;
 	spin_unlock(&b->cache_lock);
 	return;
 }
 
+/**
+ * nfsd_cache_reply_sent - record a reply's transport position
+ * @rqstp: RPC transaction whose reply phase has just ended
+ *
+ * Installed as the svc_serv's sv_reply_sent hook. A transport that
+ * does not publish positions (UDP), or a reply that svc_process()
+ * dropped after dispatch, leaves rq_reply_pos at zero, and the entry
+ * is then never evicted as acknowledged.
+ *
+ * Context: nfsd thread context. Takes and releases cache_lock.
+ */
+void nfsd_cache_reply_sent(struct svc_rqst *rqstp)
+{
+	struct nfsd_thread_local_info *ntli = rqstp->rq_private;
+	struct nfsd_cacherep *rp = ntli->ntli_cacherep;
+	struct nfsd_drc_bucket *b;
+	struct nfsd_net *nn;
+
+	if (!rp)
+		return;
+	ntli->ntli_cacherep = NULL;
+
+	nn = net_generic(SVC_NET(rqstp), nfsd_net_id);
+	b = nfsd_cache_bucket_find(rp->c_key.k_xid, nn);
+	spin_lock(&b->cache_lock);
+	rp->c_pos = rqstp->rq_reply_pos;
+	rp->c_inflight = 0;
+	spin_unlock(&b->cache_lock);
+}
+
 static int
 nfsd_cache_append(struct svc_rqst *rqstp, struct kvec *data)
 {
diff --git a/fs/nfsd/nfsd.h b/fs/nfsd/nfsd.h
index 8b4bed37b46b..b4c8cafbe373 100644
--- a/fs/nfsd/nfsd.h
+++ b/fs/nfsd/nfsd.h
@@ -52,9 +52,12 @@ extern atomic_t			nfsd_th_cnt;		/* number of available threads */
 
 extern const struct seq_operations nfs_exports_op;
 
+struct nfsd_cacherep;
+
 struct nfsd_thread_local_info {
 	struct nfs4_client	**ntli_lease_breaker;
 	int			ntli_cachetype;
+	struct nfsd_cacherep	*ntli_cacherep;
 };
 
 /*
diff --git a/fs/nfsd/nfssvc.c b/fs/nfsd/nfssvc.c
index d945944c1514..d32c70e31f8b 100644
--- a/fs/nfsd/nfssvc.c
+++ b/fs/nfsd/nfssvc.c
@@ -697,6 +697,7 @@ int nfsd_create_serv(struct net *net, bool no_rpcbind)
 		goto out_put;
 	}
 	serv->sv_tls = true;
+	serv->sv_reply_sent = nfsd_cache_reply_sent;
 
 	/* svc_bind() reads this, so set it first. */
 	serv->sv_no_rpcbind = no_rpcbind;
diff --git a/fs/nfsd/trace.h b/fs/nfsd/trace.h
index 41a9407e3efa..dfb784a39712 100644
--- a/fs/nfsd/trace.h
+++ b/fs/nfsd/trace.h
@@ -1748,6 +1748,7 @@ DEFINE_EVENT(nfsd_drc_entry_class, nfsd_drc_##name,		\
 
 DEFINE_NFSD_DRC_ENTRY_EVENT(evict_pressure);
 DEFINE_NFSD_DRC_ENTRY_EVENT(evict_expired);
+DEFINE_NFSD_DRC_ENTRY_EVENT(evict_acked);
 
 TRACE_EVENT(nfsd_cb_args,
 	TP_PROTO(

-- 
2.56.0

[PATCH v8 13/14] NFSD: Remove DRC checksum and payload_misses stat

[Chuck Lever]

nfsd_cache_csum() costs CPU on every non-idempotent request. It also
prevents the use of zero-copy RDMA receives for WRITE and SYMLINK
because the payload must be in the server's memory before the DRC
lookup can run.

Commit 01a7decf7593 ("nfsd: keep a checksum of the first 256 bytes
of request") added the checksum when a growing cache made XID
collisions easier to hit. The checksum is the only guard against a
reused XID: two calls with the same procedure and equal-length
arguments match on every other key field.

Acknowledgment-based eviction retires a TCP or RDMA entry once the
transport confirms delivery and a later request from the same
connection visits its bucket. Three kinds of entry outlast that: the
entries a connection has issued since it last visited each bucket,
entries from a connection the client has since closed, which no
later request can evict because their transport is gone, and UDP
entries. Those wait out RC_EXPIRE, or longer in a bucket that no
miss visits, since only a miss prunes expired entries. A collision
needs the same XID from the same address and port while the entry
lasts. The Linux NFS client's XIDs are sequential from a seed drawn
with get_random_u32(), so an XID recurs only after 2^32 calls, and
a rebooted client does not replay its previous sequence. A client
that restarts its sequence from a fixed value and rebinds its
previous source port while the entry lasts is unguarded, but a
client that repeats a live XID cannot reliably match replies to its
own calls anyway.

A call whose XID matches an entry sent on the same connection is
therefore a retransmit. TCP and RDMA deliver the queued reply or
drop the connection, so drop the call as though the entry were
still in progress rather than replaying it. A match from another
connection, or from UDP, is replayed as before.

Remove the checksum, the nfsd_drc_mismatch tracepoint, and the
payload_misses stat that counted checksum-detected collisions.
nfsd_dispatch() no longer snapshots the argument stream before
decoding, and with k_csum gone, struct nfsd_cacherep shrinks to 136
bytes.

The "payload misses" line disappears from
/proc/fs/nfsd/reply_cache_stats. No known userspace tool parses it.

Reviewed-by: NeilBrown <neil@brown.name>
Link: https://patch.msgid.link/20260921-duplicate-reply-cache-v6-12-db5e13fd9944@kernel.org
Signed-off-by: Chuck Lever <cel@kernel.org>
---
 .../ABI/testing/procfs-nfsd-reply_cache_stats      | 11 +--
 fs/nfsd/cache.h                                    |  9 +-
 fs/nfsd/netns.h                                    |  2 -
 fs/nfsd/nfscache.c                                 | 96 ++++------------------
 fs/nfsd/nfssvc.c                                   | 10 +--
 fs/nfsd/stats.h                                    |  5 --
 fs/nfsd/trace.h                                    | 24 ------
 7 files changed, 23 insertions(+), 134 deletions(-)

diff --git a/Documentation/ABI/testing/procfs-nfsd-reply_cache_stats b/Documentation/ABI/testing/procfs-nfsd-reply_cache_stats
index 57ed5f8e6597..7a22ac7c1ccd 100644
--- a/Documentation/ABI/testing/procfs-nfsd-reply_cache_stats
+++ b/Documentation/ABI/testing/procfs-nfsd-reply_cache_stats
@@ -19,19 +19,16 @@ Description:
 		cache misses             s64     Requests not found in cache
 		not cached               s64     Idempotent requests that
 		                                 bypass the cache
-		payload misses           s64     XID matched but request
-		                                 checksum did not
 		longest chain len        u32     Longest hash chain observed
 		cachesize at longest     u32     Cache size when longest
 		                                 chain was recorded
 		=======================  ======  ==========================
 
 		Counter fields (cache hits, cache misses, not cached,
-		payload misses, mem usage) are maintained with per-cpu
-		counters and may briefly show stale values under
-		concurrent load. There is no way to reset these
-		counters; consumers should compute rates by sampling
-		over time.
+		mem usage) are maintained with per-cpu counters and
+		may briefly show stale values under concurrent load.
+		There is no way to reset these counters; consumers
+		should compute rates by sampling over time.
 
 		New fields may be appended in future kernels. Parsers
 		should match on field name, not line position.
diff --git a/fs/nfsd/cache.h b/fs/nfsd/cache.h
index 894e0b61cbfc..3491a10524d7 100644
--- a/fs/nfsd/cache.h
+++ b/fs/nfsd/cache.h
@@ -22,9 +22,7 @@ struct nfsd_net;
  */
 struct nfsd_cacherep {
 	struct {
-		/* Keep often-read xid, csum in the same cache line: */
 		__be32			k_xid;
-		__wsum			k_csum;
 		u32			k_proc;
 		u32			k_prot;
 		u32			k_vers;
@@ -80,15 +78,12 @@ enum {
 /* Cache entries expire after this time period */
 #define RC_EXPIRE		(120 * HZ)
 
-/* Checksum this amount of the request */
-#define RC_CSUMLEN		(256U)
-
 int	nfsd_drc_slab_create(void);
 void	nfsd_drc_slab_free(void);
 int	nfsd_reply_cache_init(struct nfsd_net *);
 void	nfsd_reply_cache_shutdown(struct nfsd_net *);
-int	nfsd_cache_lookup(struct svc_rqst *rqstp, unsigned int start,
-			  unsigned int len, struct nfsd_cacherep **cacherep);
+int	nfsd_cache_lookup(struct svc_rqst *rqstp,
+			  struct nfsd_cacherep **cacherep);
 void	nfsd_cache_update(struct svc_rqst *rqstp, struct nfsd_cacherep *rp,
 			  int cachetype, __be32 *statp);
 void	nfsd_cache_reply_sent(struct svc_rqst *rqstp);
diff --git a/fs/nfsd/netns.h b/fs/nfsd/netns.h
index 16be473f2c04..cb236636bf48 100644
--- a/fs/nfsd/netns.h
+++ b/fs/nfsd/netns.h
@@ -40,8 +40,6 @@ enum nfsd_net_flag {
 };
 
 enum {
-	/* cache misses due only to checksum comparison failures */
-	NFSD_STATS_PAYLOAD_MISSES,
 	/* amount of memory (in bytes) currently consumed by the DRC */
 	NFSD_STATS_DRC_MEM_USAGE,
 	NFSD_STATS_RC_HITS,		/* repcache hits */
diff --git a/fs/nfsd/nfscache.c b/fs/nfsd/nfscache.c
index d7120854dfad..24621db25615 100644
--- a/fs/nfsd/nfscache.c
+++ b/fs/nfsd/nfscache.c
@@ -13,10 +13,8 @@
 #include <linux/slab.h>
 #include <linux/vmalloc.h>
 #include <linux/sunrpc/addr.h>
-#include <linux/highmem.h>
 #include <linux/log2.h>
 #include <linux/hash.h>
-#include <net/checksum.h>
 
 #include "nfsd.h"
 #include "nfserr.h"
@@ -90,8 +88,7 @@ nfsd_hashsize(unsigned int limit)
 }
 
 static struct nfsd_cacherep *
-nfsd_cacherep_alloc(struct svc_rqst *rqstp, __wsum csum,
-		    struct nfsd_net *nn)
+nfsd_cacherep_alloc(struct svc_rqst *rqstp, struct nfsd_net *nn)
 {
 	struct nfsd_cacherep *rp;
 
@@ -110,7 +107,6 @@ nfsd_cacherep_alloc(struct svc_rqst *rqstp, __wsum csum,
 		rp->c_key.k_prot = rqstp->rq_prot;
 		rp->c_key.k_vers = rqstp->rq_vers;
 		rp->c_key.k_len = rqstp->rq_arg.len;
-		rp->c_key.k_csum = csum;
 		rp->c_xprt = rqstp->rq_xprt->xpt_id;
 		rp->c_inflight = 0;
 		rp->c_pos = 0;
@@ -383,68 +379,10 @@ nfsd_reply_cache_scan(struct shrinker *shrink, struct shrink_control *sc)
 	return freed;
 }
 
-/**
- * nfsd_cache_csum - Checksum incoming NFS Call arguments
- * @buf: buffer containing a whole RPC Call message
- * @start: starting byte of the NFS Call header
- * @remaining: size of the NFS Call header, in bytes
- *
- * Compute a weak checksum of the leading bytes of an NFS procedure
- * call header to help verify that a retransmitted Call matches an
- * entry in the duplicate reply cache.
- *
- * To avoid assumptions about how the RPC message is laid out in
- * @buf and what else it might contain (eg, a GSS MIC suffix), the
- * caller passes us the exact location and length of the NFS Call
- * header.
- *
- * Returns a 32-bit checksum value, as defined in RFC 793.
- */
-static __wsum nfsd_cache_csum(struct xdr_buf *buf, unsigned int start,
-			      unsigned int remaining)
-{
-	unsigned int base, len;
-	struct xdr_buf subbuf;
-	__wsum csum = 0;
-	void *p;
-	int idx;
-
-	if (remaining > RC_CSUMLEN)
-		remaining = RC_CSUMLEN;
-	if (xdr_buf_subsegment(buf, &subbuf, start, remaining))
-		return csum;
-
-	/* rq_arg.head first */
-	if (subbuf.head[0].iov_len) {
-		len = min_t(unsigned int, subbuf.head[0].iov_len, remaining);
-		csum = csum_partial(subbuf.head[0].iov_base, len, csum);
-		remaining -= len;
-	}
-
-	/* Continue into page array */
-	idx = subbuf.page_base / PAGE_SIZE;
-	base = subbuf.page_base & ~PAGE_MASK;
-	while (remaining) {
-		p = page_address(subbuf.pages[idx]) + base;
-		len = min_t(unsigned int, PAGE_SIZE - base, remaining);
-		csum = csum_partial(p, len, csum);
-		remaining -= len;
-		base = 0;
-		++idx;
-	}
-	return csum;
-}
-
 static int
 nfsd_cache_key_cmp(const struct nfsd_cacherep *key,
-		   const struct nfsd_cacherep *rp, struct nfsd_net *nn)
+		   const struct nfsd_cacherep *rp)
 {
-	if (key->c_key.k_xid == rp->c_key.k_xid &&
-	    key->c_key.k_csum != rp->c_key.k_csum) {
-		nfsd_stats_payload_misses_inc(nn);
-		trace_nfsd_drc_mismatch(nn, key, rp);
-	}
-
 	return memcmp(&key->c_key, &rp->c_key, sizeof(key->c_key));
 }
 
@@ -468,7 +406,7 @@ nfsd_cache_insert(struct nfsd_drc_bucket *b, struct nfsd_cacherep *key,
 		parent = *p;
 		rp = rb_entry(parent, struct nfsd_cacherep, c_node);
 
-		cmp = nfsd_cache_key_cmp(key, rp, nn);
+		cmp = nfsd_cache_key_cmp(key, rp);
 		if (cmp < 0)
 			p = &parent->rb_left;
 		else if (cmp > 0)
@@ -497,28 +435,23 @@ nfsd_cache_insert(struct nfsd_drc_bucket *b, struct nfsd_cacherep *key,
 /**
  * nfsd_cache_lookup - Find an entry in the duplicate reply cache
  * @rqstp: Incoming Call to find
- * @start: starting byte in @rqstp->rq_arg of the NFS Call header
- * @len: size of the NFS Call header, in bytes
  * @cacherep: OUT: DRC entry for this request
  *
- * Try to find an entry matching the current call in the cache. When none
- * is found, we try to grab the oldest expired entry off the LRU list. If
- * a suitable one isn't there, then drop the cache_lock and allocate a
- * new one, then search again in case one got inserted while this thread
- * didn't hold the lock.
+ * On a miss, the entry created for this call is returned in
+ * @cacherep. On a hit, the cached reply is encoded into @rqstp's
+ * response.
  *
  * Return values:
  *   %RC_DOIT: Process the request normally
  *   %RC_REPLY: Reply from cache
  *   %RC_DROPIT: Do not process the request further
  */
-int nfsd_cache_lookup(struct svc_rqst *rqstp, unsigned int start,
-		      unsigned int len, struct nfsd_cacherep **cacherep)
+int nfsd_cache_lookup(struct svc_rqst *rqstp,
+		      struct nfsd_cacherep **cacherep)
 {
 	struct nfsd_net		*nn = net_generic(SVC_NET(rqstp), nfsd_net_id);
 	struct nfsd_thread_local_info *ntli = rqstp->rq_private;
 	struct nfsd_cacherep	*rp, *found;
-	__wsum			csum;
 	struct nfsd_drc_bucket	*b;
 	int type = ntli->ntli_cachetype;
 	LIST_HEAD(dispose);
@@ -532,13 +465,11 @@ int nfsd_cache_lookup(struct svc_rqst *rqstp, unsigned int start,
 		goto out;
 	}
 
-	csum = nfsd_cache_csum(&rqstp->rq_arg, start, len);
-
 	/*
 	 * Since the common case is a cache miss followed by an insert,
 	 * preallocate an entry.
 	 */
-	rp = nfsd_cacherep_alloc(rqstp, csum, nn);
+	rp = nfsd_cacherep_alloc(rqstp, nn);
 	if (!rp) {
 		svc_xprt_recv_done(rqstp);
 		goto out;
@@ -593,6 +524,13 @@ int nfsd_cache_lookup(struct svc_rqst *rqstp, unsigned int start,
 	if (rp->c_state == RC_INPROG)
 		goto out_trace;
 
+	/*
+	 * The connection that carried the reply delivers it, so a
+	 * retransmit on that connection needs no replay.
+	 */
+	if (rp->c_pos && rp->c_xprt == rqstp->rq_xprt->xpt_id)
+		goto out_trace;
+
 	/* From the hall of fame of impractical attacks:
 	 * Is this a user who tries to snoop on the cache? */
 	rtn = RC_DOIT;
@@ -768,8 +706,6 @@ int nfsd_reply_cache_stats_show(struct seq_file *m, void *v)
 		   percpu_counter_sum_positive(&nn->counter[NFSD_STATS_RC_MISSES]));
 	seq_printf(m, "not cached:            %lld\n",
 		   percpu_counter_sum_positive(&nn->counter[NFSD_STATS_RC_NOCACHE]));
-	seq_printf(m, "payload misses:        %lld\n",
-		   percpu_counter_sum_positive(&nn->counter[NFSD_STATS_PAYLOAD_MISSES]));
 	seq_printf(m, "longest chain len:     %u\n", nn->longest_chain);
 	seq_printf(m, "cachesize at longest:  %u\n", nn->longest_chain_cachesize);
 	return 0;
diff --git a/fs/nfsd/nfssvc.c b/fs/nfsd/nfssvc.c
index d32c70e31f8b..33ee0baaa140 100644
--- a/fs/nfsd/nfssvc.c
+++ b/fs/nfsd/nfssvc.c
@@ -1094,7 +1094,6 @@ int nfsd_dispatch(struct svc_rqst *rqstp)
 	const struct svc_procedure *proc = rqstp->rq_procinfo;
 	__be32 *statp = rqstp->rq_accept_statp;
 	struct nfsd_cacherep *rp;
-	unsigned int start, len;
 	__be32 *nfs_reply;
 
 	/*
@@ -1103,13 +1102,6 @@ int nfsd_dispatch(struct svc_rqst *rqstp)
 	 */
 	ntli->ntli_cachetype = proc->pc_cachetype;
 
-	/*
-	 * ->pc_decode advances the argument stream past the NFS
-	 * Call header, so grab the header's starting location and
-	 * size now for the call to nfsd_cache_lookup().
-	 */
-	start = xdr_stream_pos(&rqstp->rq_arg_stream);
-	len = xdr_stream_remaining(&rqstp->rq_arg_stream);
 	if (!proc->pc_decode(rqstp, &rqstp->rq_arg_stream))
 		goto out_decode_err;
 
@@ -1123,7 +1115,7 @@ int nfsd_dispatch(struct svc_rqst *rqstp)
 	smp_store_release(&rqstp->rq_status_counter, rqstp->rq_status_counter | 1);
 
 	rp = NULL;
-	switch (nfsd_cache_lookup(rqstp, start, len, &rp)) {
+	switch (nfsd_cache_lookup(rqstp, &rp)) {
 	case RC_DOIT:
 		break;
 	case RC_REPLY:
diff --git a/fs/nfsd/stats.h b/fs/nfsd/stats.h
index d88d2229a4c7..6b95651cd55f 100644
--- a/fs/nfsd/stats.h
+++ b/fs/nfsd/stats.h
@@ -130,11 +130,6 @@ static inline void nfsd_stats_io_read_storage_add(struct svc_export *exp,
 					 amount);
 }
 
-static inline void nfsd_stats_payload_misses_inc(struct nfsd_net *nn)
-{
-	percpu_counter_inc(&nn->counter[NFSD_STATS_PAYLOAD_MISSES]);
-}
-
 /**
  * nfsd_stats_drc_mem_usage_add - Add memory used by a cache item
  * @nn: target network namespace
diff --git a/fs/nfsd/trace.h b/fs/nfsd/trace.h
index dfb784a39712..fc70f542c40b 100644
--- a/fs/nfsd/trace.h
+++ b/fs/nfsd/trace.h
@@ -1688,30 +1688,6 @@ TRACE_EVENT(nfsd_drc_found,
 
 );
 
-TRACE_EVENT(nfsd_drc_mismatch,
-	TP_PROTO(
-		const struct nfsd_net *nn,
-		const struct nfsd_cacherep *key,
-		const struct nfsd_cacherep *rp
-	),
-	TP_ARGS(nn, key, rp),
-	TP_STRUCT__entry(
-		__field(unsigned long long, boot_time)
-		__field(u32, xid)
-		__field(u32, cached)
-		__field(u32, ingress)
-	),
-	TP_fast_assign(
-		__entry->boot_time = nn->boot_time;
-		__entry->xid = be32_to_cpu(key->c_key.k_xid);
-		__entry->cached = (__force u32)key->c_key.k_csum;
-		__entry->ingress = (__force u32)rp->c_key.k_csum;
-	),
-	TP_printk("boot_time=%16llx xid=0x%08x cached-csum=0x%08x ingress-csum=0x%08x",
-		__entry->boot_time, __entry->xid, __entry->cached,
-		__entry->ingress)
-);
-
 DECLARE_EVENT_CLASS(nfsd_drc_entry_class,
 	TP_PROTO(
 		const struct nfsd_net *nn,

-- 
2.56.0

[PATCH v8 14/14] SUNRPC: Advance the acked position on TLS connections

[Chuck Lever]

The DRC retires an entry once its reply lies at or below the
transport's xpt_acked_pos, the position svc_tcp_update_acked()
advances as TCP acknowledgments arrive. When XPT_TLS_SESSION is set,
svc_tcp_update_acked() leaves xpt_acked_pos alone, because kTLS can
return a full count from sendmsg() while a record is still waiting to
reach TCP. DRC entries on TLS connections therefore never retire by
acknowledgment.

Advance xpt_acked_pos on a TLS connection when no record is open,
awaiting encryption, or parked on a short TCP send. Every field the
check reads is declared in include/net/tls.h, so the predicate lives
beside its one caller rather than in net/tls. After the change,
a TLS connection takes the socket lock once per reply, as a plain
TCP connection already does. Record in svcsock_tcp_acked why a reply
published no position.

On the receive side, copied_seq does not track plaintext under kTLS.
tls_strp_msg_consume() advances copied_seq by the whole ciphertext
record once the record is decrypted, before svc_tcp_recv_actor()
sees any of its plaintext, and in copy mode the strparser advances
copied_seq while pulling the ciphertext in. A record that spans the
end of one call and the start of a retransmit leaves the
retransmit's plaintext on rx_list, uncounted in xpt_recv_pending,
with copied_seq already at the mark. Withhold the position while
sk_is_readable() reports plaintext on rx_list or a record ready in
the strparser. A record the strparser is still copying in is neither
readable nor ready, so withhold the position while the strparser's
copy_mode is set as well. The strparser fields are declared in
include/net/tls.h with the socket lock named as their guard.

The close_notify alert and the FIN sent at detach go out without
xpt_mutex. Both are non-blocking and never drop the socket lock
mid-push, so their bytes either reach write_seq and lower the position,
or park and hold it.

Signed-off-by: Chuck Lever <cel@kernel.org>
---
 include/linux/sunrpc/svcsock.h |   9 ++++
 include/trace/events/sunrpc.h  |  23 +++++++--
 net/sunrpc/svcsock.c           | 111 +++++++++++++++++++++++++++++++++++------
 3 files changed, 125 insertions(+), 18 deletions(-)

diff --git a/include/linux/sunrpc/svcsock.h b/include/linux/sunrpc/svcsock.h
index 0fa62f9c6465..bd5cd79c1870 100644
--- a/include/linux/sunrpc/svcsock.h
+++ b/include/linux/sunrpc/svcsock.h
@@ -49,12 +49,21 @@ struct svc_sock {
 	u32			sk_acked_rcv;
 
 	struct completion	sk_handshake_done;
+	/* tls ULP installed, recorded once; the ULP has no removal path */
+	bool			sk_tls_ulp;
 
 	/* received data */
 	unsigned long		sk_maxpages;
 	struct page *		sk_pages[] __counted_by(sk_maxpages);
 };
 
+/* Why svc_tcp_update_acked() did or did not publish a position */
+enum svc_tcp_acked_reason {
+	SVC_TCP_ACKED_PUBLISHED,
+	SVC_TCP_ACKED_NO_ULP,
+	SVC_TCP_ACKED_TX_PENDING,
+};
+
 static inline u32 svc_sock_reclen(struct svc_sock *svsk)
 {
 	return be32_to_cpu(svsk->sk_marker) & RPC_FRAGMENT_SIZE_MASK;
diff --git a/include/trace/events/sunrpc.h b/include/trace/events/sunrpc.h
index a5d914376627..a7e96e49e0d0 100644
--- a/include/trace/events/sunrpc.h
+++ b/include/trace/events/sunrpc.h
@@ -10,6 +10,7 @@
 #include <linux/sunrpc/svc.h>
 #include <linux/sunrpc/xprtsock.h>
 #include <linux/sunrpc/svc_xprt.h>
+#include <linux/sunrpc/svcsock.h>
 #include <net/tcp_states.h>
 #include <linux/net.h>
 #include <linux/tracepoint.h>
@@ -2349,6 +2350,16 @@ TRACE_EVENT(svcsock_tcp_recv_short,
 	)
 );
 
+TRACE_DEFINE_ENUM(SVC_TCP_ACKED_PUBLISHED);
+TRACE_DEFINE_ENUM(SVC_TCP_ACKED_NO_ULP);
+TRACE_DEFINE_ENUM(SVC_TCP_ACKED_TX_PENDING);
+
+#define show_svc_tcp_acked_reason(reason)				\
+	__print_symbolic(reason,					\
+		{ SVC_TCP_ACKED_PUBLISHED,	"published" },		\
+		{ SVC_TCP_ACKED_NO_ULP,		"no_ulp" },		\
+		{ SVC_TCP_ACKED_TX_PENDING,	"tx_pending" })
+
 TRACE_EVENT(svcsock_tcp_acked,
 	TP_PROTO(
 		const struct svc_xprt *xprt,
@@ -2356,10 +2367,12 @@ TRACE_EVENT(svcsock_tcp_acked,
 		u32 write_seq,
 		u32 snd_una,
 		u64 candidate,
-		u64 acked_pos
+		u64 acked_pos,
+		enum svc_tcp_acked_reason reason
 	),
 
-	TP_ARGS(xprt, send_pos, write_seq, snd_una, candidate, acked_pos),
+	TP_ARGS(xprt, send_pos, write_seq, snd_una, candidate, acked_pos,
+		reason),
 
 	TP_STRUCT__entry(
 		__field(u64, xpt_id)
@@ -2368,6 +2381,7 @@ TRACE_EVENT(svcsock_tcp_acked,
 		__field(u64, acked_pos)
 		__field(u32, write_seq)
 		__field(u32, snd_una)
+		__field(unsigned int, reason)
 		__string(addr, xprt->xpt_remotebuf)
 	),
 
@@ -2378,13 +2392,14 @@ TRACE_EVENT(svcsock_tcp_acked,
 		__entry->acked_pos = acked_pos;
 		__entry->write_seq = write_seq;
 		__entry->snd_una = snd_una;
+		__entry->reason = reason;
 		__assign_str(addr);
 	),
 
-	TP_printk("addr=%s xpt_id=%llu send_pos=%llu write_seq=%u snd_una=%u candidate=%llu acked_pos=%llu",
+	TP_printk("addr=%s xpt_id=%llu send_pos=%llu write_seq=%u snd_una=%u candidate=%llu acked_pos=%llu reason=%s",
 		__get_str(addr), __entry->xpt_id, __entry->send_pos,
 		__entry->write_seq, __entry->snd_una, __entry->candidate,
-		__entry->acked_pos
+		__entry->acked_pos, show_svc_tcp_acked_reason(__entry->reason)
 	)
 );
 
diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index 4eaa8eb4d537..79d8d4e2c90d 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -36,6 +36,7 @@
 #include <net/udp.h>
 #include <net/tcp.h>
 #include <net/tcp_states.h>
+#include <net/tls.h>
 #include <net/tls_prot.h>
 #include <net/handshake.h>
 #include <linux/uaccess.h>
@@ -409,6 +410,14 @@ static void svc_tcp_kill_temp_xprt(struct svc_xprt *xprt)
 	sock_no_linger(svsk->sk_sock->sk);
 }
 
+/* XPT_TLS_SESSION reflects tlshd's report, not an installed TLS ULP. */
+static bool svc_tcp_has_tls_ulp(struct sock *sk)
+{
+	const struct inet_connection_sock *icsk = inet_csk(sk);
+
+	return icsk->icsk_ulp_ops && !strcmp(icsk->icsk_ulp_ops->name, "tls");
+}
+
 /**
  * svc_tcp_handshake_done - Handshake completion handler
  * @data: address of xprt to wake
@@ -426,8 +435,13 @@ static void svc_tcp_handshake_done(void *data, int status, key_serial_t peerid)
 	struct svc_sock *svsk = container_of(xprt, struct svc_sock, sk_xprt);
 
 	if (!status) {
+		struct sock *sk = svsk->sk_sk;
+
 		if (peerid != TLS_NO_PEERID)
 			set_bit(XPT_PEER_AUTH, &xprt->xpt_flags);
+		lock_sock(sk);
+		WRITE_ONCE(svsk->sk_tls_ulp, svc_tcp_has_tls_ulp(sk));
+		release_sock(sk);
 		set_bit(XPT_TLS_SESSION, &xprt->xpt_flags);
 	}
 	clear_bit(XPT_HANDSHAKE, &xprt->xpt_flags);
@@ -1401,9 +1415,60 @@ static int svc_tcp_sendmsg(struct svc_sock *svsk, struct svc_rqst *rqstp,
 }
 
 /*
- * Bytes the socket has not seen acknowledged all belong to the most
- * recent replies, so sk_send_pos less that count is the acknowledged
- * position.
+ * A false return means every byte sendmsg() has accepted has reached the
+ * TCP write queue, provided no other blocking sender is on the socket. With
+ * TLS_HW, a sender can drop the socket lock inside tcp_sendmsg_locked()
+ * while its record sits in neither partially_sent_record nor an
+ * open record. TLS_SW keeps the record on tx_list until the push
+ * succeeds. TLS_HW_RECORD keeps its send accounting in the TOE driver,
+ * beyond this check's reach, so TLS_HW_RECORD always counts as pending.
+ *
+ * tx_lock is the documented guard for the partially_sent_* fields, but
+ * every store to them and to tx_list runs with the socket owned: sendmsg(),
+ * the tx worker, and the write-space callback, which the backlog defers
+ * while a process owns the socket. Owning the socket therefore reads a
+ * consistent snapshot without contending with the worker for tx_lock.
+ */
+static bool svc_tls_tx_pending(const struct sock *sk)
+{
+	struct tls_context *ctx = tls_get_ctx(sk);
+	struct tls_sw_context_tx *sw;
+
+	sock_owned_by_me(sk);
+	if (ctx->partially_sent_record || ctx->pending_open_record_frags)
+		return true;
+	switch (ctx->tx_conf) {
+	case TLS_SW:
+		sw = tls_sw_ctx_tx(ctx);
+		return sw->open_rec || !list_empty(&sw->tx_list);
+	case TLS_HW:
+		return false;
+	default:
+		return true;
+	}
+}
+
+/*
+ * In copy mode the strparser pulls a record's segments out of the
+ * receive queue as they arrive, advancing copied_seq, and reports the
+ * record readable only once the last segment is in. copy_mode stays
+ * set until the record is consumed. include/net/tls.h documents the
+ * bitfield as serialized by the socket lock.
+ */
+static bool svc_tls_rx_pending(const struct sock *sk)
+{
+	struct tls_context *ctx = tls_get_ctx(sk);
+
+	sock_owned_by_me(sk);
+	return tls_sw_ctx_rx(ctx) && tls_sw_ctx_rx(ctx)->strp.copy_mode;
+}
+
+/*
+ * Unacknowledged bytes follow every older reply, and each reply occupies at
+ * least as many wire bytes as sk_send_pos counts for it. sk_send_pos less
+ * the unacknowledged count is therefore at or below the position the client
+ * has acknowledged. Backchannel calls, TLS framing, alerts, the FIN sent at
+ * detach, and a stale snd_una only lower the result.
  *
  * A call the peer sent before it acknowledged a reply precedes the
  * acknowledgment in the stream. When a segment of the call is lost on
@@ -1418,13 +1483,17 @@ static int svc_tcp_sendmsg(struct svc_sock *svsk, struct svc_rqst *rqstp,
  * from moving between the reads and orders them against
  * svc_tcp_acked_pos().
  *
- * Under kTLS, tls_sw_sendmsg() can return the full plaintext count
- * with part of a record still waiting for socket write space, leaving
- * write_seq short of the reply. Whether a record is pending is private
- * to net/tls, so a TLS session publishes no acknowledged position.
+ * kTLS can return a full count from sendmsg() with a record not yet
+ * handed to TCP, and the tls tx worker advances write_seq outside
+ * xpt_mutex. Evaluate svc_tls_tx_pending() and read both sequence numbers
+ * under one lock_sock() so all three describe the same instant. xpt_mutex
+ * keeps every other blocking sender off the socket, as svc_tls_tx_pending()
+ * requires.
+ *
  * tlshd installs the ULP before XPT_TLS_SESSION reports the session,
- * so neither does a reply sent while XPT_HANDSHAKE is set. Test
- * XPT_HANDSHAKE first: it is cleared after XPT_TLS_SESSION is set.
+ * so a reply sent while XPT_HANDSHAKE is set publishes no position. Test
+ * XPT_HANDSHAKE first: svc_tcp_handshake_done() sets XPT_TLS_SESSION before
+ * clearing XPT_HANDSHAKE.
  */
 static void svc_tcp_update_acked(struct svc_sock *svsk)
 {
@@ -1432,15 +1501,21 @@ static void svc_tcp_update_acked(struct svc_sock *svsk)
 	struct tcp_sock *tp = tcp_sk(sk);
 	u64 unacked, acked = 0, acked_pos;
 	u32 write_seq, snd_una, mark;
+	enum svc_tcp_acked_reason reason = SVC_TCP_ACKED_PUBLISHED;
+	bool tls;
 
-	if (test_bit(XPT_HANDSHAKE, &svsk->sk_xprt.xpt_flags) ||
-	    test_bit(XPT_TLS_SESSION, &svsk->sk_xprt.xpt_flags))
+	if (test_bit(XPT_HANDSHAKE, &svsk->sk_xprt.xpt_flags))
 		return;
+	tls = test_bit(XPT_TLS_SESSION, &svsk->sk_xprt.xpt_flags);
+	if (tls && !READ_ONCE(svsk->sk_tls_ulp))
+		reason = SVC_TCP_ACKED_NO_ULP;
 	lock_sock(sk);
+	if (tls && reason == SVC_TCP_ACKED_PUBLISHED && svc_tls_tx_pending(sk))
+		reason = SVC_TCP_ACKED_TX_PENDING;
 	write_seq = READ_ONCE(tp->write_seq);
 	snd_una = READ_ONCE(tp->snd_una);
 	unacked = write_seq - snd_una;
-	if (unacked < svsk->sk_send_pos)
+	if (reason == SVC_TCP_ACKED_PUBLISHED && unacked < svsk->sk_send_pos)
 		acked = svsk->sk_send_pos - unacked;
 	if (svc_xprt_advance_acked(&svsk->sk_xprt, acked)) {
 		if (RB_EMPTY_ROOT(&tp->out_of_order_queue))
@@ -1454,7 +1529,7 @@ static void svc_tcp_update_acked(struct svc_sock *svsk)
 	acked_pos = atomic64_read(&svsk->sk_xprt.xpt_acked_pos);
 	release_sock(sk);
 	trace_svcsock_tcp_acked(&svsk->sk_xprt, svsk->sk_send_pos, write_seq,
-				snd_una, acked, acked_pos);
+				snd_una, acked, acked_pos, reason);
 }
 
 static int svc_tcp_recv_pending(const struct svc_rqst *rqstp)
@@ -1475,6 +1550,13 @@ static int svc_tcp_recv_pending(const struct svc_rqst *rqstp)
  * from its first octet, so a partially read retransmit withholds the
  * position too.
  *
+ * Under kTLS, copied_seq advances by the whole ciphertext record when
+ * the strparser consumes it, before svc_tcp_recv_actor() sees any of
+ * its plaintext. Plaintext left on rx_list and a record ready in the
+ * strparser withhold the position as well; sk_is_readable() reports
+ * both. A record the strparser is still copying in is neither, and
+ * svc_tls_rx_pending() withholds the position for it.
+ *
  * Context: process context. May take and release the socket lock.
  * Return: xpt_acked_pos, or 0 while it cannot be acted on.
  */
@@ -1494,7 +1576,8 @@ static u64 svc_tcp_acked_pos(struct svc_rqst *rqstp)
 
 	lock_sock(sk);
 	pos = atomic64_read(&xprt->xpt_acked_pos);
-	if (svc_tcp_recv_pending(rqstp) ||
+	if (svc_tcp_recv_pending(rqstp) || sk_is_readable(sk) ||
+	    (READ_ONCE(svsk->sk_tls_ulp) && svc_tls_rx_pending(sk)) ||
 	    before(READ_ONCE(tcp_sk(sk)->copied_seq), svsk->sk_acked_rcv))
 		pos = 0;
 	release_sock(sk);

-- 
2.56.0