NFSD: Use direct I/O only for the last operation in a COMPOUND

A direct read widens the requested byte range to the file system's alignment and leaves the payload at a nonzero rq_res.page_base. The xdr_stream encoder does not account for page_base:
xdr_reserve_space_vec() sizes each chunk from page_len alone, and xdr_get_next_encode_buffer() opens each new page at its first byte. The stream's position lags the end of the payload by page_base bytes.

An NFSv4 operation encoded after such a READ overwrites the tail of the payload, and the transport sends alignment padding in place of that operation's result. The client sees corrupted file data followed by an undecodable reply. NFSv2 and NFSv3 are unaffected:
their READ encoders pass page_base explicitly, and nothing is encoded into the pages after the payload.

Have nfsd4_read() permit a direct read only for the last operation in its COMPOUND, next to the same decision it already makes for splice reads, and pass the result to nfsd_iter_read() as a new argument. The NFSv2 and NFSv3 READ path always permits a direct read. The decision is NFSv4 reply-encoding policy, so it stays out of fs/nfsd/vfs.c.

Fixes d686e64e931c ("NFSD: Implement NFSD_IO_DIRECT for NFS READ")

Signed-off-by Chuck Lever <[email protected]>

---
Changes in v2:
- Split from the "NFSD short subjects" series and post on its own
- Move NFSv4-specific logic out of fs/nfsd/vfs.c
- Link to v1: https://lore.kernel.org/r/[email protected]

 fs/nfsd/nfs4proc.c |  2 ++
 fs/nfsd/nfs4xdr.c  |  3 ++-
 fs/nfsd/vfs.c      | 15 ++++++++++++---
 fs/nfsd/vfs.h      |  3 ++-
 fs/nfsd/xdr4.h     |  1 +
 5 files changed, 19 insertions(+), 5 deletions(-)

fs/nfsd/nfs4proc.c +2 -0

index a50f9dcb3a31..aaca7d3c3e04 100644
@@ -1143,6 +1143,8 @@ nfsd4_read(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
 
 		argp->splice_ok = false;
 	}
+	/* The xdr_stream encoder does not account for rq_res.page_base */
+	read->rd_dio_ok = nfsd4_last_compound_op(rqstp);
 
 	/* check stateid */
 	status = nfs4_preprocess_stateid_op(rqstp, cstate, &cstate->current_fh,

fs/nfsd/nfs4xdr.c +2 -1

index c4fd840da4d9..feb50abccff4 100644
@@ -5308,7 +5308,8 @@ static __be32 nfsd4_encode_readv(struct nfsd4_compoundres *resp,
 	__be32 nfserr;
 
 	nfserr = nfsd_iter_read(resp->rqstp, read->rd_fhp, read->rd_nf,
-				read->rd_offset, &maxcount, &read->rd_eof);
+				read->rd_offset, &maxcount, &read->rd_eof,
+				read->rd_dio_ok);
 	read->rd_length = maxcount;
 	if (nfserr)
 		return nfserr;

fs/nfsd/vfs.c +12 -3

index cb4bf7b235b4..97077f18a1b7 100644
@@ -1234,6 +1234,8 @@ nfsd_direct_read(struct svc_rqst *rqstp, struct svc_fh *fhp,
  * @offset: starting byte offset
  * @count: IN: requested number of bytes; OUT: number of bytes read
  * @eof: OUT: set non-zero if operation reached the end of the file
+ * @dio_ok: false forces a buffered read even when the export selects
+ *	    direct I/O
  *
  * Some filesystems or situations cannot use nfsd_splice_read. This
  * function is the slightly less-performant fallback for those cases.

@@ -1241,12 +1243,17 @@ nfsd_direct_read(struct svc_rqst *rqstp, struct svc_fh *fhp,
  * records. On return, @rqstp->rq_next_page points past the last page
  * offered to the read.
  *
+ * A direct read leaves its alignment pad at the front of the pages and
+ * records the pad length in @rqstp->rq_res.page_base. The xdr_stream
+ * encoder does not account for page_base, so pass @dio_ok false when
+ * more XDR items follow the payload in the pages.
+ *
  * Returns nfs_ok on success, otherwise an nfserr stat value is
  * returned.
  */
 __be32 nfsd_iter_read(struct svc_rqst *rqstp, struct svc_fh *fhp,
 		      struct nfsd_file *nf, loff_t offset, unsigned long *count,
-		      u32 *eof)
+		      u32 *eof, bool dio_ok)
 {
 	struct xdr_buf *buf = &rqstp->rq_res;
 	struct page **page = buf->pages + (buf->page_len >> PAGE_SHIFT);

@@ -1265,7 +1272,8 @@ __be32 nfsd_iter_read(struct svc_rqst *rqstp, struct svc_fh *fhp,
 		break;
 	case NFSD_IO_DIRECT:
 		/* When dio_read_offset_align is zero, dio is not supported */
-		if (nf->nf_dio_read_offset_align && !rqstp->rq_res.page_len)
+		if (dio_ok && nf->nf_dio_read_offset_align &&
+		    !rqstp->rq_res.page_len)
 			return nfsd_direct_read(rqstp, fhp, nf, offset,
 						count, page, eof);
 		fallthrough;

@@ -1649,7 +1657,8 @@ __be32 nfsd_read(struct svc_rqst *rqstp, struct svc_fh *fhp,
 	if (file->f_op->splice_read && nfsd_read_splice_ok(rqstp))
 		err = nfsd_splice_read(rqstp, fhp, file, offset, count, eof);
 	else
-		err = nfsd_iter_read(rqstp, fhp, nf, offset, count, eof);
+		err = nfsd_iter_read(rqstp, fhp, nf, offset, count, eof,
+				     true);
 
 	nfsd_file_put(nf);
 	trace_nfsd_read_done(rqstp, fhp, offset, *count);

fs/nfsd/vfs.h +2 -1

index 1f05c83a3ea6..0b0152ba56b3 100644
@@ -182,7 +182,8 @@ __be32		nfsd_splice_read(struct svc_rqst *rqstp, struct svc_fh *fhp,
 				u32 *eof);
 __be32		nfsd_iter_read(struct svc_rqst *rqstp, struct svc_fh *fhp,
 				struct nfsd_file *nf, loff_t offset,
-				unsigned long *count, u32 *eof);
+				unsigned long *count, u32 *eof,
+				bool dio_ok);
 bool		nfsd_read_splice_ok(struct svc_rqst *rqstp);
 __be32		nfsd_read(struct svc_rqst *rqstp, struct svc_fh *fhp,
 				loff_t offset, unsigned long *count,

fs/nfsd/xdr4.h +0 -0

index ec198993429e..61afbabc1773 100644