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