localio.c

   1// SPDX-License-Identifier: GPL-2.0-only
   2/*
   3 * NFS client support for local clients to bypass network stack
   4 *
   5 * Copyright (C) 2014 Weston Andros Adamson <dros@primarydata.com>
   6 * Copyright (C) 2019 Trond Myklebust <trond.myklebust@hammerspace.com>
   7 * Copyright (C) 2024 Mike Snitzer <snitzer@hammerspace.com>
   8 * Copyright (C) 2024 NeilBrown <neilb@suse.de>
   9 */
  10
  11#include <linux/module.h>
  12#include <linux/errno.h>
  13#include <linux/vfs.h>
  14#include <linux/file.h>
  15#include <linux/inet.h>
  16#include <linux/sunrpc/addr.h>
  17#include <linux/inetdevice.h>
  18#include <net/addrconf.h>
  19#include <linux/nfs_common.h>
  20#include <linux/nfslocalio.h>
  21#include <linux/bvec.h>
  22
  23#include <linux/nfs.h>
  24#include <linux/nfs_fs.h>
  25#include <linux/nfs_xdr.h>
  26
  27#include "internal.h"
  28#include "pnfs.h"
  29#include "nfstrace.h"
  30
  31#define NFSDBG_FACILITY		NFSDBG_VFS
  32
  33#define NFSLOCAL_MAX_IOS	3
  34
  35struct nfs_local_kiocb {
  36	struct kiocb		kiocb;
  37	struct bio_vec		*bvec;
  38	struct nfs_pgio_header	*hdr;
  39	struct work_struct	work;
  40	void (*aio_complete_work)(struct work_struct *);
  41	struct nfsd_file	*localio;
  42	/* Begin mostly DIO-specific members */
  43	size_t                  end_len;
  44	short int		end_iter_index;
  45	atomic_t		n_iters;
  46	struct iov_iter		iters[NFSLOCAL_MAX_IOS];
  47	bool			iter_is_dio_aligned[NFSLOCAL_MAX_IOS];
  48	/* End mostly DIO-specific members */
  49};
  50
  51struct nfs_local_fsync_ctx {
  52	struct nfsd_file	*localio;
  53	struct nfs_commit_data	*data;
  54	struct work_struct	work;
  55	struct completion	*done;
  56};
  57
  58static bool localio_enabled __read_mostly = true;
  59module_param(localio_enabled, bool, 0644);
  60
  61static int nfs_local_do_read(struct nfs_local_kiocb *iocb,
  62			     const struct rpc_call_ops *call_ops);
  63static int nfs_local_do_write(struct nfs_local_kiocb *iocb,
  64			      const struct rpc_call_ops *call_ops);
  65
  66static inline bool nfs_client_is_local(const struct nfs_client *clp)
  67{
  68	return !!rcu_access_pointer(clp->cl_uuid.net);
  69}
  70
  71bool nfs_server_is_local(const struct nfs_client *clp)
  72{
  73	return nfs_client_is_local(clp) && localio_enabled;
  74}
  75EXPORT_SYMBOL_GPL(nfs_server_is_local);
  76
  77/*
  78 * UUID_IS_LOCAL XDR functions
  79 */
  80
  81static void localio_xdr_enc_uuidargs(struct rpc_rqst *req,
  82				     struct xdr_stream *xdr,
  83				     const void *data)
  84{
  85	const u8 *uuid = data;
  86
  87	encode_opaque_fixed(xdr, uuid, UUID_SIZE);
  88}
  89
  90static int localio_xdr_dec_uuidres(struct rpc_rqst *req,
  91				   struct xdr_stream *xdr,
  92				   void *result)
  93{
  94	/* void return */
  95	return 0;
  96}
  97
  98static const struct rpc_procinfo nfs_localio_procedures[] = {
  99	[LOCALIOPROC_UUID_IS_LOCAL] = {
 100		.p_proc = LOCALIOPROC_UUID_IS_LOCAL,
 101		.p_encode = localio_xdr_enc_uuidargs,
 102		.p_decode = localio_xdr_dec_uuidres,
 103		.p_arglen = XDR_QUADLEN(UUID_SIZE),
 104		.p_replen = 0,
 105		.p_statidx = LOCALIOPROC_UUID_IS_LOCAL,
 106		.p_name = "UUID_IS_LOCAL",
 107	},
 108};
 109
 110static unsigned int nfs_localio_counts[ARRAY_SIZE(nfs_localio_procedures)];
 111static const struct rpc_version nfslocalio_version1 = {
 112	.number			= 1,
 113	.nrprocs		= ARRAY_SIZE(nfs_localio_procedures),
 114	.procs			= nfs_localio_procedures,
 115	.counts			= nfs_localio_counts,
 116};
 117
 118static const struct rpc_version *nfslocalio_version[] = {
 119       [1]			= &nfslocalio_version1,
 120};
 121
 122extern const struct rpc_program nfslocalio_program;
 123static struct rpc_stat		nfslocalio_rpcstat = { &nfslocalio_program };
 124
 125const struct rpc_program nfslocalio_program = {
 126	.name			= "nfslocalio",
 127	.number			= NFS_LOCALIO_PROGRAM,
 128	.nrvers			= ARRAY_SIZE(nfslocalio_version),
 129	.version		= nfslocalio_version,
 130	.stats			= &nfslocalio_rpcstat,
 131};
 132
 133/*
 134 * nfs_init_localioclient - Initialise an NFS localio client connection
 135 */
 136static struct rpc_clnt *nfs_init_localioclient(struct nfs_client *clp)
 137{
 138	struct rpc_clnt *rpcclient_localio;
 139
 140	rpcclient_localio = rpc_bind_new_program(clp->cl_rpcclient,
 141						 &nfslocalio_program, 1);
 142
 143	dprintk_rcu("%s: server (%s) %s NFS LOCALIO.\n",
 144		__func__, rpc_peeraddr2str(clp->cl_rpcclient, RPC_DISPLAY_ADDR),
 145		(IS_ERR(rpcclient_localio) ? "does not support" : "supports"));
 146
 147	return rpcclient_localio;
 148}
 149
 150static bool nfs_server_uuid_is_local(struct nfs_client *clp)
 151{
 152	u8 uuid[UUID_SIZE];
 153	struct rpc_message msg = {
 154		.rpc_argp = &uuid,
 155	};
 156	struct rpc_clnt *rpcclient_localio;
 157	int status;
 158
 159	rpcclient_localio = nfs_init_localioclient(clp);
 160	if (IS_ERR(rpcclient_localio))
 161		return false;
 162
 163	export_uuid(uuid, &clp->cl_uuid.uuid);
 164
 165	msg.rpc_proc = &nfs_localio_procedures[LOCALIOPROC_UUID_IS_LOCAL];
 166	status = rpc_call_sync(rpcclient_localio, &msg, 0);
 167	dprintk("%s: NFS reply UUID_IS_LOCAL: status=%d\n",
 168		__func__, status);
 169	rpc_shutdown_client(rpcclient_localio);
 170
 171	/* Server is only local if it initialized required struct members */
 172	if (status || !rcu_access_pointer(clp->cl_uuid.net) || !clp->cl_uuid.dom)
 173		return false;
 174
 175	return true;
 176}
 177
 178/*
 179 * nfs_local_probe - probe local i/o support for an nfs_server and nfs_client
 180 * - called after alloc_client and init_client (so cl_rpcclient exists)
 181 * - this function is idempotent, it can be called for old or new clients
 182 */
 183static void nfs_local_probe(struct nfs_client *clp)
 184{
 185	/* Disallow localio if disabled via sysfs or AUTH_SYS isn't used */
 186	if (!localio_enabled ||
 187	    clp->cl_rpcclient->cl_auth->au_flavor != RPC_AUTH_UNIX) {
 188		nfs_localio_disable_client(clp);
 189		return;
 190	}
 191
 192	if (nfs_client_is_local(clp))
 193		return;
 194
 195	if (!nfs_uuid_begin(&clp->cl_uuid))
 196		return;
 197	if (nfs_server_uuid_is_local(clp))
 198		nfs_localio_enable_client(clp);
 199	nfs_uuid_end(&clp->cl_uuid);
 200}
 201
 202void nfs_local_probe_async_work(struct work_struct *work)
 203{
 204	struct nfs_client *clp =
 205		container_of(work, struct nfs_client, cl_local_probe_work);
 206
 207	if (!refcount_inc_not_zero(&clp->cl_count))
 208		return;
 209	nfs_local_probe(clp);
 210	nfs_put_client(clp);
 211}
 212
 213void nfs_local_probe_async(struct nfs_client *clp)
 214{
 215	queue_work(nfsiod_workqueue, &clp->cl_local_probe_work);
 216}
 217EXPORT_SYMBOL_GPL(nfs_local_probe_async);
 218
 219static inline void nfs_local_file_put(struct nfsd_file *localio)
 220{
 221	/* nfs_to_nfsd_file_put_local() expects an __rcu pointer
 222	 * but we have a __kernel pointer.  It is always safe
 223	 * to cast a __kernel pointer to an __rcu pointer
 224	 * because the cast only weakens what is known about the pointer.
 225	 */
 226	struct nfsd_file __rcu *nf = (struct nfsd_file __rcu*) localio;
 227
 228	nfs_to_nfsd_file_put_local(&nf);
 229}
 230
 231/*
 232 * __nfs_local_open_fh - open a local filehandle in terms of nfsd_file.
 233 *
 234 * Returns a pointer to a struct nfsd_file or ERR_PTR.
 235 * Caller must release returned nfsd_file with nfs_to_nfsd_file_put_local().
 236 */
 237static struct nfsd_file *
 238__nfs_local_open_fh(struct nfs_client *clp, const struct cred *cred,
 239		    struct nfs_fh *fh, struct nfs_file_localio *nfl,
 240		    struct nfsd_file __rcu **pnf,
 241		    const fmode_t mode)
 242{
 243	int status = 0;
 244	struct nfsd_file *localio;
 245
 246	localio = nfs_open_local_fh(&clp->cl_uuid, clp->cl_rpcclient,
 247				    cred, fh, nfl, pnf, mode);
 248	if (IS_ERR(localio)) {
 249		status = PTR_ERR(localio);
 250		switch (status) {
 251		case -ENOMEM:
 252		case -ENXIO:
 253		case -ENOENT:
 254			/* Revalidate localio */
 255			nfs_localio_disable_client(clp);
 256			nfs_local_probe(clp);
 257		}
 258	}
 259	trace_nfs_local_open_fh(fh, mode, status);
 260	return localio;
 261}
 262
 263/*
 264 * nfs_local_open_fh - open a local filehandle in terms of nfsd_file.
 265 * First checking if the open nfsd_file is already cached, otherwise
 266 * must __nfs_local_open_fh and insert the nfsd_file in nfs_file_localio.
 267 *
 268 * Returns a pointer to a struct nfsd_file or NULL.
 269 */
 270struct nfsd_file *
 271nfs_local_open_fh(struct nfs_client *clp, const struct cred *cred,
 272		  struct nfs_fh *fh, struct nfs_file_localio *nfl,
 273		  const fmode_t mode)
 274{
 275	struct nfsd_file *nf, __rcu **pnf;
 276
 277	if (!nfs_server_is_local(clp))
 278		return NULL;
 279	if (mode & ~(FMODE_READ | FMODE_WRITE))
 280		return NULL;
 281
 282	if (mode & FMODE_WRITE)
 283		pnf = &nfl->rw_file;
 284	else
 285		pnf = &nfl->ro_file;
 286
 287	nf = __nfs_local_open_fh(clp, cred, fh, nfl, pnf, mode);
 288	if (IS_ERR(nf))
 289		return NULL;
 290	return nf;
 291}
 292EXPORT_SYMBOL_GPL(nfs_local_open_fh);
 293
 294/*
 295 * Ensure all page cache allocations are done from GFP_NOFS context to
 296 * prevent direct reclaim recursion back into NFS via nfs_writepages.
 297 */
 298static void
 299nfs_local_mapping_set_gfp_nofs_context(struct address_space *m)
 300{
 301	gfp_t gfp_mask = mapping_gfp_mask(m);
 302
 303	mapping_set_gfp_mask(m, (gfp_mask & ~(__GFP_FS)));
 304}
 305
 306static void
 307nfs_local_iocb_free(struct nfs_local_kiocb *iocb)
 308{
 309	kfree(iocb->bvec);
 310	kfree(iocb);
 311}
 312
 313static struct nfs_local_kiocb *
 314nfs_local_iocb_alloc(struct nfs_pgio_header *hdr,
 315		     struct file *file, gfp_t flags)
 316{
 317	struct nfs_local_kiocb *iocb;
 318
 319	iocb = kzalloc(sizeof(*iocb), flags);
 320	if (iocb == NULL)
 321		return NULL;
 322
 323	iocb->bvec = kmalloc_array(hdr->page_array.npages,
 324				   sizeof(struct bio_vec), flags);
 325	if (iocb->bvec == NULL) {
 326		kfree(iocb);
 327		return NULL;
 328	}
 329
 330	nfs_local_mapping_set_gfp_nofs_context(file->f_mapping);
 331	init_sync_kiocb(&iocb->kiocb, file);
 332
 333	iocb->hdr = hdr;
 334	iocb->kiocb.ki_pos = hdr->args.offset;
 335	iocb->kiocb.ki_flags &= ~IOCB_APPEND;
 336	iocb->kiocb.ki_complete = NULL;
 337	iocb->aio_complete_work = NULL;
 338
 339	iocb->end_iter_index = -1;
 340
 341	return iocb;
 342}
 343
 344static bool
 345nfs_is_local_dio_possible(struct nfs_local_kiocb *iocb, int rw,
 346			  size_t len, struct nfs_local_dio *local_dio)
 347{
 348	struct nfs_pgio_header *hdr = iocb->hdr;
 349	loff_t offset = hdr->args.offset;
 350	u32 nf_dio_mem_align, nf_dio_offset_align, nf_dio_read_offset_align;
 351	loff_t start_end, orig_end, middle_end;
 352
 353	nfs_to->nfsd_file_dio_alignment(iocb->localio, &nf_dio_mem_align,
 354			&nf_dio_offset_align, &nf_dio_read_offset_align);
 355	if (rw == ITER_DEST)
 356		nf_dio_offset_align = nf_dio_read_offset_align;
 357
 358	if (unlikely(!nf_dio_mem_align || !nf_dio_offset_align))
 359		return false;
 360	if (unlikely(len < nf_dio_offset_align))
 361		return false;
 362
 363	local_dio->mem_align = nf_dio_mem_align;
 364	local_dio->offset_align = nf_dio_offset_align;
 365
 366	start_end = round_up(offset, nf_dio_offset_align);
 367	orig_end = offset + len;
 368	middle_end = round_down(orig_end, nf_dio_offset_align);
 369
 370	local_dio->middle_offset = start_end;
 371	local_dio->end_offset = middle_end;
 372
 373	local_dio->start_len = start_end - offset;
 374	local_dio->middle_len = middle_end - start_end;
 375	local_dio->end_len = orig_end - middle_end;
 376
 377	if (rw == ITER_DEST)
 378		trace_nfs_local_dio_read(hdr->inode, offset, len, local_dio);
 379	else
 380		trace_nfs_local_dio_write(hdr->inode, offset, len, local_dio);
 381	return true;
 382}
 383
 384static bool nfs_iov_iter_aligned_bvec(const struct iov_iter *i,
 385		unsigned int addr_mask, unsigned int len_mask)
 386{
 387	const struct bio_vec *bvec = i->bvec;
 388	size_t skip = i->iov_offset;
 389	size_t size = i->count;
 390
 391	if (size & len_mask)
 392		return false;
 393	do {
 394		size_t len = bvec->bv_len;
 395
 396		if (len > size)
 397			len = size;
 398		if ((unsigned long)(bvec->bv_offset + skip) & addr_mask)
 399			return false;
 400		bvec++;
 401		size -= len;
 402		skip = 0;
 403	} while (size);
 404
 405	return true;
 406}
 407
 408static void
 409nfs_local_iter_setup(struct iov_iter *iter, int rw, struct bio_vec *bvec,
 410		     unsigned int nvecs, unsigned long total,
 411		     size_t start, size_t len)
 412{
 413	iov_iter_bvec(iter, rw, bvec, nvecs, total);
 414	if (start)
 415		iov_iter_advance(iter, start);
 416	iov_iter_truncate(iter, len);
 417}
 418
 419/*
 420 * Setup as many as 3 iov_iter based on extents described by @local_dio.
 421 * Returns the number of iov_iter that were setup.
 422 */
 423static int
 424nfs_local_iters_setup_dio(struct nfs_local_kiocb *iocb, int rw,
 425			  unsigned int nvecs, unsigned long total,
 426			  struct nfs_local_dio *local_dio)
 427{
 428	int n_iters = 0;
 429	struct iov_iter *iters = iocb->iters;
 430
 431	/* Setup misaligned start? */
 432	if (local_dio->start_len) {
 433		nfs_local_iter_setup(&iters[n_iters], rw, iocb->bvec,
 434				     nvecs, total, 0, local_dio->start_len);
 435		++n_iters;
 436	}
 437
 438	/*
 439	 * Setup DIO-aligned middle, if there is no misaligned end (below)
 440	 * then AIO completion is used, see nfs_local_call_{read,write}
 441	 */
 442	nfs_local_iter_setup(&iters[n_iters], rw, iocb->bvec, nvecs,
 443			     total, local_dio->start_len, local_dio->middle_len);
 444
 445	iocb->iter_is_dio_aligned[n_iters] =
 446		nfs_iov_iter_aligned_bvec(&iters[n_iters],
 447			local_dio->mem_align-1, local_dio->offset_align-1);
 448
 449	if (unlikely(!iocb->iter_is_dio_aligned[n_iters])) {
 450		trace_nfs_local_dio_misaligned(iocb->hdr->inode,
 451			local_dio->start_len, local_dio->middle_len, local_dio);
 452		return 0; /* no DIO-aligned IO possible */
 453	}
 454	iocb->end_iter_index = n_iters;
 455	++n_iters;
 456
 457	/* Setup misaligned end? */
 458	if (local_dio->end_len) {
 459		nfs_local_iter_setup(&iters[n_iters], rw, iocb->bvec,
 460				     nvecs, total, local_dio->start_len +
 461				     local_dio->middle_len, local_dio->end_len);
 462		iocb->end_iter_index = n_iters;
 463		++n_iters;
 464	}
 465
 466	atomic_set(&iocb->n_iters, n_iters);
 467	return n_iters;
 468}
 469
 470static noinline_for_stack void
 471nfs_local_iters_init(struct nfs_local_kiocb *iocb, int rw)
 472{
 473	struct nfs_pgio_header *hdr = iocb->hdr;
 474	struct page **pagevec = hdr->page_array.pagevec;
 475	unsigned long v, total;
 476	unsigned int base;
 477	size_t len;
 478
 479	v = 0;
 480	total = hdr->args.count;
 481	base = hdr->args.pgbase;
 482	pagevec += base >> PAGE_SHIFT;
 483	base &= ~PAGE_MASK;
 484	while (total && v < hdr->page_array.npages) {
 485		len = min_t(size_t, total, PAGE_SIZE - base);
 486		bvec_set_page(&iocb->bvec[v], *pagevec, len, base);
 487		total -= len;
 488		++pagevec;
 489		++v;
 490		base = 0;
 491	}
 492	len = hdr->args.count - total;
 493
 494	/*
 495	 * For each iocb, iocb->n_iters is always at least 1 and we always
 496	 * end io after first nfs_local_pgio_done call unless misaligned DIO.
 497	 */
 498	atomic_set(&iocb->n_iters, 1);
 499
 500	if (test_bit(NFS_IOHDR_ODIRECT, &hdr->flags)) {
 501		struct nfs_local_dio local_dio;
 502
 503		if (nfs_is_local_dio_possible(iocb, rw, len, &local_dio) &&
 504		    nfs_local_iters_setup_dio(iocb, rw, v, len, &local_dio) != 0) {
 505			/* Ensure DIO WRITE's IO on stable storage upon completion */
 506			if (rw == ITER_SOURCE)
 507				iocb->kiocb.ki_flags |= IOCB_DSYNC|IOCB_SYNC;
 508			return; /* is DIO-aligned */
 509		}
 510	}
 511
 512	/* Use buffered IO */
 513	iov_iter_bvec(&iocb->iters[0], rw, iocb->bvec, v, len);
 514}
 515
 516static void
 517nfs_local_hdr_release(struct nfs_pgio_header *hdr,
 518		const struct rpc_call_ops *call_ops)
 519{
 520	call_ops->rpc_call_done(&hdr->task, hdr);
 521	call_ops->rpc_release(hdr);
 522}
 523
 524static void
 525nfs_local_pgio_init(struct nfs_pgio_header *hdr,
 526		const struct rpc_call_ops *call_ops)
 527{
 528	hdr->task.tk_ops = call_ops;
 529	if (!hdr->task.tk_start)
 530		hdr->task.tk_start = ktime_get();
 531}
 532
 533static bool
 534nfs_local_pgio_done(struct nfs_local_kiocb *iocb, long status, bool force)
 535{
 536	struct nfs_pgio_header *hdr = iocb->hdr;
 537
 538	/* Must handle partial completions */
 539	if (status >= 0) {
 540		hdr->res.count += status;
 541		/* @hdr was initialized to 0 (zeroed during allocation) */
 542		if (hdr->task.tk_status == 0)
 543			hdr->res.op_status = NFS4_OK;
 544	} else {
 545		hdr->res.op_status = nfs_localio_errno_to_nfs4_stat(status);
 546		hdr->task.tk_status = status;
 547	}
 548
 549	if (force)
 550		return true;
 551
 552	BUG_ON(atomic_read(&iocb->n_iters) <= 0);
 553	return atomic_dec_and_test(&iocb->n_iters);
 554}
 555
 556static void
 557nfs_local_iocb_release(struct nfs_local_kiocb *iocb)
 558{
 559	nfs_local_file_put(iocb->localio);
 560	nfs_local_iocb_free(iocb);
 561}
 562
 563static void nfs_local_pgio_restart(struct nfs_local_kiocb *iocb,
 564				   struct nfs_pgio_header *hdr)
 565{
 566	int status = 0;
 567
 568	iocb->kiocb.ki_pos = hdr->args.offset;
 569	iocb->kiocb.ki_flags &= ~(IOCB_DSYNC | IOCB_SYNC | IOCB_DIRECT);
 570	iocb->kiocb.ki_complete = NULL;
 571	iocb->aio_complete_work = NULL;
 572	iocb->end_iter_index = -1;
 573
 574	switch (hdr->rw_mode) {
 575	case FMODE_READ:
 576		nfs_local_iters_init(iocb, ITER_DEST);
 577		status = nfs_local_do_read(iocb, hdr->task.tk_ops);
 578		break;
 579	case FMODE_WRITE:
 580		nfs_local_iters_init(iocb, ITER_SOURCE);
 581		status = nfs_local_do_write(iocb, hdr->task.tk_ops);
 582		break;
 583	default:
 584		status = -EOPNOTSUPP;
 585	}
 586
 587	if (status != 0) {
 588		nfs_local_iocb_release(iocb);
 589		hdr->task.tk_status = status;
 590		nfs_local_hdr_release(hdr, hdr->task.tk_ops);
 591	}
 592}
 593
 594static void nfs_local_pgio_release(struct nfs_local_kiocb *iocb)
 595{
 596	struct nfs_pgio_header *hdr = iocb->hdr;
 597	struct rpc_task *task = &hdr->task;
 598
 599	task->tk_action = NULL;
 600	task->tk_ops->rpc_call_done(task, hdr);
 601
 602	if (task->tk_action == NULL) {
 603		nfs_local_iocb_release(iocb);
 604		task->tk_ops->rpc_release(hdr);
 605	} else
 606		nfs_local_pgio_restart(iocb, hdr);
 607}
 608
 609/*
 610 * Complete the I/O from iocb->kiocb.ki_complete()
 611 *
 612 * Note that this function can be called from a bottom half context,
 613 * hence we need to queue the rpc_call_done() etc to a workqueue
 614 */
 615static inline void nfs_local_pgio_aio_complete(struct nfs_local_kiocb *iocb)
 616{
 617	INIT_WORK(&iocb->work, iocb->aio_complete_work);
 618	queue_work(nfsiod_workqueue, &iocb->work);
 619}
 620
 621static void nfs_local_read_done(struct nfs_local_kiocb *iocb)
 622{
 623	struct nfs_pgio_header *hdr = iocb->hdr;
 624	struct file *filp = iocb->kiocb.ki_filp;
 625	long status = hdr->task.tk_status;
 626
 627	if ((iocb->kiocb.ki_flags & IOCB_DIRECT) && status == -EINVAL) {
 628		/* Underlying FS will return -EINVAL if misaligned DIO is attempted. */
 629		pr_info_ratelimited("nfs: Unexpected direct I/O read alignment failure\n");
 630	}
 631
 632	/*
 633	 * Must clear replen otherwise NFSv3 data corruption will occur
 634	 * if/when switching from LOCALIO back to using normal RPC.
 635	 */
 636	hdr->res.replen = 0;
 637
 638	/* nfs_readpage_result() handles short read */
 639
 640	if (hdr->args.offset + hdr->res.count >= i_size_read(file_inode(filp)))
 641		hdr->res.eof = true;
 642
 643	dprintk("%s: read %ld bytes eof %d.\n", __func__,
 644			status > 0 ? status : 0, hdr->res.eof);
 645}
 646
 647static inline void nfs_local_read_iocb_done(struct nfs_local_kiocb *iocb)
 648{
 649	nfs_local_read_done(iocb);
 650	nfs_local_pgio_release(iocb);
 651}
 652
 653static void nfs_local_read_aio_complete_work(struct work_struct *work)
 654{
 655	struct nfs_local_kiocb *iocb =
 656		container_of(work, struct nfs_local_kiocb, work);
 657
 658	nfs_local_read_iocb_done(iocb);
 659}
 660
 661static void nfs_local_read_aio_complete(struct kiocb *kiocb, long ret)
 662{
 663	struct nfs_local_kiocb *iocb =
 664		container_of(kiocb, struct nfs_local_kiocb, kiocb);
 665
 666	/* AIO completion of DIO read should always be last to complete */
 667	if (unlikely(!nfs_local_pgio_done(iocb, ret, false)))
 668		return;
 669
 670	nfs_local_pgio_aio_complete(iocb); /* Calls nfs_local_read_aio_complete_work */
 671}
 672
 673static void nfs_local_call_read(struct work_struct *work)
 674{
 675	struct nfs_local_kiocb *iocb =
 676		container_of(work, struct nfs_local_kiocb, work);
 677	struct file *filp = iocb->kiocb.ki_filp;
 678	ssize_t status;
 679	int n_iters;
 680
 681	n_iters = atomic_read(&iocb->n_iters);
 682	for (int i = 0; i < n_iters ; i++) {
 683		if (iocb->iter_is_dio_aligned[i]) {
 684			iocb->kiocb.ki_flags |= IOCB_DIRECT;
 685			/* Only use AIO completion if DIO-aligned segment is last */
 686			if (i == iocb->end_iter_index) {
 687				iocb->kiocb.ki_complete = nfs_local_read_aio_complete;
 688				iocb->aio_complete_work = nfs_local_read_aio_complete_work;
 689			}
 690		} else
 691			iocb->kiocb.ki_flags &= ~IOCB_DIRECT;
 692
 693		scoped_with_creds(filp->f_cred)
 694			status = filp->f_op->read_iter(&iocb->kiocb, &iocb->iters[i]);
 695
 696		if (status == -EIOCBQUEUED)
 697			continue;
 698		/* Break on completion, errors, or short reads */
 699		if (nfs_local_pgio_done(iocb, status, false) || status < 0 ||
 700		    (size_t)status < iov_iter_count(&iocb->iters[i])) {
 701			nfs_local_read_iocb_done(iocb);
 702			break;
 703		}
 704	}
 705}
 706
 707static int
 708nfs_local_do_read(struct nfs_local_kiocb *iocb,
 709		  const struct rpc_call_ops *call_ops)
 710{
 711	struct nfs_pgio_header *hdr = iocb->hdr;
 712
 713	dprintk("%s: vfs_read count=%u pos=%llu\n",
 714		__func__, hdr->args.count, hdr->args.offset);
 715
 716	nfs_local_pgio_init(hdr, call_ops);
 717	hdr->res.eof = false;
 718
 719	INIT_WORK(&iocb->work, nfs_local_call_read);
 720	queue_work(nfslocaliod_workqueue, &iocb->work);
 721
 722	return 0;
 723}
 724
 725static void
 726nfs_copy_boot_verifier(struct nfs_write_verifier *verifier, struct inode *inode)
 727{
 728	struct nfs_client *clp = NFS_SERVER(inode)->nfs_client;
 729	u32 *verf = (u32 *)verifier->data;
 730	unsigned int seq;
 731
 732	do {
 733		seq = read_seqbegin(&clp->cl_boot_lock);
 734		verf[0] = (u32)clp->cl_nfssvc_boot.tv_sec;
 735		verf[1] = (u32)clp->cl_nfssvc_boot.tv_nsec;
 736	} while (read_seqretry(&clp->cl_boot_lock, seq));
 737}
 738
 739static void
 740nfs_reset_boot_verifier(struct inode *inode)
 741{
 742	struct nfs_client *clp = NFS_SERVER(inode)->nfs_client;
 743
 744	write_seqlock(&clp->cl_boot_lock);
 745	ktime_get_real_ts64(&clp->cl_nfssvc_boot);
 746	write_sequnlock(&clp->cl_boot_lock);
 747}
 748
 749static void
 750nfs_set_local_verifier(struct inode *inode,
 751		struct nfs_writeverf *verf,
 752		enum nfs3_stable_how how)
 753{
 754	nfs_copy_boot_verifier(&verf->verifier, inode);
 755	verf->committed = how;
 756}
 757
 758/* Factored out from fs/nfsd/vfs.h:fh_getattr() */
 759static int __vfs_getattr(const struct path *p, struct kstat *stat, int version)
 760{
 761	u32 request_mask = STATX_BASIC_STATS;
 762
 763	if (version == 4)
 764		request_mask |= (STATX_BTIME | STATX_CHANGE_COOKIE);
 765	return vfs_getattr(p, stat, request_mask, AT_STATX_SYNC_AS_STAT);
 766}
 767
 768/* Copied from fs/nfsd/nfsfh.c:nfsd4_change_attribute() */
 769static u64 __nfsd4_change_attribute(const struct kstat *stat,
 770				    const struct inode *inode)
 771{
 772	u64 chattr;
 773
 774	if (stat->result_mask & STATX_CHANGE_COOKIE) {
 775		chattr = stat->change_cookie;
 776		if (S_ISREG(inode->i_mode) &&
 777		    !(stat->attributes & STATX_ATTR_CHANGE_MONOTONIC)) {
 778			chattr += (u64)stat->ctime.tv_sec << 30;
 779			chattr += stat->ctime.tv_nsec;
 780		}
 781	} else {
 782		chattr = time_to_chattr(&stat->ctime);
 783	}
 784	return chattr;
 785}
 786
 787static void nfs_local_vfs_getattr(struct nfs_local_kiocb *iocb)
 788{
 789	struct kstat stat;
 790	struct file *filp = iocb->kiocb.ki_filp;
 791	struct nfs_pgio_header *hdr = iocb->hdr;
 792	struct nfs_fattr *fattr = hdr->res.fattr;
 793	int version = NFS_PROTO(hdr->inode)->version;
 794
 795	if (unlikely(!fattr) || __vfs_getattr(&filp->f_path, &stat, version))
 796		return;
 797
 798	fattr->valid = (NFS_ATTR_FATTR_FILEID |
 799			NFS_ATTR_FATTR_CHANGE |
 800			NFS_ATTR_FATTR_SIZE |
 801			NFS_ATTR_FATTR_ATIME |
 802			NFS_ATTR_FATTR_MTIME |
 803			NFS_ATTR_FATTR_CTIME |
 804			NFS_ATTR_FATTR_SPACE_USED);
 805
 806	fattr->fileid = stat.ino;
 807	fattr->size = stat.size;
 808	fattr->atime = stat.atime;
 809	fattr->mtime = stat.mtime;
 810	fattr->ctime = stat.ctime;
 811	if (version == 4) {
 812		fattr->change_attr =
 813			__nfsd4_change_attribute(&stat, file_inode(filp));
 814	} else
 815		fattr->change_attr = nfs_timespec_to_change_attr(&fattr->ctime);
 816	fattr->du.nfs3.used = stat.blocks << 9;
 817}
 818
 819static void nfs_local_write_done(struct nfs_local_kiocb *iocb)
 820{
 821	struct nfs_pgio_header *hdr = iocb->hdr;
 822	long status = hdr->task.tk_status;
 823
 824	dprintk("%s: wrote %ld bytes.\n", __func__, status > 0 ? status : 0);
 825
 826	if ((iocb->kiocb.ki_flags & IOCB_DIRECT) && status == -EINVAL) {
 827		/* Underlying FS will return -EINVAL if misaligned DIO is attempted. */
 828		pr_info_ratelimited("nfs: Unexpected direct I/O write alignment failure\n");
 829	}
 830
 831	if (status < 0)
 832		nfs_reset_boot_verifier(hdr->inode);
 833}
 834
 835static inline void nfs_local_write_iocb_done(struct nfs_local_kiocb *iocb)
 836{
 837	nfs_local_write_done(iocb);
 838	nfs_local_vfs_getattr(iocb);
 839	nfs_local_pgio_release(iocb);
 840}
 841
 842static void nfs_local_write_aio_complete_work(struct work_struct *work)
 843{
 844	struct nfs_local_kiocb *iocb =
 845		container_of(work, struct nfs_local_kiocb, work);
 846
 847	nfs_local_write_iocb_done(iocb);
 848}
 849
 850static void nfs_local_write_aio_complete(struct kiocb *kiocb, long ret)
 851{
 852	struct nfs_local_kiocb *iocb =
 853		container_of(kiocb, struct nfs_local_kiocb, kiocb);
 854
 855	/* AIO completion of DIO write should always be last to complete */
 856	if (unlikely(!nfs_local_pgio_done(iocb, ret, false)))
 857		return;
 858
 859	nfs_local_pgio_aio_complete(iocb); /* Calls nfs_local_write_aio_complete_work */
 860}
 861
 862static void nfs_local_call_write(struct work_struct *work)
 863{
 864	struct nfs_local_kiocb *iocb =
 865		container_of(work, struct nfs_local_kiocb, work);
 866	struct file *filp = iocb->kiocb.ki_filp;
 867	unsigned long old_flags = current->flags;
 868	ssize_t status;
 869	int n_iters;
 870
 871	current->flags |= PF_LOCAL_THROTTLE | PF_MEMALLOC_NOIO;
 872
 873	file_start_write(filp);
 874	n_iters = atomic_read(&iocb->n_iters);
 875	for (int i = 0; i < n_iters ; i++) {
 876		if (iocb->iter_is_dio_aligned[i]) {
 877			iocb->kiocb.ki_flags |= IOCB_DIRECT;
 878			/* Only use AIO completion if DIO-aligned segment is last */
 879			if (i == iocb->end_iter_index) {
 880				iocb->kiocb.ki_complete = nfs_local_write_aio_complete;
 881				iocb->aio_complete_work = nfs_local_write_aio_complete_work;
 882			}
 883		} else
 884			iocb->kiocb.ki_flags &= ~IOCB_DIRECT;
 885
 886		scoped_with_creds(filp->f_cred)
 887			status = filp->f_op->write_iter(&iocb->kiocb, &iocb->iters[i]);
 888
 889		if (status == -EIOCBQUEUED)
 890			continue;
 891		/* Break on completion, errors, or short writes */
 892		if (nfs_local_pgio_done(iocb, status, false) || status < 0 ||
 893		    (size_t)status < iov_iter_count(&iocb->iters[i])) {
 894			nfs_local_write_iocb_done(iocb);
 895			break;
 896		}
 897	}
 898	file_end_write(filp);
 899
 900	current->flags = old_flags;
 901}
 902
 903static int
 904nfs_local_do_write(struct nfs_local_kiocb *iocb,
 905		   const struct rpc_call_ops *call_ops)
 906{
 907	struct nfs_pgio_header *hdr = iocb->hdr;
 908
 909	dprintk("%s: vfs_write count=%u pos=%llu %s\n",
 910		__func__, hdr->args.count, hdr->args.offset,
 911		(hdr->args.stable == NFS_UNSTABLE) ?  "unstable" : "stable");
 912
 913	switch (hdr->args.stable) {
 914	default:
 915		break;
 916	case NFS_DATA_SYNC:
 917		iocb->kiocb.ki_flags |= IOCB_DSYNC;
 918		break;
 919	case NFS_FILE_SYNC:
 920		iocb->kiocb.ki_flags |= IOCB_DSYNC|IOCB_SYNC;
 921	}
 922
 923	nfs_local_pgio_init(hdr, call_ops);
 924
 925	nfs_set_local_verifier(hdr->inode, hdr->res.verf, hdr->args.stable);
 926
 927	INIT_WORK(&iocb->work, nfs_local_call_write);
 928	queue_work(nfslocaliod_workqueue, &iocb->work);
 929
 930	return 0;
 931}
 932
 933static struct nfs_local_kiocb *
 934nfs_local_iocb_init(struct nfs_pgio_header *hdr, struct nfsd_file *localio)
 935{
 936	struct file *file = nfs_to->nfsd_file_file(localio);
 937	struct nfs_local_kiocb *iocb;
 938	gfp_t gfp_mask;
 939	int rw;
 940
 941	if (hdr->rw_mode & FMODE_READ) {
 942		if (!file->f_op->read_iter)
 943			return ERR_PTR(-EOPNOTSUPP);
 944		gfp_mask = GFP_KERNEL;
 945		rw = ITER_DEST;
 946	} else {
 947		if (!file->f_op->write_iter)
 948			return ERR_PTR(-EOPNOTSUPP);
 949		gfp_mask = GFP_NOIO;
 950		rw = ITER_SOURCE;
 951	}
 952
 953	iocb = nfs_local_iocb_alloc(hdr, file, gfp_mask);
 954	if (iocb == NULL)
 955		return ERR_PTR(-ENOMEM);
 956	iocb->hdr = hdr;
 957	iocb->localio = localio;
 958
 959	nfs_local_iters_init(iocb, rw);
 960
 961	return iocb;
 962}
 963
 964int nfs_local_doio(struct nfs_client *clp, struct nfsd_file *localio,
 965		   struct nfs_pgio_header *hdr,
 966		   const struct rpc_call_ops *call_ops)
 967{
 968	struct nfs_local_kiocb *iocb;
 969	int status = 0;
 970
 971	if (!hdr->args.count)
 972		return 0;
 973
 974	iocb = nfs_local_iocb_init(hdr, localio);
 975	if (IS_ERR(iocb))
 976		return PTR_ERR(iocb);
 977
 978	switch (hdr->rw_mode) {
 979	case FMODE_READ:
 980		status = nfs_local_do_read(iocb, call_ops);
 981		break;
 982	case FMODE_WRITE:
 983		status = nfs_local_do_write(iocb, call_ops);
 984		break;
 985	default:
 986		dprintk("%s: invalid mode: %d\n", __func__,
 987			hdr->rw_mode);
 988		status = -EOPNOTSUPP;
 989	}
 990
 991	if (status != 0) {
 992		nfs_local_iocb_release(iocb);
 993		hdr->task.tk_status = status;
 994		nfs_local_hdr_release(hdr, call_ops);
 995	}
 996	return status;
 997}
 998
 999static void
1000nfs_local_init_commit(struct nfs_commit_data *data,
1001		const struct rpc_call_ops *call_ops)
1002{
1003	data->task.tk_ops = call_ops;
1004}
1005
1006static int
1007nfs_local_run_commit(struct file *filp, struct nfs_commit_data *data)
1008{
1009	loff_t start = data->args.offset;
1010	loff_t end = LLONG_MAX;
1011
1012	if (data->args.count > 0) {
1013		end = start + data->args.count - 1;
1014		if (end < start)
1015			end = LLONG_MAX;
1016	}
1017
1018	nfs_local_mapping_set_gfp_nofs_context(filp->f_mapping);
1019
1020	dprintk("%s: commit %llu - %llu\n", __func__, start, end);
1021	return vfs_fsync_range(filp, start, end, 0);
1022}
1023
1024static void
1025nfs_local_commit_done(struct nfs_commit_data *data, int status)
1026{
1027	if (status >= 0) {
1028		nfs_set_local_verifier(data->inode,
1029				data->res.verf,
1030				NFS_FILE_SYNC);
1031		data->res.op_status = NFS4_OK;
1032		data->task.tk_status = 0;
1033	} else {
1034		nfs_reset_boot_verifier(data->inode);
1035		data->res.op_status = nfs_localio_errno_to_nfs4_stat(status);
1036		data->task.tk_status = status;
1037	}
1038}
1039
1040static void
1041nfs_local_release_commit_data(struct nfsd_file *localio,
1042		struct nfs_commit_data *data,
1043		const struct rpc_call_ops *call_ops)
1044{
1045	nfs_local_file_put(localio);
1046	call_ops->rpc_call_done(&data->task, data);
1047	call_ops->rpc_release(data);
1048}
1049
1050static void
1051nfs_local_fsync_ctx_free(struct nfs_local_fsync_ctx *ctx)
1052{
1053	nfs_local_release_commit_data(ctx->localio, ctx->data,
1054				      ctx->data->task.tk_ops);
1055	kfree(ctx);
1056}
1057
1058static void
1059nfs_local_fsync_work(struct work_struct *work)
1060{
1061	unsigned long old_flags = current->flags;
1062	struct nfs_local_fsync_ctx *ctx;
1063	int status;
1064
1065	ctx = container_of(work, struct nfs_local_fsync_ctx, work);
1066
1067	current->flags |= PF_LOCAL_THROTTLE | PF_MEMALLOC_NOIO;
1068
1069	status = nfs_local_run_commit(nfs_to->nfsd_file_file(ctx->localio),
1070				      ctx->data);
1071	nfs_local_commit_done(ctx->data, status);
1072	if (ctx->done != NULL)
1073		complete(ctx->done);
1074	nfs_local_fsync_ctx_free(ctx);
1075
1076	current->flags = old_flags;
1077}
1078
1079static struct nfs_local_fsync_ctx *
1080nfs_local_fsync_ctx_alloc(struct nfs_commit_data *data,
1081			  struct nfsd_file *localio, gfp_t flags)
1082{
1083	struct nfs_local_fsync_ctx *ctx = kmalloc(sizeof(*ctx), flags);
1084
1085	if (ctx != NULL) {
1086		ctx->localio = localio;
1087		ctx->data = data;
1088		INIT_WORK(&ctx->work, nfs_local_fsync_work);
1089		ctx->done = NULL;
1090	}
1091	return ctx;
1092}
1093
1094int nfs_local_commit(struct nfsd_file *localio,
1095		     struct nfs_commit_data *data,
1096		     const struct rpc_call_ops *call_ops, int how)
1097{
1098	struct nfs_local_fsync_ctx *ctx;
1099
1100	ctx = nfs_local_fsync_ctx_alloc(data, localio, GFP_NOIO);
1101	if (!ctx) {
1102		nfs_local_commit_done(data, -ENOMEM);
1103		nfs_local_release_commit_data(localio, data, call_ops);
1104		return -ENOMEM;
1105	}
1106
1107	nfs_local_init_commit(data, call_ops);
1108
1109	if (how & FLUSH_SYNC) {
1110		DECLARE_COMPLETION_ONSTACK(done);
1111		ctx->done = &done;
1112		queue_work(nfslocaliod_workqueue, &ctx->work);
1113		wait_for_completion(&done);
1114	} else
1115		queue_work(nfslocaliod_workqueue, &ctx->work);
1116
1117	return 0;
1118}