[PATCH 0/5] SUNRPC: Receive svcsock TCP records with ->read_sock

[PATCH 0/5] SUNRPC: Receive svcsock TCP records with ->read_sock

[Chuck Lever]

kTLS splits a TLS control record across two destinations. The record
type arrives in a control-message buffer. An alert's level and
description land in the buffer the caller set aside for application
data. Every in-kernel consumer thus constructs the same recvmsg()
sequence: install a cmsg buffer, discover mid-call that a control
record arrived, take the body from the data buffer, re-issue the
receive with a kvec, and rewind the iterator that the first copy
advanced.

No consumer has gotten that sequence quite right. tls_alert_recv()
read a kvec holding no valid data on the server side (bee47cb026e7)
and again, found independently, on the client side (cc5d59081fa2).
xprtsock processed only the first cmsg and dropped every record type
behind it (9559d2fffd4f). nvmet-tcp and nvme-tcp were built on the
same split and never took the corresponding repair. bee47cb026e7 was
itself a repair, and it left two more defects that stood for a year
(dbabcbc9cf46, b6c603a2415f).

netdev rejected the read_sock_rectype proto op that would have made
that mistake unrepresentable for every consumer. Instead, this series
fixes the svcsock instance alone. Data records and control records
now arrive through separate calls with separate buffers, and no
buffer carries both application data and record metadata. The
MSG_CTRUNC recovery that bee47cb026e7 installed is removed with the
old path. xprtsock, nvmet-tcp, and nvme-tcp are left for future work.

Classifying a record before consuming it needs a receive that reports
the record type without also deciding what that type means for the
transport. That split sits in front of the conversion (patch 1).

Two types of receive that used to keep a connection running now close
it:

- A handshake record during an established TLS session was drained
  and the receive retried, and svcsock has no handler for a
  post-handshake handshake record.

- ->read_sock stops at an urgent octet and consumes none of it.
  recvmsg() walked past that octet and cleared the condition. So,
  when ingested via read_sock, an RPC stream that carries urgent
  data can no longer advance, and the connection must be dropped.

Signed-off-by: Chuck Lever <cel@kernel.org>

---
Chuck Lever (5):
      SUNRPC: Separate the TLS control-record receive from its policy
      SUNRPC: Close the transport on an unhandled TLS record type
      SUNRPC: Flush a received record's pages once it is complete
      SUNRPC: Receive RPC records with ->read_sock
      SUNRPC: Bypass sock_recvmsg() for the TLS control-record receive

 net/sunrpc/svcsock.c | 555 +++++++++++++++++++++++++++------------------------
 1 file changed, 295 insertions(+), 260 deletions(-)
---
base-commit: 01c2994ccb0197cb44b0db89aacab460110f6347
change-id: 20260821-tls-read-sock-2-28c236db7037

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

[PATCH 1/5] SUNRPC: Separate the TLS control-record receive from its policy

[Chuck Lever]

svc_tcp_sock_recv_cmsg() receives the record at the head of the kTLS
receive queue and decides what its content type means for the
transport. The receive and the decision are one step, so a caller
cannot learn a record's type without also acting on it. A later
caller classifies the head of the queue with MSG_PEEK before it
decides whether to consume the record. It needs the receive without
the decision.

Move the receive into svc_tcp_recv_cmsg(), which takes the recvmsg()
flags and reports the octet count, the record type, and the message
flags. The alert policy stays in svc_tcp_sock_recv_cmsg(), the
helper's only caller in this patch.

The zeroed control buffer replaces the msg_controllen check. An
unfilled buffer reports record type zero, so a positive octet count
with no record type now returns -EBADMSG instead of a count. The
alert parse runs on a second msghdr over the alert buffer, so the
copy and the parse share no iterator state. The rewind that commit
bee47cb026e7 ("sunrpc: fix handling of server side tls alerts")
added by hand is no longer needed.

Signed-off-by: Chuck Lever <cel@kernel.org>
---
 net/sunrpc/svcsock.c | 144 ++++++++++++++++++++++-----------------------------
 1 file changed, 63 insertions(+), 81 deletions(-)

diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index 5a2d52284d75..b402923c40f1 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -271,97 +271,79 @@ svc_tcp_sock_drain_record(struct socket *sock)
 	}
 }
 
-static int
-svc_tcp_sock_process_cmsg(struct socket *sock, struct msghdr *msg,
-			  struct cmsghdr *cmsg, int ret)
-{
-	u8 content_type = tls_get_record_type(sock->sk, cmsg);
-	u8 level, description;
-
-	switch (content_type) {
-	case 0:
-		break;
-	case TLS_RECORD_TYPE_DATA:
-		/* TLS sets EOR at the end of each application data
-		 * record, even though there might be more frames
-		 * waiting to be decrypted.
-		 */
-		msg->msg_flags &= ~MSG_EOR;
-		break;
-	case TLS_RECORD_TYPE_ALERT:
-		tls_alert_recv(sock->sk, msg, &level, &description);
-		/* RFC 8446 Section 6: every alert but a closure alert is
-		 * an error alert.
-		 */
-		switch (description) {
-		case TLS_ALERT_DESC_CLOSE_NOTIFY:
-		case TLS_ALERT_DESC_USER_CANCELED:
-			ret = -EAGAIN;
-			break;
-		default:
-			ret = -ENOTCONN;
-		}
-		break;
-	default:
-		/* discard this record type */
-		ret = -EAGAIN;
-	}
-	return ret;
-}
-
-static int
-svc_tcp_sock_recv_cmsg(struct socket *sock, unsigned int *msg_flags)
+static int svc_tcp_recv_cmsg(struct socket *sock, int flags,
+			     struct kvec *payload, u8 *type,
+			     unsigned int *msg_flags)
 {
 	union {
 		struct cmsghdr	cmsg;
 		u8		buf[CMSG_SPACE(sizeof(u8))];
-	} u;
-	u8 alert[2];
-	struct kvec alert_kvec = {
-		.iov_base = alert,
-		.iov_len = sizeof(alert),
-	};
+	} u = {};
 	struct msghdr msg = {
-		.msg_flags = *msg_flags,
-		.msg_control = &u,
-		.msg_controllen = sizeof(u),
+		.msg_control	= &u,
+		.msg_controllen	= sizeof(u),
 	};
 	int ret;
 
-	iov_iter_kvec(&msg.msg_iter, ITER_DEST, &alert_kvec, 1,
-		      alert_kvec.iov_len);
-	ret = sock_recvmsg(sock, &msg, MSG_DONTWAIT);
-	/* put_cmsg() shrinks msg_controllen, so a short one means
-	 * kTLS filled in u.cmsg.
-	 */
-	if (ret >= 0 && msg.msg_controllen < sizeof(u)) {
-		u8 content_type = tls_get_record_type(sock->sk, &u.cmsg);
-
-		/* Returning the count would credit the RPC stream with
-		 * octets that never reached the caller's buffer.
-		 */
-		if (content_type != TLS_RECORD_TYPE_ALERT) {
-			/* An application data record carries RPC payload.
-			 * Draining one breaks RPC fragment framing.
-			 */
-			if (content_type != TLS_RECORD_TYPE_DATA &&
-			    !(msg.msg_flags & MSG_EOR))
-				svc_tcp_sock_drain_record(sock);
-			return -EAGAIN;
-		}
-		/* An Alert record carries exactly one two-octet message
-		 * (RFC 8446 Section 5.1). alert_kvec caps the receive at two,
-		 * so a longer record produces the same count. MSG_EOR appears
-		 * only once kTLS has drained the whole record.
-		 */
-		if (ret != sizeof(alert) || !(msg.msg_flags & MSG_EOR))
-			return -EBADMSG;
-		iov_iter_revert(&msg.msg_iter, ret);
-		ret = svc_tcp_sock_process_cmsg(sock, &msg, &u.cmsg, -EAGAIN);
-	}
+	iov_iter_kvec(&msg.msg_iter, ITER_DEST, payload, 1, payload->iov_len);
+	ret = sock_recvmsg(sock, &msg, flags);
+	if (ret < 0)
+		return ret;
+	*msg_flags = msg.msg_flags;
+	*type = tls_get_record_type(sock->sk, &u.cmsg);
+	if (!*type && ret)
+		return -EBADMSG;
 	return ret;
 }
 
+static int
+svc_tcp_sock_recv_cmsg(struct socket *sock)
+{
+	u8 alert[2], type, level, description;
+	struct kvec recv_kvec = {
+		.iov_base	= alert,
+		.iov_len	= sizeof(alert),
+	};
+	unsigned int msg_flags;
+	struct msghdr msg = {};
+	int ret;
+
+	ret = svc_tcp_recv_cmsg(sock, MSG_DONTWAIT, &recv_kvec, &type,
+				&msg_flags);
+	if (ret < 0 || !type)
+		return ret;
+	if (type != TLS_RECORD_TYPE_ALERT) {
+		/* An application data record carries RPC payload.
+		 * Draining one breaks RPC fragment framing.
+		 */
+		if (type != TLS_RECORD_TYPE_DATA && !(msg_flags & MSG_EOR))
+			svc_tcp_sock_drain_record(sock);
+		return -EAGAIN;
+	}
+	/* An Alert record carries exactly one two-octet message (RFC
+	 * 8446 Section 5.1). recv_kvec caps the receive at two, so a
+	 * longer record produces the same count. MSG_EOR appears only
+	 * once kTLS has drained the whole record.
+	 */
+	if (ret != sizeof(alert) || !(msg_flags & MSG_EOR))
+		return -EBADMSG;
+
+	iov_iter_kvec(&msg.msg_iter, ITER_DEST, &recv_kvec, 1,
+		      recv_kvec.iov_len);
+	tls_alert_recv(sock->sk, &msg, &level, &description);
+
+	/* RFC 8446 Section 6: every alert but a closure alert is
+	 * an error alert.
+	 */
+	switch (description) {
+	case TLS_ALERT_DESC_CLOSE_NOTIFY:
+	case TLS_ALERT_DESC_USER_CANCELED:
+		return -EAGAIN;
+	default:
+		return -ENOTCONN;
+	}
+}
+
 static int
 svc_tcp_sock_recvmsg(struct svc_sock *svsk, struct msghdr *msg)
 {
@@ -372,7 +354,7 @@ svc_tcp_sock_recvmsg(struct svc_sock *svsk, struct msghdr *msg)
 	if (msg->msg_flags & MSG_CTRUNC) {
 		msg->msg_flags &= ~(MSG_CTRUNC | MSG_EOR);
 		if (ret == 0 || ret == -EIO) {
-			ret = svc_tcp_sock_recv_cmsg(sock, &msg->msg_flags);
+			ret = svc_tcp_sock_recv_cmsg(sock);
 			/* A control record delivers nothing to the caller,
 			 * and kTLS announces no data_ready for records it
 			 * already holds. Mark the transport ready so that

-- 
2.54.0

[PATCH 2/5] SUNRPC: Close the transport on an unhandled TLS record type

[Chuck Lever]

svc_tcp_sock_recv_cmsg() drains a TLS record that is neither an alert
nor application data, then returns -EAGAIN so the receive loop retries.
It runs only on an established TLS session. A handshake record there
carries a post-handshake message, and the server has no handler for
one. Draining a KeyUpdate only delays the close. kTLS sets
key_update_pending when it decrypts that record, and a later receive
returns -EKEYEXPIRED. kTLS flags only a KeyUpdate, so any other
post-handshake message disappears and the connection keeps running.

Return -EPROTO for an unhandled record type. Any error but -EAGAIN
closes the transport. An application data record keeps its -EAGAIN
return. No NFS client is known to send a handshake record on an
established connection.

Signed-off-by: Chuck Lever <cel@kernel.org>
---
 net/sunrpc/svcsock.c | 47 +++++++----------------------------------------
 1 file changed, 7 insertions(+), 40 deletions(-)

diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index b402923c40f1..ae1f3c474f8b 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -238,39 +238,6 @@ static int svc_one_sock_name(struct svc_sock *svsk, char *buf, int remaining)
 	return len;
 }
 
-/*
- * kTLS delivers a record only up to the caller's buffer and keeps
- * the remainder on its receive list, where no further data_ready
- * announces it. Consume the whole record.
- */
-static void
-svc_tcp_sock_drain_record(struct socket *sock)
-{
-	union {
-		struct cmsghdr	cmsg;
-		u8		buf[CMSG_SPACE(sizeof(u8))];
-	} u;
-	u8 discard[64];
-	struct kvec discard_kvec = {
-		.iov_base = discard,
-		.iov_len = sizeof(discard),
-	};
-
-	for (;;) {
-		struct msghdr msg = {
-			.msg_control = &u,
-			.msg_controllen = sizeof(u),
-		};
-
-		iov_iter_kvec(&msg.msg_iter, ITER_DEST, &discard_kvec, 1,
-			      discard_kvec.iov_len);
-		if (sock_recvmsg(sock, &msg, MSG_DONTWAIT) <= 0)
-			break;
-		if (msg.msg_flags & MSG_EOR)
-			break;
-	}
-}
-
 static int svc_tcp_recv_cmsg(struct socket *sock, int flags,
 			     struct kvec *payload, u8 *type,
 			     unsigned int *msg_flags)
@@ -312,14 +279,14 @@ svc_tcp_sock_recv_cmsg(struct socket *sock)
 				&msg_flags);
 	if (ret < 0 || !type)
 		return ret;
-	if (type != TLS_RECORD_TYPE_ALERT) {
-		/* An application data record carries RPC payload.
-		 * Draining one breaks RPC fragment framing.
-		 */
-		if (type != TLS_RECORD_TYPE_DATA && !(msg_flags & MSG_EOR))
-			svc_tcp_sock_drain_record(sock);
+	/* A data record reaches here only when kTLS queued an empty one
+	 * ahead of the control record. Consuming it takes no payload,
+	 * and the retry picks up the control record.
+	 */
+	if (type == TLS_RECORD_TYPE_DATA)
 		return -EAGAIN;
-	}
+	if (type != TLS_RECORD_TYPE_ALERT)
+		return -EPROTO;
 	/* An Alert record carries exactly one two-octet message (RFC
 	 * 8446 Section 5.1). recv_kvec caps the receive at two, so a
 	 * longer record produces the same count. MSG_EOR appears only

-- 
2.54.0

[PATCH 3/5] SUNRPC: Flush a received record's pages once it is complete

[Chuck Lever]

svc_tcp_read_msg() flushes the destination pages after every
receive. A record that arrives in several pieces is therefore
flushed once per piece. A page spanning two pieces is flushed
twice. Nothing reads the message body before the record is
complete, so no reader needs the intermediate flushes.

Flush every page of the record in one pass from svc_tcp_recvfrom(),
once the last fragment has arrived. Remove svc_flush_bvec() and its
ARCH_IMPLEMENTS_FLUSH_DCACHE_PAGE guard. The guard avoided setting
up a bvec iterator, and a plain walk of rq_pages compiles away on
its own where flush_dcache_page() is an empty inline.

Signed-off-by: Chuck Lever <cel@kernel.org>
---
 net/sunrpc/svcsock.c | 36 +++++++++++++++---------------------
 1 file changed, 15 insertions(+), 21 deletions(-)

diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index ae1f3c474f8b..c7b94bd1898b 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -334,25 +334,6 @@ svc_tcp_sock_recvmsg(struct svc_sock *svsk, struct msghdr *msg)
 	return ret;
 }
 
-#if ARCH_IMPLEMENTS_FLUSH_DCACHE_PAGE
-static void svc_flush_bvec(const struct bio_vec *bvec, size_t size, size_t seek)
-{
-	struct bvec_iter bi = {
-		.bi_size	= size + seek,
-	};
-	struct bio_vec bv;
-
-	bvec_iter_advance(bvec, &bi, seek & PAGE_MASK);
-	for_each_bvec(bv, bvec, bi, bi)
-		flush_dcache_page(bv.bv_page);
-}
-#else
-static inline void svc_flush_bvec(const struct bio_vec *bvec, size_t size,
-				  size_t seek)
-{
-}
-#endif
-
 /*
  * Read from @rqstp's transport socket. The incoming message fills whole
  * pages in @rqstp's rq_pages array until the last page of the message
@@ -380,8 +361,6 @@ static ssize_t svc_tcp_read_msg(struct svc_rqst *rqstp, size_t buflen,
 		buflen -= seek;
 	}
 	len = svc_tcp_sock_recvmsg(svsk, &msg);
-	if (len > 0)
-		svc_flush_bvec(bvec, len, seek);
 
 	/* If we read a full record, then assume there may be more
 	 * data to read (stream based sockets only!)
@@ -1156,6 +1135,19 @@ static void svc_tcp_fragment_received(struct svc_sock *svsk)
 	svsk->sk_marker = xdr_zero;
 }
 
+/*
+ * Nothing reads the message body before the record is complete, so
+ * a single flush after the last fragment is enough.
+ */
+static void svc_tcp_flush_pages(struct svc_sock *svsk,
+				struct svc_rqst *rqstp)
+{
+	unsigned int pg, pages = DIV_ROUND_UP(svsk->sk_datalen, PAGE_SIZE);
+
+	for (pg = 0; pg < pages; pg++)
+		flush_dcache_page(rqstp->rq_pages[pg]);
+}
+
 /**
  * svc_tcp_recvfrom - Receive data from a TCP socket
  * @rqstp: request structure into which to receive an RPC Call
@@ -1202,6 +1194,8 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 	if (svsk->sk_datalen < 8)
 		goto err_nuts;
 
+	svc_tcp_flush_pages(svsk, rqstp);
+
 	rqstp->rq_arg.len = svsk->sk_datalen;
 	rqstp->rq_arg.page_base = 0;
 	if (rqstp->rq_arg.len <= rqstp->rq_arg.head[0].iov_len) {

-- 
2.54.0

[PATCH 4/5] SUNRPC: Receive RPC records with ->read_sock

[Chuck Lever]

svc_tcp_recvfrom() reads an RPC record with two recvmsg() calls, one
for the four-octet fragment marker and one for the body. Neither
supplies a control-message buffer, so kTLS reports a TLS control
record on either one by raising MSG_CTRUNC and delivering nothing.
Both receives have to recognize that outcome and hand off to a
recovery path that re-reads the record. Commit bee47cb026e7 ("sunrpc:
fix handling of server side tls alerts") built that recovery path. It
stays on both receives for as long as one recvmsg() serves data and
control alike.

Instead, parse the record stream in a ->read_sock actor. The data
path then carries no control-message buffer at all. A control record
at the head stops ->read_sock short, so svc_tcp_recv_ctrl_record()
classifies the head of the receive queue with MSG_PEEK before
consuming anything.

skb_copy_bits() walks an skb from its head to reach the copy offset,
so an actor that copies a page at a time rewalks a large GRO or kTLS
skb once per page. Copy each callback's share of the record into
rq_bvec with one skb_copy_datagram_iter() instead.

A non-final fragment carries four octets of marker and may carry no
payload. sk_datalen advances only by the payload, so neither a run
of empty fragments nor a run of one-octet fragments trips the
sv_max_mesg check before desc->count runs out. Only desc->count
bounds such a run, four or five octets at a time. Cap the fragments
per socket-lock hold.

recvmsg() skips an urgent octet and clears the condition, so the old
receive path advanced past it. ->read_sock consumes what precedes
the urgent octet and then stops there for good. The data path calls
no recvmsg() now, so the record stream cannot advance again. An RPC
stream carries no urgent data, so close the connection. The read
that first reaches the mark returns a positive count, so a
zero-length read does not mark the stop. Key the close on an
incomplete record.

Signed-off-by: Chuck Lever <cel@kernel.org>
---
 net/sunrpc/svcsock.c | 390 +++++++++++++++++++++++++++++++--------------------
 1 file changed, 238 insertions(+), 152 deletions(-)

diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index c7b94bd1898b..fe307d8314c4 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -8,15 +8,6 @@
  * evenly when servicing a single client. May need to modify the
  * svc_xprt_enqueue procedure...
  *
- * TCP support is largely untested and may be a little slow. The problem
- * is that we currently do two separate recvfrom's, one for the 4-byte
- * record length, and the second for the actual record. This could possibly
- * be improved by always reading a minimum size of around 100 bytes and
- * tucking any superfluous bytes away in a temporary store. Still, that
- * leaves write requests out in the rain. An alternative may be to peek at
- * the first skb in the queue, and if it matches the next TCP sequence
- * number, to extract the record marker. Yuck.
- *
  * Copyright (C) 1995, 1996 Olaf Kirch <okir@monad.swb.de>
  */
 
@@ -263,10 +254,10 @@ static int svc_tcp_recv_cmsg(struct socket *sock, int flags,
 	return ret;
 }
 
-static int
-svc_tcp_sock_recv_cmsg(struct socket *sock)
+static int svc_tcp_recv_ctrl_record(struct svc_sock *svsk)
 {
 	u8 alert[2], type, level, description;
+	struct socket *sock = svsk->sk_sock;
 	struct kvec recv_kvec = {
 		.iov_base	= alert,
 		.iov_len	= sizeof(alert),
@@ -275,18 +266,41 @@ svc_tcp_sock_recv_cmsg(struct socket *sock)
 	struct msghdr msg = {};
 	int ret;
 
-	ret = svc_tcp_recv_cmsg(sock, MSG_DONTWAIT, &recv_kvec, &type,
-				&msg_flags);
-	if (ret < 0 || !type)
-		return ret;
-	/* A data record reaches here only when kTLS queued an empty one
-	 * ahead of the control record. Consuming it takes no payload,
-	 * and the retry picks up the control record.
+	if (!test_bit(XPT_TLS_SESSION, &svsk->sk_xprt.xpt_flags))
+		return 0;
+
+	/* A data record can become ready between ->read_sock returning
+	 * and this probe. A plain receive would take two octets of it
+	 * as RPC payload, so peek.
 	 */
-	if (type == TLS_RECORD_TYPE_DATA)
-		return -EAGAIN;
+	ret = svc_tcp_recv_cmsg(sock, MSG_DONTWAIT | MSG_PEEK,
+				&recv_kvec, &type, &msg_flags);
+	if (ret == -EAGAIN || (!ret && !type))
+		return 0;
+	if (ret < 0)
+		return ret;
+	if (type == TLS_RECORD_TYPE_DATA) {
+		/* The peek parks the decrypted record on ctx->rx_list,
+		 * where it draws no further data_ready. Re-arm or the
+		 * RPC hangs until the client times out.
+		 */
+		set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
+		return 0;
+	}
 	if (type != TLS_RECORD_TYPE_ALERT)
 		return -EPROTO;
+
+	ret = svc_tcp_recv_cmsg(sock, MSG_DONTWAIT, &recv_kvec, &type,
+				&msg_flags);
+	/* The peek found a record at the head, so an -EAGAIN here is
+	 * spurious. Propagating it strands the record with no later
+	 * announcement, so return -EBADMSG, which closes the transport.
+	 */
+	if (ret == -EAGAIN)
+		return -EBADMSG;
+	if (ret < 0)
+		return ret;
+
 	/* An Alert record carries exactly one two-octet message (RFC
 	 * 8446 Section 5.1). recv_kvec caps the receive at two, so a
 	 * longer record produces the same count. MSG_EOR appears only
@@ -300,77 +314,19 @@ svc_tcp_sock_recv_cmsg(struct socket *sock)
 	tls_alert_recv(sock->sk, &msg, &level, &description);
 
 	/* RFC 8446 Section 6: every alert but a closure alert is
-	 * an error alert.
+	 * an error alert. kTLS raises no data_ready for records it
+	 * already holds, so re-arm for what sits behind the alert.
 	 */
 	switch (description) {
 	case TLS_ALERT_DESC_CLOSE_NOTIFY:
 	case TLS_ALERT_DESC_USER_CANCELED:
+		set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
 		return -EAGAIN;
 	default:
 		return -ENOTCONN;
 	}
 }
 
-static int
-svc_tcp_sock_recvmsg(struct svc_sock *svsk, struct msghdr *msg)
-{
-	int ret;
-	struct socket *sock = svsk->sk_sock;
-
-	ret = sock_recvmsg(sock, msg, MSG_DONTWAIT);
-	if (msg->msg_flags & MSG_CTRUNC) {
-		msg->msg_flags &= ~(MSG_CTRUNC | MSG_EOR);
-		if (ret == 0 || ret == -EIO) {
-			ret = svc_tcp_sock_recv_cmsg(sock);
-			/* A control record delivers nothing to the caller,
-			 * and kTLS announces no data_ready for records it
-			 * already holds. Mark the transport ready so that
-			 * the records behind this one can be received.
-			 */
-			if (ret == -EAGAIN)
-				set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
-		}
-	}
-	return ret;
-}
-
-/*
- * Read from @rqstp's transport socket. The incoming message fills whole
- * pages in @rqstp's rq_pages array until the last page of the message
- * has been received into a partial page.
- */
-static ssize_t svc_tcp_read_msg(struct svc_rqst *rqstp, size_t buflen,
-				size_t seek)
-{
-	struct svc_sock *svsk =
-		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
-	struct bio_vec *bvec = rqstp->rq_bvec;
-	struct msghdr msg = { NULL };
-	unsigned int i;
-	ssize_t len;
-	size_t t;
-
-	clear_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
-
-	for (i = 0, t = 0; t < buflen; i++, t += PAGE_SIZE)
-		bvec_set_page(&bvec[i], rqstp->rq_pages[i], PAGE_SIZE, 0);
-
-	iov_iter_bvec(&msg.msg_iter, ITER_DEST, bvec, i, buflen);
-	if (seek) {
-		iov_iter_advance(&msg.msg_iter, seek);
-		buflen -= seek;
-	}
-	len = svc_tcp_sock_recvmsg(svsk, &msg);
-
-	/* If we read a full record, then assume there may be more
-	 * data to read (stream based sockets only!)
-	 */
-	if (len == buflen)
-		set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
-
-	return len;
-}
-
 /*
  * Set socket snd and rcv buffer lengths
  */
@@ -993,14 +949,14 @@ static struct svc_xprt *svc_tcp_accept(struct svc_xprt *xprt)
 	return NULL;
 }
 
-static size_t svc_tcp_restore_pages(struct svc_sock *svsk,
-				    struct svc_rqst *rqstp)
+static void svc_tcp_restore_pages(struct svc_sock *svsk,
+				  struct svc_rqst *rqstp)
 {
 	size_t len = svsk->sk_datalen;
 	unsigned int i, npages;
 
 	if (!len)
-		return 0;
+		return;
 	npages = (len + PAGE_SIZE - 1) >> PAGE_SHIFT;
 	for (i = 0; i < npages; i++) {
 		if (rqstp->rq_pages[i] != NULL)
@@ -1010,7 +966,6 @@ static size_t svc_tcp_restore_pages(struct svc_sock *svsk,
 		svsk->sk_pages[i] = NULL;
 	}
 	rqstp->rq_arg.head[0].iov_base = page_address(rqstp->rq_pages[0]);
-	return len;
 }
 
 static void svc_tcp_save_pages(struct svc_sock *svsk, struct svc_rqst *rqstp)
@@ -1049,50 +1004,6 @@ static void svc_tcp_clear_pages(struct svc_sock *svsk)
 	svsk->sk_datalen = 0;
 }
 
-/*
- * Receive fragment record header into sk_marker.
- */
-static ssize_t svc_tcp_read_marker(struct svc_sock *svsk,
-				   struct svc_rqst *rqstp)
-{
-	ssize_t want, len;
-
-	/* If we haven't gotten the record length yet,
-	 * get the next four bytes.
-	 */
-	if (svsk->sk_tcplen < sizeof(rpc_fraghdr)) {
-		struct msghdr	msg = { NULL };
-		struct kvec	iov;
-
-		want = sizeof(rpc_fraghdr) - svsk->sk_tcplen;
-		iov.iov_base = ((char *)&svsk->sk_marker) + svsk->sk_tcplen;
-		iov.iov_len  = want;
-		iov_iter_kvec(&msg.msg_iter, ITER_DEST, &iov, 1, want);
-		len = svc_tcp_sock_recvmsg(svsk, &msg);
-		if (len < 0)
-			return len;
-		svsk->sk_tcplen += len;
-		if (len < want) {
-			/* call again to read the remaining bytes */
-			goto err_short;
-		}
-		trace_svcsock_marker(&svsk->sk_xprt, svsk->sk_marker);
-		if (svc_sock_reclen(svsk) + svsk->sk_datalen >
-		    svsk->sk_xprt.xpt_server->sv_max_mesg)
-			goto err_too_large;
-	}
-	return svc_sock_reclen(svsk);
-
-err_too_large:
-	net_notice_ratelimited("svc: %s oversized RPC fragment (%u octets) from %pISpc\n",
-			       svsk->sk_xprt.xpt_server->sv_name,
-			       svc_sock_reclen(svsk),
-			       (struct sockaddr *)&svsk->sk_xprt.xpt_remote);
-	svc_xprt_deferred_close(&svsk->sk_xprt);
-err_short:
-	return -EAGAIN;
-}
-
 static int receive_cb_reply(struct svc_sock *svsk, struct svc_rqst *rqstp)
 {
 	struct rpc_xprt *bc_xprt = svsk->sk_xprt.xpt_bc_xprt;
@@ -1130,14 +1041,31 @@ static int receive_cb_reply(struct svc_sock *svsk, struct svc_rqst *rqstp)
 
 static void svc_tcp_fragment_received(struct svc_sock *svsk)
 {
-	/* If we have more data, signal svc_xprt_enqueue() to try again */
 	svsk->sk_tcplen = 0;
 	svsk->sk_marker = xdr_zero;
 }
 
 /*
- * Nothing reads the message body before the record is complete, so
- * a single flush after the last fragment is enough.
+ * A non-final fragment carries four octets of marker and may carry
+ * no payload at all. sk_datalen advances only by the payload, so a
+ * run of tiny fragments exhausts ->read_sock's byte budget before
+ * the sv_max_mesg check trips, and a run of empty ones never trips
+ * it. Cap the fragments per socket-lock hold. The cap leaves the
+ * record incomplete, and svc_tcp_recvfrom() resumes it on the next
+ * call.
+ */
+#define SVC_TCP_MAX_FRAGS		256
+
+struct svc_tcp_recv_ctx {
+	struct svc_rqst		*rqstp;
+	unsigned int		frags;
+	bool			complete;
+};
+
+/*
+ * Nothing reads the message body before the message is complete, and
+ * partial receives refill the same pages. Flush once here, after the
+ * socket lock is released, rather than once per copy in the actor.
  */
 static void svc_tcp_flush_pages(struct svc_sock *svsk,
 				struct svc_rqst *rqstp)
@@ -1148,15 +1076,127 @@ static void svc_tcp_flush_pages(struct svc_sock *svsk,
 		flush_dcache_page(rqstp->rq_pages[pg]);
 }
 
+/*
+ * Mapping the message's unfilled remainder would re-map untouched
+ * pages on every call, at a cost that grows with the message rather
+ * than with the octets copied.
+ */
+static void svc_tcp_recv_iter_init(struct svc_rqst *rqstp,
+				   struct iov_iter *iter, size_t body_off,
+				   size_t len)
+{
+	unsigned int first = body_off >> PAGE_SHIFT;
+	size_t seek = offset_in_page(body_off);
+	unsigned int i, pages = DIV_ROUND_UP(seek + len, PAGE_SIZE);
+
+	for (i = 0; i < pages; i++)
+		bvec_set_page(&rqstp->rq_bvec[i], rqstp->rq_pages[first + i],
+			      PAGE_SIZE, 0);
+
+	iov_iter_bvec(iter, ITER_DEST, rqstp->rq_bvec, pages, seek + len);
+	iov_iter_advance(iter, seek);
+}
+
+/*
+ * ->read_sock actor, called under the socket lock. sk_datalen is both
+ * the count of body octets received so far and their write offset into
+ * rq_pages.
+ */
+static int svc_tcp_recv_actor(read_descriptor_t *desc, struct sk_buff *skb,
+			      unsigned int offset, size_t len)
+{
+	struct svc_tcp_recv_ctx *ctx = desc->arg.data;
+	struct svc_rqst *rqstp = ctx->rqstp;
+	struct svc_sock *svsk =
+		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
+	size_t reclen, received, want, take, n;
+	size_t consumed = 0;
+
+	if (!desc->count)
+		return 0;
+
+	len = min(len, desc->count);
+
+	if (svsk->sk_tcplen < sizeof(rpc_fraghdr)) {
+		want = sizeof(rpc_fraghdr) - svsk->sk_tcplen;
+		n = min(want, len);
+
+		if (skb_copy_bits(skb, offset,
+				  (char *)&svsk->sk_marker + svsk->sk_tcplen,
+				  n))
+			goto fault;
+		svsk->sk_tcplen += n;
+		offset += n;
+		len -= n;
+		consumed += n;
+		desc->count -= n;
+
+		if (svsk->sk_tcplen < sizeof(rpc_fraghdr))
+			return consumed;
+
+		trace_svcsock_marker(&svsk->sk_xprt, svsk->sk_marker);
+		if (svc_sock_reclen(svsk) + svsk->sk_datalen >
+		    svsk->sk_xprt.xpt_server->sv_max_mesg) {
+			net_notice_ratelimited("svc: %s oversized RPC fragment (%u octets) from %pISpc\n",
+					       svsk->sk_xprt.xpt_server->sv_name,
+					       svc_sock_reclen(svsk),
+					       (struct sockaddr *)&svsk->sk_xprt.xpt_remote);
+			desc->error = -EMSGSIZE;
+			desc->count = 0;
+			return consumed;
+		}
+	}
+
+	reclen = svc_sock_reclen(svsk);
+	received = svsk->sk_tcplen - sizeof(rpc_fraghdr);
+	want = reclen - received;
+	take = min(want, len);
+
+	if (take) {
+		struct iov_iter iter;
+
+		svc_tcp_recv_iter_init(rqstp, &iter, svsk->sk_datalen, take);
+		if (skb_copy_datagram_iter(skb, offset, &iter, take))
+			goto fault;
+		svsk->sk_datalen += take;
+		svsk->sk_tcplen += take;
+		consumed += take;
+		desc->count -= take;
+	}
+
+	if (take == want) {
+		if (svc_sock_final_rec(svsk)) {
+			ctx->complete = true;
+			desc->count = 0;
+		} else {
+			svc_tcp_fragment_received(svsk);
+			if (++ctx->frags >= SVC_TCP_MAX_FRAGS)
+				desc->count = 0;
+		}
+	}
+
+	return consumed;
+
+fault:
+	desc->error = -EFAULT;
+	desc->count = 0;
+	return consumed;
+}
+
+static bool svc_tcp_at_urg_mark(struct sock *sk)
+{
+	const struct tcp_sock *tp = tcp_sk(sk);
+
+	return tp->urg_data && tp->urg_seq == tp->copied_seq;
+}
+
 /**
  * svc_tcp_recvfrom - Receive data from a TCP socket
  * @rqstp: request structure into which to receive an RPC Call
  *
  * Called in a loop when XPT_DATA has been set.
  *
- * Read the 4-byte stream record marker, then use the record length
- * in that marker to set up exactly the resources needed to receive
- * the next RPC message into @rqstp.
+ * Context: Process context. Takes and releases the socket lock.
  *
  * Returns:
  *   On success, the number of bytes in a received RPC Call, or
@@ -1171,26 +1211,64 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 	struct svc_sock	*svsk =
 		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
 	struct svc_serv	*serv = svsk->sk_xprt.xpt_server;
-	size_t want, base;
+	struct svc_tcp_recv_ctx ctx = {
+		.rqstp		= rqstp,
+	};
+	read_descriptor_t desc = {
+		.arg.data	= &ctx,
+		.count		= serv->sv_max_mesg + sizeof(rpc_fraghdr),
+	};
+	struct socket *sock = svsk->sk_sock;
 	ssize_t len;
 	__be32 *p;
 	__be32 calldir;
 
 	clear_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
-	len = svc_tcp_read_marker(svsk, rqstp);
-	if (len < 0)
-		goto error;
+	svc_tcp_restore_pages(svsk, rqstp);
 
-	base = svc_tcp_restore_pages(svsk, rqstp);
-	want = len - (svsk->sk_tcplen - sizeof(rpc_fraghdr));
-	len = svc_tcp_read_msg(rqstp, base + want, base);
-	if (len >= 0) {
-		trace_svcsock_tcp_recv(&svsk->sk_xprt, len);
-		svsk->sk_tcplen += len;
-		svsk->sk_datalen += len;
+	lock_sock(sock->sk);
+	len = sock->ops->read_sock(sock->sk, &desc, svc_tcp_recv_actor);
+	/* ->read_sock stops at urgent data and consumes none of it.
+	 * Only recvmsg() clears the condition, and this path calls
+	 * none, so every later read stops at the same octet. An RPC
+	 * stream carries no urgent data, so close the connection.
+	 *
+	 * The read that first reaches the mark consumes the octets
+	 * ahead of it, so the stop does not show up as a zero len. A
+	 * record completed ahead of the mark is returned first. The
+	 * XPT_DATA set below brings the next call back here with
+	 * nothing left to consume.
+	 */
+	if (!ctx.complete && svc_tcp_at_urg_mark(sock->sk))
+		desc.error = -EPROTO;
+	release_sock(sock->sk);
+
+	/* ->read_sock returns the octets consumed before an actor
+	 * failure, so a positive len can accompany desc.error.
+	 */
+	if (desc.error < 0) {
+		len = desc.error;
+		goto err_nuts;
 	}
-	if (len != want || !svc_sock_final_rec(svsk))
+	if (len >= 0)
+		trace_svcsock_tcp_recv(&svsk->sk_xprt, len);
+
+	if (!ctx.complete) {
+		if (!desc.count) {
+			set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
+			goto err_incomplete;
+		}
+		/* A zero return leaves no record at the head to classify.
+		 * -EINVAL means a control record sits there. Screen the
+		 * other errors out first, because a probe calls
+		 * sock_error(), whose xchg clears sk->sk_err as it reads.
+		 */
+		if (len <= 0 && len != -EINVAL)
+			goto err_incomplete;
+
+		len = svc_tcp_recv_ctrl_record(svsk);
 		goto err_incomplete;
+	}
 	if (svsk->sk_datalen < 8)
 		goto err_nuts;
 
@@ -1211,6 +1289,12 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 	else
 		clear_bit(RQ_LOCAL, &rqstp->rq_flags);
 
+	/* Completing one message stops ->read_sock with whatever
+	 * follows still queued, and no path from here re-arms XPT_DATA.
+	 * The queued message would wait for unrelated traffic.
+	 */
+	set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
+
 	p = (__be32 *)rqstp->rq_arg.head[0].iov_base;
 	calldir = p[1];
 	if (calldir)
@@ -1235,19 +1319,21 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 	svc_tcp_save_pages(svsk, rqstp);
 	if (len < 0 && len != -EAGAIN)
 		goto err_delete;
-	if (len == want)
-		svc_tcp_fragment_received(svsk);
-	else
+	if (svsk->sk_tcplen >= sizeof(rpc_fraghdr))
 		trace_svcsock_tcp_recv_short(&svsk->sk_xprt,
 				svc_sock_reclen(svsk),
 				svsk->sk_tcplen - sizeof(rpc_fraghdr));
+	else
+		trace_svcsock_tcp_recv_eagain(&svsk->sk_xprt, 0);
 	goto err_noclose;
 error:
-	if (len != -EAGAIN)
-		goto err_delete;
 	trace_svcsock_tcp_recv_eagain(&svsk->sk_xprt, 0);
 	goto err_noclose;
 err_nuts:
+	/* svc_tcp_save_pages() has not run, so svsk->sk_pages[] is
+	 * empty. A non-zero sk_datalen makes the teardown-time
+	 * svc_tcp_clear_pages() walk empty slots and WARN.
+	 */
 	svsk->sk_datalen = 0;
 err_delete:
 	trace_svcsock_tcp_recv_err(&svsk->sk_xprt, len);

-- 
2.54.0

[PATCH 5/5] SUNRPC: Bypass sock_recvmsg() for the TLS control-record receive

[Chuck Lever]

svc_tcp_recvfrom() parses the RPC record stream with ->read_sock,
which calls neither security_socket_recvmsg() nor the
sock:sock_recv_length tracepoint. svc_tcp_recv_cmsg() still goes
through sock_recvmsg(), so an LSM mediates only the TLS control
records on a server socket, and sock:sock_recv_length reports only
those. Partial coverage is worse than none. It makes the RPC stream
look mediated and observed when it is not.

Until the record stream moved to ->read_sock, an LSM saw every octet
NFSD read from a TCP socket. An SELinux policy that denies
SOCKET__READ to NFSD blocked the receive. After this change no call
on the server's TCP receive path consults an LSM, so that denial has
no effect.

Dispatch ->recvmsg directly so the whole receive path behaves one
way. sock_recvmsg_nosec() reaches ->recvmsg through
INDIRECT_CALL_INET(), so on a retpoline build the direct dispatch
costs one indirect call per control record. Control records are rare
on an established connection.

Signed-off-by: Chuck Lever <cel@kernel.org>
---
 net/sunrpc/svcsock.c | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index fe307d8314c4..ef7ac080fcd3 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -229,10 +229,16 @@ static int svc_one_sock_name(struct svc_sock *svsk, char *buf, int remaining)
 	return len;
 }
 
+/*
+ * The ->read_sock data path invokes neither security_socket_recvmsg()
+ * nor the sock:sock_recv_length tracepoint. Dispatch ->recvmsg
+ * directly so the whole receive path behaves one way.
+ */
 static int svc_tcp_recv_cmsg(struct socket *sock, int flags,
 			     struct kvec *payload, u8 *type,
 			     unsigned int *msg_flags)
 {
+	const struct proto_ops *ops = READ_ONCE(sock->ops);
 	union {
 		struct cmsghdr	cmsg;
 		u8		buf[CMSG_SPACE(sizeof(u8))];
@@ -244,7 +250,7 @@ static int svc_tcp_recv_cmsg(struct socket *sock, int flags,
 	int ret;
 
 	iov_iter_kvec(&msg.msg_iter, ITER_DEST, payload, 1, payload->iov_len);
-	ret = sock_recvmsg(sock, &msg, flags);
+	ret = ops->recvmsg(sock, &msg, msg_data_left(&msg), flags);
 	if (ret < 0)
 		return ret;
 	*msg_flags = msg.msg_flags;

-- 
2.54.0