[PATCH net-next v2 0/6] Deliver TLS control records to kernel read_sock consumers

[PATCH net-next v2 0/6] Deliver TLS control records to kernel read_sock consumers

[Chuck Lever]

When kTLS is active, kernel consumers of the read_sock API cannot
see TLS control records: tls_sw_read_sock() rejects Alerts and
Handshake records with -EINVAL. A consumer that needs those records
falls back to sock_recvmsg() with ancillary buffers, which is why
NFSD's svcsock runs a MSG_CTRUNC recovery dance on every receive.

This series adds a read_sock_rectype proto_ops method that delivers
non-data records through a separate callback, then converts its first
consumer, NFSD's svcsock, to use it. NFS client (xprtsock), NVMe
target, and NVMe host are meant to move onto the same mechanism in
later series. The existing read_sock data path is unchanged.

The improvement is more than cosmetic. All of these consumers need
to handle TLS Alert messages efficiently and securely (in particular,
for KeyUpdate). A few design points worth mentioning:

- The new sk_read_rectype_actor_t callback does not share the data
  actor's byte-counting contract. It returns 0 to consume a record or
  a negative value to requeue it and stop delivery. The return is
  never a byte count. svc_tcp_rectype_actor() relies on this,
  returning 0 to consume every non-data record while stopping the
  read loop through desc->count on a fatal alert.

- The no-data cap (patch 1) is a prerequisite, not a stand-alone fix.
  Once control records reach read_sock, a record carrying no payload
  stops advancing the caller's read descriptor, so a peer streaming
  such records would pin the socket lock and the kernel receive
  context for as long as the flood lasts. Bounding consecutive
  no-data records supplies the return boundary a system call would
  otherwise provide. The cap is scoped to tls_sw_read_sock() alone:
  splice and recvmsg run in the caller's own context, reschedule, and
  drop the lock on return, so they need nothing.

- The svcsock conversion is split so the new path can be reviewed
  against the old. The old path is then removed by the last patch
  in the series.

- read_sock_rectype has no direct userspace entry point, so its
  selftest coverage (patch 4) drives the shared decryption and
  rx_list delivery pipeline through the recvmsg and splice paths
  instead.

---
Changes in v2:
- Renamed the read_sock_cmsg proto_ops method to read_sock_rectype.
- Non-data records now use a record-type actor, not RFC cmsg-style.
- Bound no-data records so they can't pin the socket lock (new patch 1).
- Added selftests for data/control record interleaving (new patch 4).
- Link to v1: https://lore.kernel.org/r/20260217222033.1929211-1-cel@kernel.org

---
Chuck Lever (6):
      net/tls: Bound consecutive no-data records in tls_sw_read_sock()
      net: Introduce read_sock_rectype proto_ops for control record delivery
      tls: Implement read_sock_rectype for kTLS software path
      selftests/tls: Add tests for data/control record interleaving
      SUNRPC: Use read_sock_rectype for svcsock TCP receives
      SUNRPC: Remove sock_recvmsg path from svcsock TCP receives

 include/linux/net.h               |  28 +++
 net/sunrpc/svcsock.c              | 381 +++++++++++++++++---------------------
 net/tls/tls.h                     |   3 +
 net/tls/tls_main.c                |   5 +
 net/tls/tls_sw.c                  |  54 +++++-
 tools/testing/selftests/net/tls.c | 298 ++++++++++++++++++++++++++++-
 6 files changed, 556 insertions(+), 213 deletions(-)
---
base-commit: 298bb2b8903323f6ef2eab4819a2e477765f0ff1
change-id: 20260327-tcp-read-sock-778a49660ff7

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

[PATCH net-next v2 1/6] net/tls: Bound consecutive no-data records in tls_sw_read_sock()

[Chuck Lever]

A record that delivers no payload -- an empty TLS 1.3 data record
today, a control record once read_sock_rectype() lands -- leaves
tls_sw_read_sock() in its loop without advancing the caller's read
descriptor. A peer that streams such records keeps the receive loop
running, and the socket lock held, for as long as the records
arrive.

Cap the number of consecutive no-data records consumed per call. The
count resets on any record that delivers bytes, so a normal stream
is unaffected; a peer supplying only empty records is bounded to
TLS_RX_NODATA_LIMIT iterations before the call returns 0. read_sock
consumers treat that as "no progress, re-poll" rather than EOF, so
the connection stays up and makes progress once real data arrives.

Only tls_sw_read_sock() needs this cap. Its consumers drive the receive
loop from kernel context -- a work item or service thread holding the
socket lock across the whole call with no return to userspace -- so an
unbounded empty-record stream keeps that context and the lock pinned
for as long as the flood lasts. The cap supplies the return boundary
that a system call would otherwise provide. tls_sw_splice_read()
and tls_sw_recvmsg() already have one: they run in the calling task's
context, reschedule while draining the socket backlog (cond_resched()
in __release_sock()), and drop the socket lock when the call returns. A
flood there costs the caller only its own scheduler time, so the cap
would add nothing.

Signed-off-by: Chuck Lever <cel@kernel.org>
---
 net/tls/tls_sw.c | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/net/tls/tls_sw.c b/net/tls/tls_sw.c
index d4afc90fd796..087950ca639c 100644
--- a/net/tls/tls_sw.c
+++ b/net/tls/tls_sw.c
@@ -2049,6 +2049,11 @@ ssize_t tls_sw_splice_read(struct socket *sock,  loff_t *ppos,
 	goto splice_read_end;
 }
 
+/* Consecutive empty data records deliver no bytes; cap them per
+ * call so a peer streaming them cannot hold the socket lock here.
+ */
+#define TLS_RX_NODATA_LIMIT 16
+
 int tls_sw_read_sock(struct sock *sk, read_descriptor_t *desc,
 		     sk_read_actor_t read_actor)
 {
@@ -2057,6 +2062,7 @@ int tls_sw_read_sock(struct sock *sk, read_descriptor_t *desc,
 	struct tls_prot_info *prot = &tls_ctx->prot_info;
 	struct strp_msg *rxm = NULL;
 	struct sk_buff *skb = NULL;
+	unsigned int nodata_count = 0;
 	struct sk_psock *psock;
 	size_t flushed_at = 0;
 	bool released = true;
@@ -2122,7 +2128,13 @@ int tls_sw_read_sock(struct sock *sk, read_descriptor_t *desc,
 		 * here instead.
 		 */
 		if (rxm->full_len == 0) {
+			err = 0;
 			consume_skb(skb);
+			/* tls_rx_reader_release() announces any parsed record
+			 * on exit, so returning 0 here cannot strand it.
+			 */
+			if (++nodata_count >= TLS_RX_NODATA_LIMIT)
+				break;
 			continue;
 		}
 
@@ -2133,6 +2145,7 @@ int tls_sw_read_sock(struct sock *sk, read_descriptor_t *desc,
 			goto read_sock_requeue;
 		}
 		copied += used;
+		nodata_count = 0;
 		if (used < rxm->full_len) {
 			rxm->offset += used;
 			rxm->full_len -= used;

-- 
2.54.0

[PATCH net-next v2 2/6] net: Introduce read_sock_rectype proto_ops for control record delivery

[Chuck Lever]

From: Chuck Lever <chuck.lever@oracle.com>

Kernel TCP consumers that use the read_sock interface
(proto_ops.read_sock) cannot receive TLS control messages (Alerts,
Handshake records) when kTLS is active. The current
tls_sw_read_sock() method rejects non-data records with -EINVAL, and
the sk_read_actor_t callback has no channel for delivering record-
type metadata.

Four kernel subsystems are affected: NFSD (sunrpc svcsock), NFS
client (sunrpc xprtsock), NVMe target (nvmet-tcp), and NVMe host
(nvme-tcp). Each of these either falls back to the sock_recvmsg()
API or lacks TLS alert handling entirely.

A new read_sock_rectype method in struct proto_ops provides a
separate code path that delivers non-data TLS records to a callback,
without changing the behavior seen by existing read_sock consumers.

The new sk_read_rectype_actor_t callback type extends the
sk_read_actor_t signature with a rectype parameter carrying the
protocol-layer record type (for example, TLS_RECORD_TYPE_ALERT). The
record-type callback returns 0 to consume a record or a negative
value to requeue it and stop delivery; unlike the data callback, its
return value does not count bytes.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
 include/linux/net.h | 28 ++++++++++++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/include/linux/net.h b/include/linux/net.h
index 277188a40c72..7a19a743a617 100644
--- a/include/linux/net.h
+++ b/include/linux/net.h
@@ -198,6 +198,13 @@ struct sk_buff;
 struct proto_accept_arg;
 typedef int (*sk_read_actor_t)(read_descriptor_t *, struct sk_buff *,
 			       unsigned int, size_t);
+/* rectype carries the transport record type, for example a
+ * TLS_RECORD_TYPE_* value.
+ */
+typedef int (*sk_read_rectype_actor_t)(read_descriptor_t *,
+				       struct sk_buff *,
+				       unsigned int, size_t,
+				       u8 rectype);
 typedef int (*skb_read_actor_t)(struct sock *, struct sk_buff *);
 
 
@@ -264,6 +271,27 @@ struct proto_ops {
 	 */
 	int		(*read_sock)(struct sock *sk, read_descriptor_t *desc,
 				     sk_read_actor_t recv_actor);
+	/*
+	 * read_sock_rectype splits delivery across two callbacks:
+	 * recv_actor for data records, per the sk_read_actor_t
+	 * convention, and rectype_actor for all other records,
+	 * with rectype identifying each. A NULL rectype_actor
+	 * leaves non-data records pending. rectype_actor returns 0
+	 * to consume a record or negative to leave it pending for
+	 * redelivery and stop delivery; the negative return is a
+	 * backpressure signal, not a fatal error. Both callbacks
+	 * report errors and early stop the way recv_actor does:
+	 * by setting desc->count to 0 and recording the reason in
+	 * desc->error, per the read_descriptor_t convention and
+	 * independent of the return value. The return value reports
+	 * only data bytes consumed by recv_actor; the caller
+	 * detects an error or early stop via desc->count and
+	 * desc->error.
+	 */
+	int		(*read_sock_rectype)(struct sock *sk,
+					     read_descriptor_t *desc,
+					     sk_read_actor_t recv_actor,
+					     sk_read_rectype_actor_t rectype_actor);
 	/* This is different from read_sock(), it reads an entire skb at a time. */
 	int		(*read_skb)(struct sock *sk, skb_read_actor_t recv_actor);
 	int		(*sendmsg_locked)(struct sock *sk, struct msghdr *msg,

-- 
2.54.0

[PATCH net-next v2 3/6] tls: Implement read_sock_rectype for kTLS software path

[Chuck Lever]

From: Chuck Lever <chuck.lever@oracle.com>

tls_sw_read_sock() rejects non-data records (alerts, handshake
messages) with -EINVAL. Kernel consumers that need TLS alert
delivery, such as NFSD, NFS client, and NVMe target, must fall back
to the sock_recvmsg() API to receive control messages via CMSG.

Implement the new read_sock_rectype() method for these consumers,
delivering non-data records to a callback through the kTLS software
receive path.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
 net/tls/tls.h      |  3 +++
 net/tls/tls_main.c |  5 +++++
 net/tls/tls_sw.c   | 41 ++++++++++++++++++++++++++++++++++++-----
 3 files changed, 44 insertions(+), 5 deletions(-)

diff --git a/net/tls/tls.h b/net/tls/tls.h
index 60a37bdaaa25..c21d2a985e13 100644
--- a/net/tls/tls.h
+++ b/net/tls/tls.h
@@ -168,6 +168,9 @@ ssize_t tls_sw_splice_read(struct socket *sock, loff_t *ppos,
 			   size_t len, unsigned int flags);
 int tls_sw_read_sock(struct sock *sk, read_descriptor_t *desc,
 		     sk_read_actor_t read_actor);
+int tls_sw_read_sock_rectype(struct sock *sk, read_descriptor_t *desc,
+			     sk_read_actor_t read_actor,
+			     sk_read_rectype_actor_t rectype_actor);
 
 int tls_device_sendmsg(struct sock *sk, struct msghdr *msg, size_t size);
 void tls_device_splice_eof(struct socket *sock);
diff --git a/net/tls/tls_main.c b/net/tls/tls_main.c
index 8c588cdab733..4963e0caf6d5 100644
--- a/net/tls/tls_main.c
+++ b/net/tls/tls_main.c
@@ -949,12 +949,17 @@ static void build_proto_ops(struct proto_ops ops[TLS_NUM_CONFIG][TLS_NUM_CONFIG]
 	ops[TLS_BASE][TLS_SW  ].splice_read	= tls_sw_splice_read;
 	ops[TLS_BASE][TLS_SW  ].poll		= tls_sk_poll;
 	ops[TLS_BASE][TLS_SW  ].read_sock	= tls_sw_read_sock;
+	ops[TLS_BASE][TLS_SW  ].read_sock_rectype	= tls_sw_read_sock_rectype;
 
 	ops[TLS_SW  ][TLS_SW  ] = ops[TLS_SW  ][TLS_BASE];
 	ops[TLS_SW  ][TLS_SW  ].splice_read	= tls_sw_splice_read;
 	ops[TLS_SW  ][TLS_SW  ].poll		= tls_sk_poll;
 	ops[TLS_SW  ][TLS_SW  ].read_sock	= tls_sw_read_sock;
+	ops[TLS_SW  ][TLS_SW  ].read_sock_rectype	= tls_sw_read_sock_rectype;
 
+	/* TLS_HW (device offload) RX entries inherit
+	 * read_sock{,_rectype} from SW via the struct copies below.
+	 */
 #ifdef CONFIG_TLS_DEVICE
 	ops[TLS_HW  ][TLS_BASE] = ops[TLS_BASE][TLS_BASE];
 
diff --git a/net/tls/tls_sw.c b/net/tls/tls_sw.c
index 087950ca639c..af347b5b17fa 100644
--- a/net/tls/tls_sw.c
+++ b/net/tls/tls_sw.c
@@ -2054,8 +2054,9 @@ ssize_t tls_sw_splice_read(struct socket *sock,  loff_t *ppos,
  */
 #define TLS_RX_NODATA_LIMIT 16
 
-int tls_sw_read_sock(struct sock *sk, read_descriptor_t *desc,
-		     sk_read_actor_t read_actor)
+static int __tls_sw_read_sock(struct sock *sk, read_descriptor_t *desc,
+			      sk_read_actor_t read_actor,
+			      sk_read_rectype_actor_t rectype_actor)
 {
 	struct tls_context *tls_ctx = tls_get_ctx(sk);
 	struct tls_sw_context_rx *ctx = tls_sw_ctx_rx(tls_ctx);
@@ -2115,10 +2116,27 @@ int tls_sw_read_sock(struct sock *sk, read_descriptor_t *desc,
 			tls_rx_rec_done(ctx);
 		}
 
-		/* read_sock does not support reading control messages */
+		/* Control records (alerts, handshake) reach a consumer
+		 * only through rectype_actor; without one, read_sock
+		 * rejects them.
+		 */
 		if (tlm->control != TLS_RECORD_TYPE_DATA) {
-			err = -EINVAL;
-			goto read_sock_requeue;
+			if (!rectype_actor) {
+				err = -EINVAL;
+				goto read_sock_requeue;
+			}
+			err = rectype_actor(desc, skb, rxm->offset,
+					 rxm->full_len,
+					 tlm->control);
+			if (err < 0)
+				goto read_sock_requeue;
+			err = 0;
+			/* rectype_actor consumes the whole record; no partial path */
+			consume_skb(skb);
+			skb = NULL;
+			if (++nodata_count >= TLS_RX_NODATA_LIMIT)
+				break;
+			continue;
 		}
 
 		/* An empty data record (legal in TLS 1.3) gives a zero
@@ -2164,6 +2182,19 @@ int tls_sw_read_sock(struct sock *sk, read_descriptor_t *desc,
 	goto read_sock_end;
 }
 
+int tls_sw_read_sock(struct sock *sk, read_descriptor_t *desc,
+		     sk_read_actor_t read_actor)
+{
+	return __tls_sw_read_sock(sk, desc, read_actor, NULL);
+}
+
+int tls_sw_read_sock_rectype(struct sock *sk, read_descriptor_t *desc,
+			     sk_read_actor_t read_actor,
+			     sk_read_rectype_actor_t rectype_actor)
+{
+	return __tls_sw_read_sock(sk, desc, read_actor, rectype_actor);
+}
+
 bool tls_sw_sock_is_readable(struct sock *sk)
 {
 	struct tls_context *tls_ctx = tls_get_ctx(sk);

-- 
2.54.0

[PATCH net-next v2 4/6] selftests/tls: Add tests for data/control record interleaving

[Chuck Lever]

From: Chuck Lever <chuck.lever@oracle.com>

The new read_sock_rectype proto_ops method delivers interleaved
data and control TLS records to kernel consumers through
separate callbacks. The existing selftest coverage for these
interleaving patterns is limited: data_control_data only
peeks, and splice_cmsg_to_pipe tests a single control record
in isolation.

Add seven tests that exercise the record patterns
read_sock_rectype is designed to handle:

- splice_data_cmsg_data: data-control-data via splice, with
  the control record drained through recvmsg between the two
  splice calls

- splice_multi_cmsg_data: data-control-control-data with
  distinct content types, verifying that each control record
  is independently drainable with its type preserved, and
  splice resumes afterward

- recv_data_cmsg_data: complete consumption of a
  data-control-data sequence through recv and recvmsg

- peek_cmsg_after_data: peek at an interleaved control
  record after the preceding data record has been consumed,
  then consume it

- cmsg_before_data: control record as the first record in the
  stream, followed by data

- mixed_control_types: two different control record types
  (distinct content_type values) interleaved with data,
  verifying type preservation through delivery

- data_cmsg_eof: trailing control record followed by
  connection close, verifying the receiver drains the
  control record and then observes EOF

The fixture teardown is also updated to skip closing fd
when a test has already closed it (as data_cmsg_eof does).

These tests exercise the record decryption and rx_list
delivery pipeline shared by the recvmsg, splice, and
read_sock paths. read_sock_rectype itself is a kernel-internal
API without a direct userspace entry point, so the tests
validate through the userspace-accessible paths.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
 tools/testing/selftests/net/tls.c | 298 +++++++++++++++++++++++++++++++++++++-
 1 file changed, 297 insertions(+), 1 deletion(-)

diff --git a/tools/testing/selftests/net/tls.c b/tools/testing/selftests/net/tls.c
index cbdd3ea28b99..8136306b5caa 100644
--- a/tools/testing/selftests/net/tls.c
+++ b/tools/testing/selftests/net/tls.c
@@ -427,7 +427,8 @@ FIXTURE_SETUP(tls)
 
 FIXTURE_TEARDOWN(tls)
 {
-	close(self->fd);
+	if (self->fd >= 0)
+		close(self->fd);
 	close(self->cfd);
 }
 
@@ -897,6 +898,114 @@ TEST_F(tls, splice_dec_cmsg_to_pipe)
 	EXPECT_EQ(memcmp(test_str, buf, send_len), 0);
 }
 
+/* Verify splice handles data-control-data: splice reads the data
+ * records successfully while the intervening control record must
+ * be drained via recvmsg before splice can continue.
+ */
+TEST_F(tls, splice_data_cmsg_data)
+{
+	char mem_send[TLS_PAYLOAD_MAX_LEN];
+	char mem_recv[TLS_PAYLOAD_MAX_LEN];
+	int send_len = 4096;
+	char *ctrl_str = "control";
+	int ctrl_len = strlen(ctrl_str) + 1;
+	char ctrl_buf[8];
+	int p[2];
+
+	if (self->notls)
+		SKIP(return, "no TLS support");
+
+	memrnd(mem_send, sizeof(mem_send));
+
+	ASSERT_GE(pipe(p), 0);
+
+	/* Send: data, control, data */
+	EXPECT_EQ(send(self->fd, mem_send, send_len, 0), send_len);
+	EXPECT_EQ(tls_send_cmsg(self->fd, 100, ctrl_str, ctrl_len, 0),
+		  ctrl_len);
+	EXPECT_EQ(send(self->fd, &mem_send[send_len], send_len, 0), send_len);
+
+	/* Splice first data record */
+	EXPECT_EQ(splice(self->cfd, NULL, p[1], NULL, send_len, 0), send_len);
+	EXPECT_EQ(read(p[0], mem_recv, send_len), send_len);
+	EXPECT_EQ(memcmp(mem_send, mem_recv, send_len), 0);
+
+	/* Splice hits control record, fails */
+	EXPECT_EQ(splice(self->cfd, NULL, p[1], NULL, send_len, 0), -1);
+	EXPECT_EQ(errno, EINVAL);
+
+	/* Drain the control record via recvmsg */
+	EXPECT_EQ(tls_recv_cmsg(_metadata, self->cfd, 100,
+				ctrl_buf, sizeof(ctrl_buf), MSG_WAITALL),
+		  ctrl_len);
+	EXPECT_EQ(memcmp(ctrl_str, ctrl_buf, ctrl_len), 0);
+
+	/* Splice second data record */
+	EXPECT_EQ(splice(self->cfd, NULL, p[1], NULL, send_len, 0), send_len);
+	EXPECT_EQ(read(p[0], mem_recv, send_len), send_len);
+	EXPECT_EQ(memcmp(&mem_send[send_len], mem_recv, send_len), 0);
+}
+
+/* Verify that multiple consecutive control records between data
+ * records can each be drained individually, and splice resumes
+ * afterward.  The two control records use different content types
+ * to verify type preservation across the splice boundary.
+ */
+TEST_F(tls, splice_multi_cmsg_data)
+{
+	char mem_send[TLS_PAYLOAD_MAX_LEN];
+	char mem_recv[TLS_PAYLOAD_MAX_LEN];
+	int send_len = 4096;
+	char *ctrl1 = "alert1";
+	char *ctrl2 = "alert2";
+	int ctrl_len = strlen(ctrl1) + 1;
+	char ctrl_buf[7];
+	int p[2];
+
+	if (self->notls)
+		SKIP(return, "no TLS support");
+
+	memrnd(mem_send, sizeof(mem_send));
+
+	ASSERT_GE(pipe(p), 0);
+
+	/* Send: data, control(100), control(200), data */
+	EXPECT_EQ(send(self->fd, mem_send, send_len, 0), send_len);
+	EXPECT_EQ(tls_send_cmsg(self->fd, 100, ctrl1, ctrl_len, 0), ctrl_len);
+	EXPECT_EQ(tls_send_cmsg(self->fd, 200, ctrl2, ctrl_len, 0), ctrl_len);
+	EXPECT_EQ(send(self->fd, &mem_send[send_len], send_len, 0), send_len);
+
+	/* Splice first data */
+	EXPECT_EQ(splice(self->cfd, NULL, p[1], NULL, send_len, 0), send_len);
+	EXPECT_EQ(read(p[0], mem_recv, send_len), send_len);
+	EXPECT_EQ(memcmp(mem_send, mem_recv, send_len), 0);
+
+	/* Splice fails on first control record */
+	EXPECT_EQ(splice(self->cfd, NULL, p[1], NULL, send_len, 0), -1);
+	EXPECT_EQ(errno, EINVAL);
+
+	/* Drain first control (type 100) */
+	EXPECT_EQ(tls_recv_cmsg(_metadata, self->cfd, 100,
+				ctrl_buf, sizeof(ctrl_buf), MSG_WAITALL),
+		  ctrl_len);
+	EXPECT_EQ(memcmp(ctrl1, ctrl_buf, ctrl_len), 0);
+
+	/* Splice fails on second control record */
+	EXPECT_EQ(splice(self->cfd, NULL, p[1], NULL, send_len, 0), -1);
+	EXPECT_EQ(errno, EINVAL);
+
+	/* Drain second control (type 200) */
+	EXPECT_EQ(tls_recv_cmsg(_metadata, self->cfd, 200,
+				ctrl_buf, sizeof(ctrl_buf), MSG_WAITALL),
+		  ctrl_len);
+	EXPECT_EQ(memcmp(ctrl2, ctrl_buf, ctrl_len), 0);
+
+	/* Splice second data */
+	EXPECT_EQ(splice(self->cfd, NULL, p[1], NULL, send_len, 0), send_len);
+	EXPECT_EQ(read(p[0], mem_recv, send_len), send_len);
+	EXPECT_EQ(memcmp(&mem_send[send_len], mem_recv, send_len), 0);
+}
+
 TEST_F(tls, recv_and_splice)
 {
 	int send_len = TLS_PAYLOAD_MAX_LEN;
@@ -1682,6 +1791,193 @@ TEST_F(tls, data_control_data)
 	EXPECT_EQ(recv(self->cfd, buf, sizeof(buf), MSG_PEEK), send_len);
 }
 
+/* Fully consume a data-control-data sequence.  The existing
+ * data_control_data test only peeks; this exercises complete
+ * record delivery through recv and recvmsg.
+ */
+TEST_F(tls, recv_data_cmsg_data)
+{
+	char *data1 = "first_data";
+	char *ctrl = "ctrl_msg";
+	char *data2 = "second_data";
+	int d1_len = strlen(data1) + 1;
+	int c_len = strlen(ctrl) + 1;
+	int d2_len = strlen(data2) + 1;
+	char buf[20];
+
+	if (self->notls)
+		SKIP(return, "no TLS support");
+
+	EXPECT_EQ(send(self->fd, data1, d1_len, 0), d1_len);
+	EXPECT_EQ(tls_send_cmsg(self->fd, 100, ctrl, c_len, 0), c_len);
+	EXPECT_EQ(send(self->fd, data2, d2_len, 0), d2_len);
+
+	/* First data record */
+	EXPECT_EQ(recv(self->cfd, buf, sizeof(buf), MSG_WAITALL), d1_len);
+	EXPECT_EQ(memcmp(buf, data1, d1_len), 0);
+
+	/* recv without cmsg buffer fails on control record */
+	EXPECT_EQ(recv(self->cfd, buf, sizeof(buf), 0), -1);
+	EXPECT_EQ(errno, EIO);
+
+	/* Drain control via recvmsg */
+	EXPECT_EQ(tls_recv_cmsg(_metadata, self->cfd, 100,
+				buf, sizeof(buf), MSG_WAITALL), c_len);
+	EXPECT_EQ(memcmp(buf, ctrl, c_len), 0);
+
+	/* Second data record */
+	EXPECT_EQ(recv(self->cfd, buf, d2_len, MSG_WAITALL), d2_len);
+	EXPECT_EQ(memcmp(buf, data2, d2_len), 0);
+}
+
+/* Peek at an interleaved control record after the preceding data
+ * record has been consumed, then consume it.  MSG_PEEK exposes the
+ * control record's type without consuming it.
+ */
+TEST_F(tls, peek_cmsg_after_data)
+{
+	char *data = "leading";
+	char *ctrl = "middle";
+	char *tail = "trailing";
+	int d_len = strlen(data) + 1;
+	int c_len = strlen(ctrl) + 1;
+	int t_len = strlen(tail) + 1;
+	char buf[20];
+
+	if (self->notls)
+		SKIP(return, "no TLS support");
+
+	EXPECT_EQ(send(self->fd, data, d_len, 0), d_len);
+	EXPECT_EQ(tls_send_cmsg(self->fd, 100, ctrl, c_len, 0), c_len);
+	EXPECT_EQ(send(self->fd, tail, t_len, 0), t_len);
+
+	/* Consume leading data */
+	EXPECT_EQ(recv(self->cfd, buf, sizeof(buf), MSG_WAITALL), d_len);
+	EXPECT_EQ(memcmp(buf, data, d_len), 0);
+
+	/* Peek at the control record */
+	EXPECT_EQ(tls_recv_cmsg(_metadata, self->cfd, 100,
+				buf, sizeof(buf), MSG_PEEK), c_len);
+	EXPECT_EQ(memcmp(buf, ctrl, c_len), 0);
+
+	/* Consume the control record */
+	EXPECT_EQ(tls_recv_cmsg(_metadata, self->cfd, 100,
+				buf, sizeof(buf), 0), c_len);
+	EXPECT_EQ(memcmp(buf, ctrl, c_len), 0);
+
+	/* Trailing data */
+	EXPECT_EQ(recv(self->cfd, buf, t_len, MSG_WAITALL), t_len);
+	EXPECT_EQ(memcmp(buf, tail, t_len), 0);
+}
+
+/* Control record as the first record in the stream, followed by
+ * data.  The control record must be drained before the data record
+ * becomes available.
+ */
+TEST_F(tls, cmsg_before_data)
+{
+	char *ctrl = "alert";
+	char *data = "payload";
+	int c_len = strlen(ctrl) + 1;
+	int d_len = strlen(data) + 1;
+	char buf[20];
+
+	if (self->notls)
+		SKIP(return, "no TLS support");
+
+	EXPECT_EQ(tls_send_cmsg(self->fd, 100, ctrl, c_len, 0), c_len);
+	EXPECT_EQ(send(self->fd, data, d_len, 0), d_len);
+
+	/* recv without cmsg fails */
+	EXPECT_EQ(recv(self->cfd, buf, sizeof(buf), 0), -1);
+	EXPECT_EQ(errno, EIO);
+
+	/* Drain control via recvmsg */
+	EXPECT_EQ(tls_recv_cmsg(_metadata, self->cfd, 100,
+				buf, sizeof(buf), MSG_WAITALL), c_len);
+	EXPECT_EQ(memcmp(buf, ctrl, c_len), 0);
+
+	/* Data follows */
+	EXPECT_EQ(recv(self->cfd, buf, d_len, MSG_WAITALL), d_len);
+	EXPECT_EQ(memcmp(buf, data, d_len), 0);
+}
+
+/* Two different control record types interleaved with data.
+ * Each control record is delivered with its own type preserved;
+ * verify both types arrive intact.
+ */
+TEST_F(tls, mixed_control_types)
+{
+	char *data1 = "data1";
+	char *ctrl1 = "handshake";
+	char *ctrl2 = "alert_msg";
+	char *data2 = "data2";
+	int d1_len = strlen(data1) + 1;
+	int c1_len = strlen(ctrl1) + 1;
+	int c2_len = strlen(ctrl2) + 1;
+	int d2_len = strlen(data2) + 1;
+	char buf[20];
+
+	if (self->notls)
+		SKIP(return, "no TLS support");
+
+	EXPECT_EQ(send(self->fd, data1, d1_len, 0), d1_len);
+	EXPECT_EQ(tls_send_cmsg(self->fd, 100, ctrl1, c1_len, 0), c1_len);
+	EXPECT_EQ(tls_send_cmsg(self->fd, 200, ctrl2, c2_len, 0), c2_len);
+	EXPECT_EQ(send(self->fd, data2, d2_len, 0), d2_len);
+
+	/* First data */
+	EXPECT_EQ(recv(self->cfd, buf, sizeof(buf), MSG_WAITALL), d1_len);
+	EXPECT_EQ(memcmp(buf, data1, d1_len), 0);
+
+	/* First control (type 100) */
+	EXPECT_EQ(tls_recv_cmsg(_metadata, self->cfd, 100,
+				buf, sizeof(buf), MSG_WAITALL), c1_len);
+	EXPECT_EQ(memcmp(buf, ctrl1, c1_len), 0);
+
+	/* Second control (type 200) */
+	EXPECT_EQ(tls_recv_cmsg(_metadata, self->cfd, 200,
+				buf, sizeof(buf), MSG_WAITALL), c2_len);
+	EXPECT_EQ(memcmp(buf, ctrl2, c2_len), 0);
+
+	/* Second data */
+	EXPECT_EQ(recv(self->cfd, buf, d2_len, MSG_WAITALL), d2_len);
+	EXPECT_EQ(memcmp(buf, data2, d2_len), 0);
+}
+
+/* Trailing control record with no data following it.  The sender
+ * closes the connection after the control record; the receiver
+ * drains the control and then observes EOF.
+ */
+TEST_F(tls, data_cmsg_eof)
+{
+	char *data = "payload";
+	char *ctrl = "final";
+	int d_len = strlen(data) + 1;
+	int c_len = strlen(ctrl) + 1;
+	char buf[20];
+
+	if (self->notls)
+		SKIP(return, "no TLS support");
+
+	EXPECT_EQ(send(self->fd, data, d_len, 0), d_len);
+	EXPECT_EQ(tls_send_cmsg(self->fd, 100, ctrl, c_len, 0), c_len);
+	EXPECT_EQ(close(self->fd), 0);
+	self->fd = -1;
+
+	/* Consume data */
+	EXPECT_EQ(recv(self->cfd, buf, sizeof(buf), MSG_WAITALL), d_len);
+	EXPECT_EQ(memcmp(buf, data, d_len), 0);
+
+	/* Drain trailing control */
+	EXPECT_EQ(tls_recv_cmsg(_metadata, self->cfd, 100,
+				buf, sizeof(buf), 0), c_len);
+	EXPECT_EQ(memcmp(buf, ctrl, c_len), 0);
+
+	/* Next recv returns EOF */
+	EXPECT_EQ(recv(self->cfd, buf, sizeof(buf), 0), 0);
+}
+
 TEST_F(tls, shutdown)
 {
 	char const *test_str = "test_read";

-- 
2.54.0

[PATCH net-next v2 5/6] SUNRPC: Use read_sock_rectype for svcsock TCP receives

[Chuck Lever]

From: Chuck Lever <chuck.lever@oracle.com>

The svcsock TCP receive path uses sock_recvmsg() with ancillary data
buffers to detect TLS alerts when kTLS is active. This CMSG-based
approach requires a MSG_CTRUNC recovery dance on every receive and
cannot deliver control records through the read_sock interface.

When the socket provides a read_sock_rectype method (now set by
kTLS), svc_tcp_recvfrom() now dispatches to a new
svc_tcp_recvfrom_readsock() path. Two actor callbacks handle the
data:

svc_tcp_recv_actor() parses the RPC record byte stream directly from
skbs. Fragment header bytes fill sk_marker first; subsequent body
bytes are copied into rq_pages at the position tracked by
sk_datalen. When the last fragment of a complete RPC message
arrives, the actor sets desc->count to zero, stopping the read loop.

svc_tcp_rectype_actor() handles non-data TLS records. For fatal
alerts, the transport is marked for deferred close and the read loop
is stopped via desc->count. All non-data records are consumed by
returning 0 so that subsequent data records remain deliverable.

Sockets without read_sock_rectype (plain TCP, non-kTLS) continue to
use the existing sock_recvmsg() path unchanged. A follow-up patch
retires that path in favor of read_sock.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
 net/sunrpc/svcsock.c | 244 +++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 244 insertions(+)

diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index 50e5e7f5b762..e40931d11491 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -1134,6 +1134,247 @@ static void svc_tcp_fragment_received(struct svc_sock *svsk)
 	svsk->sk_marker = xdr_zero;
 }
 
+/*
+ * read_sock_rectype data actor: receives decrypted application data
+ * from the TLS layer, parsing the RPC record stream (fragment
+ * headers and message bodies) and assembling complete RPC messages
+ * into rqstp->rq_pages.
+ */
+static int svc_tcp_recv_actor(read_descriptor_t *desc,
+			      struct sk_buff *skb,
+			      unsigned int offset, size_t len)
+{
+	struct svc_rqst *rqstp = desc->arg.data;
+	struct svc_sock *svsk =
+		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
+	size_t reclen, received, want, take, done, n;
+	size_t consumed = 0;
+
+	if (!desc->count)
+		return 0;
+
+	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;
+
+		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);
+	done = 0;
+
+	while (done < take) {
+		unsigned int pg = svsk->sk_datalen >> PAGE_SHIFT;
+		unsigned int pg_off = svsk->sk_datalen & (PAGE_SIZE - 1);
+		size_t chunk = min(take - done,
+				   PAGE_SIZE - (size_t)pg_off);
+
+		if (skb_copy_bits(skb, offset,
+				  page_address(rqstp->rq_pages[pg]) + pg_off,
+				  chunk))
+			goto fault;
+		flush_dcache_page(rqstp->rq_pages[pg]);
+		offset += chunk;
+		done += chunk;
+		svsk->sk_datalen += chunk;
+	}
+	svsk->sk_tcplen += take;
+	consumed += take;
+
+	if (svsk->sk_tcplen - sizeof(rpc_fraghdr) >= reclen) {
+		if (svc_sock_final_rec(svsk))
+			desc->count = 0;
+		else
+			svc_tcp_fragment_received(svsk);
+	}
+
+	return consumed;
+
+fault:
+	desc->error = -EFAULT;
+	desc->count = 0;
+	return consumed;
+}
+
+/*
+ * read_sock_rectype non-data record actor: receives non-data TLS
+ * records (alerts, handshake messages) and translates them into
+ * transport-level actions.
+ *
+ * Returns 0 to consume the record and allow the TLS layer to
+ * continue delivering subsequent records. A negative return
+ * causes the TLS layer to requeue the skb on its rx_list,
+ * blocking all further record delivery on this connection.
+ */
+static int svc_tcp_rectype_actor(read_descriptor_t *desc,
+				 struct sk_buff *skb,
+				 unsigned int offset, size_t len,
+				 u8 rectype)
+{
+	struct svc_rqst *rqstp = desc->arg.data;
+	struct svc_sock *svsk =
+		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
+
+	switch (rectype) {
+	case TLS_RECORD_TYPE_ALERT: {
+		u8 alert[2] = {}, level, description;
+		struct kvec kvec = {
+			.iov_base = alert,
+			.iov_len = sizeof(alert),
+		};
+		struct msghdr msg = {};
+
+		if (skb_copy_bits(skb, offset, alert, min(len, sizeof(alert))))
+			break;
+		iov_iter_kvec(&msg.msg_iter, ITER_DEST, &kvec, 1, sizeof(alert));
+		tls_alert_recv(svsk->sk_sk, &msg, &level, &description);
+		if (level == TLS_ALERT_LEVEL_FATAL) {
+			svc_xprt_deferred_close(&svsk->sk_xprt);
+			desc->error = -ENOTCONN;
+			desc->count = 0;
+		}
+		break;
+	}
+	default:
+		break;
+	}
+	return 0;
+}
+
+static int svc_tcp_recvfrom_readsock(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;
+	struct sock	*sk = svsk->sk_sk;
+	read_descriptor_t desc = {
+		.arg.data = rqstp,
+	};
+	ssize_t len;
+	__be32 *p;
+	__be32 calldir;
+
+	clear_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
+
+	svc_tcp_restore_pages(svsk, rqstp);
+	rqstp->rq_arg.head[0].iov_base = page_address(rqstp->rq_pages[0]);
+
+	desc.count = serv->sv_max_mesg;
+	lock_sock(sk);
+	len = svsk->sk_sock->ops->read_sock_rectype(sk, &desc,
+						    svc_tcp_recv_actor,
+						    svc_tcp_rectype_actor);
+	release_sock(sk);
+
+	if (desc.error < 0) {
+		len = desc.error;
+		goto err_discard;
+	}
+	if (desc.count != 0) {
+		if (len > 0)
+			set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
+		goto err_incomplete;
+	}
+
+	if (svsk->sk_datalen < 8)
+		goto err_nuts;
+
+	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) {
+		rqstp->rq_arg.head[0].iov_len = rqstp->rq_arg.len;
+		rqstp->rq_arg.page_len = 0;
+	} else {
+		rqstp->rq_arg.page_len = rqstp->rq_arg.len -
+			rqstp->rq_arg.head[0].iov_len;
+	}
+
+	rqstp->rq_xprt_ctxt   = NULL;
+	rqstp->rq_prot	      = IPPROTO_TCP;
+	if (test_bit(XPT_LOCAL, &svsk->sk_xprt.xpt_flags))
+		set_bit(RQ_LOCAL, &rqstp->rq_flags);
+	else
+		clear_bit(RQ_LOCAL, &rqstp->rq_flags);
+
+	p = (__be32 *)rqstp->rq_arg.head[0].iov_base;
+	calldir = p[1];
+	if (calldir)
+		len = receive_cb_reply(svsk, rqstp);
+
+	/* Reset TCP read info */
+	svsk->sk_datalen = 0;
+	svc_tcp_fragment_received(svsk);
+
+	if (len < 0)
+		goto error;
+
+	trace_svcsock_tcp_recv(&svsk->sk_xprt, rqstp->rq_arg.len);
+	svc_xprt_copy_addrs(rqstp, &svsk->sk_xprt);
+	if (serv->sv_stats)
+		serv->sv_stats->nettcpcnt++;
+
+	svc_sock_secure_port(rqstp);
+	set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
+	svc_xprt_received(rqstp->rq_xprt);
+	return rqstp->rq_arg.len;
+
+err_incomplete:
+	svc_tcp_save_pages(svsk, rqstp);
+	if (len < 0 && len != -EAGAIN)
+		goto err_delete;
+	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));
+	goto err_noclose;
+error:
+	if (len != -EAGAIN)
+		goto err_delete;
+	trace_svcsock_tcp_recv_eagain(&svsk->sk_xprt, 0);
+	goto err_noclose;
+err_nuts:
+	svsk->sk_datalen = 0;
+	goto err_delete;
+err_discard:
+	/*
+	 * Clear sk_datalen so the teardown-time
+	 * svc_tcp_clear_pages() does not walk the emptied
+	 * svsk->sk_pages[].
+	 */
+	svsk->sk_datalen = 0;
+err_delete:
+	trace_svcsock_tcp_recv_err(&svsk->sk_xprt, len);
+	svc_xprt_deferred_close(&svsk->sk_xprt);
+err_noclose:
+	svc_xprt_received(rqstp->rq_xprt);
+	return 0;
+}
+
 /**
  * svc_tcp_recvfrom - Receive data from a TCP socket
  * @rqstp: request structure into which to receive an RPC Call
@@ -1162,6 +1403,9 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 	__be32 *p;
 	__be32 calldir;
 
+	if (svsk->sk_sock->ops->read_sock_rectype)
+		return svc_tcp_recvfrom_readsock(rqstp);
+
 	clear_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
 	len = svc_tcp_read_marker(svsk, rqstp);
 	if (len < 0)

-- 
2.54.0

[PATCH net-next v2 6/6] SUNRPC: Remove sock_recvmsg path from svcsock TCP receives

[Chuck Lever]

From: Chuck Lever <chuck.lever@oracle.com>

The svcsock TCP receive path maintains two code paths: one
using read_sock/read_sock_rectype and a legacy path using
sock_recvmsg. Plain TCP sockets already provide read_sock
(tcp_read_sock) in their proto_ops, so a single
read_sock-based receive path handles all cases relevant to
NFSD, using read_sock_rectype under kTLS and read_sock
otherwise.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
 net/sunrpc/svcsock.c | 329 ++++-----------------------------------------------
 1 file changed, 26 insertions(+), 303 deletions(-)

diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index e40931d11491..9b9e0da9e73c 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>
  */
 
@@ -238,138 +229,6 @@ static int svc_one_sock_name(struct svc_sock *svsk, char *buf, int remaining)
 	return len;
 }
 
-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);
-		ret = (level == TLS_ALERT_LEVEL_FATAL) ?
-			-ENOTCONN : -EAGAIN;
-		break;
-	default:
-		/* discard this record type */
-		ret = -EAGAIN;
-	}
-	return ret;
-}
-
-static int
-svc_tcp_sock_recv_cmsg(struct socket *sock, 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),
-	};
-	struct msghdr msg = {
-		.msg_flags = *msg_flags,
-		.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);
-	if (ret > 0 &&
-	    tls_get_record_type(sock->sk, &u.cmsg) == TLS_RECORD_TYPE_ALERT) {
-		iov_iter_revert(&msg.msg_iter, ret);
-		ret = svc_tcp_sock_process_cmsg(sock, &msg, &u.cmsg, -EAGAIN);
-	}
-	return ret;
-}
-
-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, &msg->msg_flags);
-	}
-	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
- * 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 (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!)
-	 */
-	if (len == buflen)
-		set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
-
-	return len;
-}
-
 /*
  * Set socket snd and rcv buffer lengths
  */
@@ -1048,50 +907,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;
@@ -1135,10 +950,10 @@ static void svc_tcp_fragment_received(struct svc_sock *svsk)
 }
 
 /*
- * read_sock_rectype data actor: receives decrypted application data
- * from the TLS layer, parsing the RPC record stream (fragment
- * headers and message bodies) and assembling complete RPC messages
- * into rqstp->rq_pages.
+ * read_sock data actor: receives application data from the
+ * transport socket, parsing the RPC record stream (fragment
+ * headers and message bodies) and assembling complete RPC
+ * messages into rqstp->rq_pages.
  */
 static int svc_tcp_recv_actor(read_descriptor_t *desc,
 			      struct sk_buff *skb,
@@ -1266,7 +1081,21 @@ static int svc_tcp_rectype_actor(read_descriptor_t *desc,
 	return 0;
 }
 
-static int svc_tcp_recvfrom_readsock(struct svc_rqst *rqstp)
+/**
+ * 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.
+ *
+ * Returns:
+ *   On success, the number of bytes in a received RPC Call, or
+ *   %0 if a complete RPC Call message was not ready to return
+ *
+ * The zero return case handles partial receives and callback Replies.
+ * The state of a partial receive is preserved in the svc_sock for
+ * the next call to svc_tcp_recvfrom.
+ */
+static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 {
 	struct svc_sock	*svsk =
 		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
@@ -1286,9 +1115,13 @@ static int svc_tcp_recvfrom_readsock(struct svc_rqst *rqstp)
 
 	desc.count = serv->sv_max_mesg;
 	lock_sock(sk);
-	len = svsk->sk_sock->ops->read_sock_rectype(sk, &desc,
-						    svc_tcp_recv_actor,
-						    svc_tcp_rectype_actor);
+	if (svsk->sk_sock->ops->read_sock_rectype)
+		len = svsk->sk_sock->ops->read_sock_rectype(sk, &desc,
+							    svc_tcp_recv_actor,
+							    svc_tcp_rectype_actor);
+	else
+		len = svsk->sk_sock->ops->read_sock(sk, &desc,
+						     svc_tcp_recv_actor);
 	release_sock(sk);
 
 	if (desc.error < 0) {
@@ -1375,116 +1208,6 @@ static int svc_tcp_recvfrom_readsock(struct svc_rqst *rqstp)
 	return 0;
 }
 
-/**
- * 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.
- *
- * Returns:
- *   On success, the number of bytes in a received RPC Call, or
- *   %0 if a complete RPC Call message was not ready to return
- *
- * The zero return case handles partial receives and callback Replies.
- * The state of a partial receive is preserved in the svc_sock for
- * the next call to svc_tcp_recvfrom.
- */
-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;
-	ssize_t len;
-	__be32 *p;
-	__be32 calldir;
-
-	if (svsk->sk_sock->ops->read_sock_rectype)
-		return svc_tcp_recvfrom_readsock(rqstp);
-
-	clear_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
-	len = svc_tcp_read_marker(svsk, rqstp);
-	if (len < 0)
-		goto error;
-
-	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;
-	}
-	if (len != want || !svc_sock_final_rec(svsk))
-		goto err_incomplete;
-	if (svsk->sk_datalen < 8)
-		goto err_nuts;
-
-	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) {
-		rqstp->rq_arg.head[0].iov_len = rqstp->rq_arg.len;
-		rqstp->rq_arg.page_len = 0;
-	} else
-		rqstp->rq_arg.page_len = rqstp->rq_arg.len - rqstp->rq_arg.head[0].iov_len;
-
-	rqstp->rq_xprt_ctxt   = NULL;
-	rqstp->rq_prot	      = IPPROTO_TCP;
-	if (test_bit(XPT_LOCAL, &svsk->sk_xprt.xpt_flags))
-		set_bit(RQ_LOCAL, &rqstp->rq_flags);
-	else
-		clear_bit(RQ_LOCAL, &rqstp->rq_flags);
-
-	p = (__be32 *)rqstp->rq_arg.head[0].iov_base;
-	calldir = p[1];
-	if (calldir)
-		len = receive_cb_reply(svsk, rqstp);
-
-	/* Reset TCP read info */
-	svsk->sk_datalen = 0;
-	svc_tcp_fragment_received(svsk);
-
-	if (len < 0)
-		goto error;
-
-	svc_xprt_copy_addrs(rqstp, &svsk->sk_xprt);
-	if (serv->sv_stats)
-		serv->sv_stats->nettcpcnt++;
-
-	svc_sock_secure_port(rqstp);
-	svc_xprt_received(rqstp->rq_xprt);
-	return rqstp->rq_arg.len;
-
-err_incomplete:
-	svc_tcp_save_pages(svsk, rqstp);
-	if (len < 0 && len != -EAGAIN)
-		goto err_delete;
-	if (len == want)
-		svc_tcp_fragment_received(svsk);
-	else
-		trace_svcsock_tcp_recv_short(&svsk->sk_xprt,
-				svc_sock_reclen(svsk),
-				svsk->sk_tcplen - sizeof(rpc_fraghdr));
-	goto err_noclose;
-error:
-	if (len != -EAGAIN)
-		goto err_delete;
-	trace_svcsock_tcp_recv_eagain(&svsk->sk_xprt, 0);
-	goto err_noclose;
-err_nuts:
-	svsk->sk_datalen = 0;
-err_delete:
-	trace_svcsock_tcp_recv_err(&svsk->sk_xprt, len);
-	svc_xprt_deferred_close(&svsk->sk_xprt);
-err_noclose:
-	svc_xprt_received(rqstp->rq_xprt);
-	return 0;	/* record not complete */
-}
-
 /*
  * MSG_SPLICE_PAGES is used exclusively to reduce the number of
  * copy operations in this path. Therefore the caller must ensure

-- 
2.54.0