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 --- 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