1 // SPDX-License-Identifier: GPL-2.0 2 #include <linux/ceph/ceph_debug.h> 3 #include <linux/ceph/striper.h> 4 5 #include <linux/module.h> 6 #include <linux/sched.h> 7 #include <linux/slab.h> 8 #include <linux/file.h> 9 #include <linux/mount.h> 10 #include <linux/namei.h> 11 #include <linux/writeback.h> 12 #include <linux/falloc.h> 13 #include <linux/iversion.h> 14 #include <linux/ktime.h> 15 #include <linux/splice.h> 16 17 #include "super.h" 18 #include "mds_client.h" 19 #include "cache.h" 20 #include "io.h" 21 #include "metric.h" 22 23 static __le32 ceph_flags_sys2wire(struct ceph_mds_client *mdsc, u32 flags) 24 { 25 struct ceph_client *cl = mdsc->fsc->client; 26 u32 wire_flags = 0; 27 28 switch (flags & O_ACCMODE) { 29 case O_RDONLY: 30 wire_flags |= CEPH_O_RDONLY; 31 break; 32 case O_WRONLY: 33 wire_flags |= CEPH_O_WRONLY; 34 break; 35 case O_RDWR: 36 wire_flags |= CEPH_O_RDWR; 37 break; 38 } 39 40 flags &= ~O_ACCMODE; 41 42 #define ceph_sys2wire(a) if (flags & a) { wire_flags |= CEPH_##a; flags &= ~a; } 43 44 ceph_sys2wire(O_CREAT); 45 ceph_sys2wire(O_EXCL); 46 ceph_sys2wire(O_TRUNC); 47 ceph_sys2wire(O_DIRECTORY); 48 ceph_sys2wire(O_NOFOLLOW); 49 50 #undef ceph_sys2wire 51 52 if (flags) 53 doutc(cl, "unused open flags: %x\n", flags); 54 55 return cpu_to_le32(wire_flags); 56 } 57 58 /* 59 * Ceph file operations 60 * 61 * Implement basic open/close functionality, and implement 62 * read/write. 63 * 64 * We implement three modes of file I/O: 65 * - buffered uses the generic_file_aio_{read,write} helpers 66 * 67 * - synchronous is used when there is multi-client read/write 68 * sharing, avoids the page cache, and synchronously waits for an 69 * ack from the OSD. 70 * 71 * - direct io takes the variant of the sync path that references 72 * user pages directly. 73 * 74 * fsync() flushes and waits on dirty pages, but just queues metadata 75 * for writeback: since the MDS can recover size and mtime there is no 76 * need to wait for MDS acknowledgement. 77 */ 78 79 /* 80 * How many pages to get in one call to iov_iter_get_pages(). This 81 * determines the size of the on-stack array used as a buffer. 82 */ 83 #define ITER_GET_BVECS_PAGES 64 84 85 static ssize_t __iter_get_bvecs(struct iov_iter *iter, size_t maxsize, 86 struct bio_vec *bvecs) 87 { 88 size_t size = 0; 89 int bvec_idx = 0; 90 91 if (maxsize > iov_iter_count(iter)) 92 maxsize = iov_iter_count(iter); 93 94 while (size < maxsize) { 95 struct page *pages[ITER_GET_BVECS_PAGES]; 96 ssize_t bytes; 97 size_t start; 98 int idx = 0; 99 100 bytes = iov_iter_get_pages2(iter, pages, maxsize - size, 101 ITER_GET_BVECS_PAGES, &start); 102 if (bytes < 0) 103 return size ?: bytes; 104 105 size += bytes; 106 107 for ( ; bytes; idx++, bvec_idx++) { 108 int len = min_t(int, bytes, PAGE_SIZE - start); 109 110 bvec_set_page(&bvecs[bvec_idx], pages[idx], len, start); 111 bytes -= len; 112 start = 0; 113 } 114 } 115 116 return size; 117 } 118 119 /* 120 * iov_iter_get_pages() only considers one iov_iter segment, no matter 121 * what maxsize or maxpages are given. For ITER_BVEC that is a single 122 * page. 123 * 124 * Attempt to get up to @maxsize bytes worth of pages from @iter. 125 * Return the number of bytes in the created bio_vec array, or an error. 126 */ 127 static ssize_t iter_get_bvecs_alloc(struct iov_iter *iter, size_t maxsize, 128 struct bio_vec **bvecs, int *num_bvecs) 129 { 130 struct bio_vec *bv; 131 size_t orig_count = iov_iter_count(iter); 132 ssize_t bytes; 133 int npages; 134 135 iov_iter_truncate(iter, maxsize); 136 npages = iov_iter_npages(iter, INT_MAX); 137 iov_iter_reexpand(iter, orig_count); 138 139 /* 140 * __iter_get_bvecs() may populate only part of the array -- zero it 141 * out. 142 */ 143 bv = kvmalloc_array(npages, sizeof(*bv), GFP_KERNEL | __GFP_ZERO); 144 if (!bv) 145 return -ENOMEM; 146 147 bytes = __iter_get_bvecs(iter, maxsize, bv); 148 if (bytes < 0) { 149 /* 150 * No pages were pinned -- just free the array. 151 */ 152 kvfree(bv); 153 return bytes; 154 } 155 156 *bvecs = bv; 157 *num_bvecs = npages; 158 return bytes; 159 } 160 161 static void put_bvecs(struct bio_vec *bvecs, int num_bvecs, bool should_dirty) 162 { 163 int i; 164 165 for (i = 0; i < num_bvecs; i++) { 166 if (bvecs[i].bv_page) { 167 if (should_dirty) 168 set_page_dirty_lock(bvecs[i].bv_page); 169 put_page(bvecs[i].bv_page); 170 } 171 } 172 kvfree(bvecs); 173 } 174 175 /* 176 * Prepare an open request. Preallocate ceph_cap to avoid an 177 * inopportune ENOMEM later. 178 */ 179 static struct ceph_mds_request * 180 prepare_open_request(struct super_block *sb, int flags, int create_mode) 181 { 182 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(sb); 183 struct ceph_mds_request *req; 184 int want_auth = USE_ANY_MDS; 185 int op = (flags & O_CREAT) ? CEPH_MDS_OP_CREATE : CEPH_MDS_OP_OPEN; 186 187 if (flags & (O_WRONLY|O_RDWR|O_CREAT|O_TRUNC)) 188 want_auth = USE_AUTH_MDS; 189 190 req = ceph_mdsc_create_request(mdsc, op, want_auth); 191 if (IS_ERR(req)) 192 goto out; 193 req->r_fmode = ceph_flags_to_mode(flags); 194 req->r_args.open.flags = ceph_flags_sys2wire(mdsc, flags); 195 req->r_args.open.mode = cpu_to_le32(create_mode); 196 out: 197 return req; 198 } 199 200 static int ceph_init_file_info(struct inode *inode, struct file *file, 201 int fmode, bool isdir) 202 { 203 struct ceph_inode_info *ci = ceph_inode(inode); 204 struct ceph_mount_options *opt = 205 ceph_inode_to_fs_client(&ci->netfs.inode)->mount_options; 206 struct ceph_client *cl = ceph_inode_to_client(inode); 207 struct ceph_file_info *fi; 208 int ret; 209 210 doutc(cl, "%p %llx.%llx %p 0%o (%s)\n", inode, ceph_vinop(inode), 211 file, inode->i_mode, isdir ? "dir" : "regular"); 212 BUG_ON(inode->i_fop->release != ceph_release); 213 214 if (isdir) { 215 struct ceph_dir_file_info *dfi = 216 kmem_cache_zalloc(ceph_dir_file_cachep, GFP_KERNEL); 217 if (!dfi) 218 return -ENOMEM; 219 220 file->private_data = dfi; 221 fi = &dfi->file_info; 222 dfi->next_offset = 2; 223 dfi->readdir_cache_idx = -1; 224 } else { 225 fi = kmem_cache_zalloc(ceph_file_cachep, GFP_KERNEL); 226 if (!fi) 227 return -ENOMEM; 228 229 if (opt->flags & CEPH_MOUNT_OPT_NOPAGECACHE) 230 fi->flags |= CEPH_F_SYNC; 231 232 file->private_data = fi; 233 } 234 235 ceph_get_fmode(ci, fmode, 1); 236 fi->fmode = fmode; 237 238 spin_lock_init(&fi->rw_contexts_lock); 239 INIT_LIST_HEAD(&fi->rw_contexts); 240 fi->filp_gen = READ_ONCE(ceph_inode_to_fs_client(inode)->filp_gen); 241 242 if ((file->f_mode & FMODE_WRITE) && ceph_has_inline_data(ci)) { 243 ret = ceph_uninline_data(file); 244 if (ret < 0) 245 goto error; 246 } 247 248 return 0; 249 250 error: 251 ceph_fscache_unuse_cookie(inode, file->f_mode & FMODE_WRITE); 252 ceph_put_fmode(ci, fi->fmode, 1); 253 kmem_cache_free(ceph_file_cachep, fi); 254 /* wake up anyone waiting for caps on this inode */ 255 wake_up_all(&ci->i_cap_wq); 256 return ret; 257 } 258 259 /* 260 * initialize private struct file data. 261 * if we fail, clean up by dropping fmode reference on the ceph_inode 262 */ 263 static int ceph_init_file(struct inode *inode, struct file *file, int fmode) 264 { 265 struct ceph_client *cl = ceph_inode_to_client(inode); 266 int ret = 0; 267 268 switch (inode->i_mode & S_IFMT) { 269 case S_IFREG: 270 ceph_fscache_use_cookie(inode, file->f_mode & FMODE_WRITE); 271 fallthrough; 272 case S_IFDIR: 273 ret = ceph_init_file_info(inode, file, fmode, 274 S_ISDIR(inode->i_mode)); 275 break; 276 277 case S_IFLNK: 278 doutc(cl, "%p %llx.%llx %p 0%o (symlink)\n", inode, 279 ceph_vinop(inode), file, inode->i_mode); 280 break; 281 282 default: 283 doutc(cl, "%p %llx.%llx %p 0%o (special)\n", inode, 284 ceph_vinop(inode), file, inode->i_mode); 285 /* 286 * we need to drop the open ref now, since we don't 287 * have .release set to ceph_release. 288 */ 289 BUG_ON(inode->i_fop->release == ceph_release); 290 291 /* call the proper open fop */ 292 ret = inode->i_fop->open(inode, file); 293 } 294 return ret; 295 } 296 297 /* 298 * try renew caps after session gets killed. 299 */ 300 int ceph_renew_caps(struct inode *inode, int fmode) 301 { 302 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb); 303 struct ceph_client *cl = mdsc->fsc->client; 304 struct ceph_inode_info *ci = ceph_inode(inode); 305 struct ceph_mds_request *req; 306 int err, flags, wanted; 307 308 spin_lock(&ci->i_ceph_lock); 309 __ceph_touch_fmode(ci, mdsc, fmode); 310 wanted = __ceph_caps_file_wanted(ci); 311 if (__ceph_is_any_real_caps(ci) && 312 (!(wanted & CEPH_CAP_ANY_WR) || ci->i_auth_cap)) { 313 int issued = __ceph_caps_issued(ci, NULL); 314 spin_unlock(&ci->i_ceph_lock); 315 doutc(cl, "%p %llx.%llx want %s issued %s updating mds_wanted\n", 316 inode, ceph_vinop(inode), ceph_cap_string(wanted), 317 ceph_cap_string(issued)); 318 ceph_check_caps(ci, 0); 319 return 0; 320 } 321 spin_unlock(&ci->i_ceph_lock); 322 323 flags = 0; 324 if ((wanted & CEPH_CAP_FILE_RD) && (wanted & CEPH_CAP_FILE_WR)) 325 flags = O_RDWR; 326 else if (wanted & CEPH_CAP_FILE_RD) 327 flags = O_RDONLY; 328 else if (wanted & CEPH_CAP_FILE_WR) 329 flags = O_WRONLY; 330 #ifdef O_LAZY 331 if (wanted & CEPH_CAP_FILE_LAZYIO) 332 flags |= O_LAZY; 333 #endif 334 335 req = prepare_open_request(inode->i_sb, flags, 0); 336 if (IS_ERR(req)) { 337 err = PTR_ERR(req); 338 goto out; 339 } 340 341 req->r_inode = inode; 342 ihold(inode); 343 req->r_num_caps = 1; 344 345 err = ceph_mdsc_do_request(mdsc, NULL, req); 346 ceph_mdsc_put_request(req); 347 out: 348 doutc(cl, "%p %llx.%llx open result=%d\n", inode, ceph_vinop(inode), 349 err); 350 return err < 0 ? err : 0; 351 } 352 353 /* 354 * If we already have the requisite capabilities, we can satisfy 355 * the open request locally (no need to request new caps from the 356 * MDS). We do, however, need to inform the MDS (asynchronously) 357 * if our wanted caps set expands. 358 */ 359 int ceph_open(struct inode *inode, struct file *file) 360 { 361 struct ceph_inode_info *ci = ceph_inode(inode); 362 struct ceph_fs_client *fsc = ceph_sb_to_fs_client(inode->i_sb); 363 struct ceph_client *cl = fsc->client; 364 struct ceph_mds_client *mdsc = fsc->mdsc; 365 struct ceph_mds_request *req; 366 struct ceph_file_info *fi = file->private_data; 367 int err; 368 int flags, fmode, wanted; 369 struct dentry *dentry; 370 char *path; 371 int pathlen; 372 u64 pathbase; 373 bool do_sync = false; 374 int mask = MAY_READ; 375 376 if (fi) { 377 doutc(cl, "file %p is already opened\n", file); 378 return 0; 379 } 380 381 /* filter out O_CREAT|O_EXCL; vfs did that already. yuck. */ 382 flags = file->f_flags & ~(O_CREAT|O_EXCL); 383 if (S_ISDIR(inode->i_mode)) { 384 flags = O_DIRECTORY; /* mds likes to know */ 385 } else if (S_ISREG(inode->i_mode)) { 386 err = fscrypt_file_open(inode, file); 387 if (err) 388 return err; 389 } 390 391 doutc(cl, "%p %llx.%llx file %p flags %d (%d)\n", inode, 392 ceph_vinop(inode), file, flags, file->f_flags); 393 fmode = ceph_flags_to_mode(flags); 394 wanted = ceph_caps_for_mode(fmode); 395 396 if (fmode & CEPH_FILE_MODE_WR) 397 mask |= MAY_WRITE; 398 dentry = d_find_alias(inode); 399 if (!dentry) { 400 do_sync = true; 401 } else { 402 path = ceph_mdsc_build_path(mdsc, dentry, &pathlen, &pathbase, 0); 403 if (IS_ERR(path)) { 404 do_sync = true; 405 err = 0; 406 } else { 407 err = ceph_mds_check_access(mdsc, path, mask); 408 } 409 ceph_mdsc_free_path(path, pathlen); 410 dput(dentry); 411 412 /* For none EACCES cases will let the MDS do the mds auth check */ 413 if (err == -EACCES) { 414 return err; 415 } else if (err < 0) { 416 do_sync = true; 417 err = 0; 418 } 419 } 420 421 /* snapped files are read-only */ 422 if (ceph_snap(inode) != CEPH_NOSNAP && (file->f_mode & FMODE_WRITE)) 423 return -EROFS; 424 425 /* trivially open snapdir */ 426 if (ceph_snap(inode) == CEPH_SNAPDIR) { 427 return ceph_init_file(inode, file, fmode); 428 } 429 430 /* 431 * No need to block if we have caps on the auth MDS (for 432 * write) or any MDS (for read). Update wanted set 433 * asynchronously. 434 */ 435 spin_lock(&ci->i_ceph_lock); 436 if (!do_sync && __ceph_is_any_real_caps(ci) && 437 (((fmode & CEPH_FILE_MODE_WR) == 0) || ci->i_auth_cap)) { 438 int mds_wanted = __ceph_caps_mds_wanted(ci, true); 439 int issued = __ceph_caps_issued(ci, NULL); 440 441 doutc(cl, "open %p fmode %d want %s issued %s using existing\n", 442 inode, fmode, ceph_cap_string(wanted), 443 ceph_cap_string(issued)); 444 __ceph_touch_fmode(ci, mdsc, fmode); 445 spin_unlock(&ci->i_ceph_lock); 446 447 /* adjust wanted? */ 448 if ((issued & wanted) != wanted && 449 (mds_wanted & wanted) != wanted && 450 ceph_snap(inode) != CEPH_SNAPDIR) 451 ceph_check_caps(ci, 0); 452 453 return ceph_init_file(inode, file, fmode); 454 } else if (!do_sync && ceph_snap(inode) != CEPH_NOSNAP && 455 (ci->i_snap_caps & wanted) == wanted) { 456 __ceph_touch_fmode(ci, mdsc, fmode); 457 spin_unlock(&ci->i_ceph_lock); 458 return ceph_init_file(inode, file, fmode); 459 } 460 461 spin_unlock(&ci->i_ceph_lock); 462 463 doutc(cl, "open fmode %d wants %s\n", fmode, ceph_cap_string(wanted)); 464 req = prepare_open_request(inode->i_sb, flags, 0); 465 if (IS_ERR(req)) { 466 err = PTR_ERR(req); 467 goto out; 468 } 469 req->r_inode = inode; 470 ihold(inode); 471 472 req->r_num_caps = 1; 473 err = ceph_mdsc_do_request(mdsc, NULL, req); 474 if (!err) 475 err = ceph_init_file(inode, file, req->r_fmode); 476 ceph_mdsc_put_request(req); 477 doutc(cl, "open result=%d on %llx.%llx\n", err, ceph_vinop(inode)); 478 out: 479 return err; 480 } 481 482 /* Clone the layout from a synchronous create, if the dir now has Dc caps */ 483 static void 484 cache_file_layout(struct inode *dst, struct inode *src) 485 { 486 struct ceph_inode_info *cdst = ceph_inode(dst); 487 struct ceph_inode_info *csrc = ceph_inode(src); 488 489 spin_lock(&cdst->i_ceph_lock); 490 if ((__ceph_caps_issued(cdst, NULL) & CEPH_CAP_DIR_CREATE) && 491 !ceph_file_layout_is_valid(&cdst->i_cached_layout)) { 492 memcpy(&cdst->i_cached_layout, &csrc->i_layout, 493 sizeof(cdst->i_cached_layout)); 494 rcu_assign_pointer(cdst->i_cached_layout.pool_ns, 495 ceph_try_get_string(csrc->i_layout.pool_ns)); 496 } 497 spin_unlock(&cdst->i_ceph_lock); 498 } 499 500 /* 501 * Try to set up an async create. We need caps, a file layout, and inode number, 502 * and either a lease on the dentry or complete dir info. If any of those 503 * criteria are not satisfied, then return false and the caller can go 504 * synchronous. 505 */ 506 static int try_prep_async_create(struct inode *dir, struct dentry *dentry, 507 struct ceph_file_layout *lo, u64 *pino) 508 { 509 struct ceph_inode_info *ci = ceph_inode(dir); 510 struct ceph_dentry_info *di = ceph_dentry(dentry); 511 int got = 0, want = CEPH_CAP_FILE_EXCL | CEPH_CAP_DIR_CREATE; 512 u64 ino; 513 514 spin_lock(&ci->i_ceph_lock); 515 /* No auth cap means no chance for Dc caps */ 516 if (!ci->i_auth_cap) 517 goto no_async; 518 519 /* Any delegated inos? */ 520 if (xa_empty(&ci->i_auth_cap->session->s_delegated_inos)) 521 goto no_async; 522 523 if (!ceph_file_layout_is_valid(&ci->i_cached_layout)) 524 goto no_async; 525 526 if ((__ceph_caps_issued(ci, NULL) & want) != want) 527 goto no_async; 528 529 if (d_in_lookup(dentry)) { 530 if (!__ceph_dir_is_complete(ci)) 531 goto no_async; 532 spin_lock(&dentry->d_lock); 533 di->lease_shared_gen = atomic_read(&ci->i_shared_gen); 534 spin_unlock(&dentry->d_lock); 535 } else if (atomic_read(&ci->i_shared_gen) != 536 READ_ONCE(di->lease_shared_gen)) { 537 goto no_async; 538 } 539 540 ino = ceph_get_deleg_ino(ci->i_auth_cap->session); 541 if (!ino) 542 goto no_async; 543 544 *pino = ino; 545 ceph_take_cap_refs(ci, want, false); 546 memcpy(lo, &ci->i_cached_layout, sizeof(*lo)); 547 rcu_assign_pointer(lo->pool_ns, 548 ceph_try_get_string(ci->i_cached_layout.pool_ns)); 549 got = want; 550 no_async: 551 spin_unlock(&ci->i_ceph_lock); 552 return got; 553 } 554 555 static void restore_deleg_ino(struct inode *dir, u64 ino) 556 { 557 struct ceph_client *cl = ceph_inode_to_client(dir); 558 struct ceph_inode_info *ci = ceph_inode(dir); 559 struct ceph_mds_session *s = NULL; 560 561 spin_lock(&ci->i_ceph_lock); 562 if (ci->i_auth_cap) 563 s = ceph_get_mds_session(ci->i_auth_cap->session); 564 spin_unlock(&ci->i_ceph_lock); 565 if (s) { 566 int err = ceph_restore_deleg_ino(s, ino); 567 if (err) 568 pr_warn_client(cl, 569 "unable to restore delegated ino 0x%llx to session: %d\n", 570 ino, err); 571 ceph_put_mds_session(s); 572 } 573 } 574 575 static void wake_async_create_waiters(struct inode *inode, 576 struct ceph_mds_session *session) 577 { 578 struct ceph_inode_info *ci = ceph_inode(inode); 579 bool check_cap = false; 580 581 spin_lock(&ci->i_ceph_lock); 582 if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE) { 583 ci->i_ceph_flags &= ~CEPH_I_ASYNC_CREATE; 584 wake_up_bit(&ci->i_ceph_flags, CEPH_ASYNC_CREATE_BIT); 585 586 if (ci->i_ceph_flags & CEPH_I_ASYNC_CHECK_CAPS) { 587 ci->i_ceph_flags &= ~CEPH_I_ASYNC_CHECK_CAPS; 588 check_cap = true; 589 } 590 } 591 ceph_kick_flushing_inode_caps(session, ci); 592 spin_unlock(&ci->i_ceph_lock); 593 594 if (check_cap) 595 ceph_check_caps(ci, CHECK_CAPS_FLUSH); 596 } 597 598 static void ceph_async_create_cb(struct ceph_mds_client *mdsc, 599 struct ceph_mds_request *req) 600 { 601 struct ceph_client *cl = mdsc->fsc->client; 602 struct dentry *dentry = req->r_dentry; 603 struct inode *dinode = d_inode(dentry); 604 struct inode *tinode = req->r_target_inode; 605 int result = req->r_err ? req->r_err : 606 le32_to_cpu(req->r_reply_info.head->result); 607 608 WARN_ON_ONCE(dinode && tinode && dinode != tinode); 609 610 /* MDS changed -- caller must resubmit */ 611 if (result == -EJUKEBOX) 612 goto out; 613 614 mapping_set_error(req->r_parent->i_mapping, result); 615 616 if (result) { 617 int pathlen = 0; 618 u64 base = 0; 619 char *path = ceph_mdsc_build_path(mdsc, req->r_dentry, &pathlen, 620 &base, 0); 621 622 pr_warn_client(cl, 623 "async create failure path=(%llx)%s result=%d!\n", 624 base, IS_ERR(path) ? "<<bad>>" : path, result); 625 ceph_mdsc_free_path(path, pathlen); 626 627 ceph_dir_clear_complete(req->r_parent); 628 if (!d_unhashed(dentry)) 629 d_drop(dentry); 630 631 if (dinode) { 632 mapping_set_error(dinode->i_mapping, result); 633 ceph_inode_shutdown(dinode); 634 wake_async_create_waiters(dinode, req->r_session); 635 } 636 } 637 638 if (tinode) { 639 u64 ino = ceph_vino(tinode).ino; 640 641 if (req->r_deleg_ino != ino) 642 pr_warn_client(cl, 643 "inode number mismatch! err=%d deleg_ino=0x%llx target=0x%llx\n", 644 req->r_err, req->r_deleg_ino, ino); 645 646 mapping_set_error(tinode->i_mapping, result); 647 wake_async_create_waiters(tinode, req->r_session); 648 } else if (!result) { 649 pr_warn_client(cl, "no req->r_target_inode for 0x%llx\n", 650 req->r_deleg_ino); 651 } 652 out: 653 ceph_mdsc_release_dir_caps(req); 654 } 655 656 static int ceph_finish_async_create(struct inode *dir, struct inode *inode, 657 struct dentry *dentry, 658 struct file *file, umode_t mode, 659 struct ceph_mds_request *req, 660 struct ceph_acl_sec_ctx *as_ctx, 661 struct ceph_file_layout *lo) 662 { 663 int ret; 664 char xattr_buf[4]; 665 struct ceph_mds_reply_inode in = { }; 666 struct ceph_mds_reply_info_in iinfo = { .in = &in }; 667 struct ceph_inode_info *ci = ceph_inode(dir); 668 struct ceph_dentry_info *di = ceph_dentry(dentry); 669 struct timespec64 now; 670 struct ceph_string *pool_ns; 671 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(dir->i_sb); 672 struct ceph_client *cl = mdsc->fsc->client; 673 struct ceph_vino vino = { .ino = req->r_deleg_ino, 674 .snap = CEPH_NOSNAP }; 675 676 ktime_get_real_ts64(&now); 677 678 iinfo.inline_version = CEPH_INLINE_NONE; 679 iinfo.change_attr = 1; 680 ceph_encode_timespec64(&iinfo.btime, &now); 681 682 if (req->r_pagelist) { 683 iinfo.xattr_len = req->r_pagelist->length; 684 iinfo.xattr_data = req->r_pagelist->mapped_tail; 685 } else { 686 /* fake it */ 687 iinfo.xattr_len = ARRAY_SIZE(xattr_buf); 688 iinfo.xattr_data = xattr_buf; 689 memset(iinfo.xattr_data, 0, iinfo.xattr_len); 690 } 691 692 in.ino = cpu_to_le64(vino.ino); 693 in.snapid = cpu_to_le64(CEPH_NOSNAP); 694 in.version = cpu_to_le64(1); // ??? 695 in.cap.caps = in.cap.wanted = cpu_to_le32(CEPH_CAP_ALL_FILE); 696 in.cap.cap_id = cpu_to_le64(1); 697 in.cap.realm = cpu_to_le64(ci->i_snap_realm->ino); 698 in.cap.flags = CEPH_CAP_FLAG_AUTH; 699 in.ctime = in.mtime = in.atime = iinfo.btime; 700 in.truncate_seq = cpu_to_le32(1); 701 in.truncate_size = cpu_to_le64(-1ULL); 702 in.xattr_version = cpu_to_le64(1); 703 in.uid = cpu_to_le32(from_kuid(&init_user_ns, 704 mapped_fsuid(req->r_mnt_idmap, 705 &init_user_ns))); 706 if (dir->i_mode & S_ISGID) { 707 in.gid = cpu_to_le32(from_kgid(&init_user_ns, dir->i_gid)); 708 709 /* Directories always inherit the setgid bit. */ 710 if (S_ISDIR(mode)) 711 mode |= S_ISGID; 712 } else { 713 in.gid = cpu_to_le32(from_kgid(&init_user_ns, 714 mapped_fsgid(req->r_mnt_idmap, 715 &init_user_ns))); 716 } 717 in.mode = cpu_to_le32((u32)mode); 718 719 in.nlink = cpu_to_le32(1); 720 in.max_size = cpu_to_le64(lo->stripe_unit); 721 722 ceph_file_layout_to_legacy(lo, &in.layout); 723 /* lo is private, so pool_ns can't change */ 724 pool_ns = rcu_dereference_raw(lo->pool_ns); 725 if (pool_ns) { 726 iinfo.pool_ns_len = pool_ns->len; 727 iinfo.pool_ns_data = pool_ns->str; 728 } 729 730 down_read(&mdsc->snap_rwsem); 731 ret = ceph_fill_inode(inode, NULL, &iinfo, NULL, req->r_session, 732 req->r_fmode, NULL); 733 up_read(&mdsc->snap_rwsem); 734 if (ret) { 735 doutc(cl, "failed to fill inode: %d\n", ret); 736 ceph_dir_clear_complete(dir); 737 if (!d_unhashed(dentry)) 738 d_drop(dentry); 739 discard_new_inode(inode); 740 } else { 741 struct dentry *dn; 742 743 doutc(cl, "d_adding new inode 0x%llx to 0x%llx/%s\n", 744 vino.ino, ceph_ino(dir), dentry->d_name.name); 745 ceph_dir_clear_ordered(dir); 746 ceph_init_inode_acls(inode, as_ctx); 747 if (inode->i_state & I_NEW) { 748 /* 749 * If it's not I_NEW, then someone created this before 750 * we got here. Assume the server is aware of it at 751 * that point and don't worry about setting 752 * CEPH_I_ASYNC_CREATE. 753 */ 754 ceph_inode(inode)->i_ceph_flags = CEPH_I_ASYNC_CREATE; 755 unlock_new_inode(inode); 756 } 757 if (d_in_lookup(dentry) || d_really_is_negative(dentry)) { 758 if (!d_unhashed(dentry)) 759 d_drop(dentry); 760 dn = d_splice_alias(inode, dentry); 761 WARN_ON_ONCE(dn && dn != dentry); 762 } 763 file->f_mode |= FMODE_CREATED; 764 ret = finish_open(file, dentry, ceph_open); 765 } 766 767 spin_lock(&dentry->d_lock); 768 di->flags &= ~CEPH_DENTRY_ASYNC_CREATE; 769 wake_up_bit(&di->flags, CEPH_DENTRY_ASYNC_CREATE_BIT); 770 spin_unlock(&dentry->d_lock); 771 772 return ret; 773 } 774 775 /* 776 * Do a lookup + open with a single request. If we get a non-existent 777 * file or symlink, return 1 so the VFS can retry. 778 */ 779 int ceph_atomic_open(struct inode *dir, struct dentry *dentry, 780 struct file *file, unsigned flags, umode_t mode) 781 { 782 struct mnt_idmap *idmap = file_mnt_idmap(file); 783 struct ceph_fs_client *fsc = ceph_sb_to_fs_client(dir->i_sb); 784 struct ceph_client *cl = fsc->client; 785 struct ceph_mds_client *mdsc = fsc->mdsc; 786 struct ceph_mds_request *req; 787 struct inode *new_inode = NULL; 788 struct dentry *dn; 789 struct ceph_acl_sec_ctx as_ctx = {}; 790 bool try_async = ceph_test_mount_opt(fsc, ASYNC_DIROPS); 791 int mask; 792 int err; 793 char *path; 794 int pathlen; 795 u64 pathbase; 796 797 doutc(cl, "%p %llx.%llx dentry %p '%pd' %s flags %d mode 0%o\n", 798 dir, ceph_vinop(dir), dentry, dentry, 799 d_unhashed(dentry) ? "unhashed" : "hashed", flags, mode); 800 801 if (dentry->d_name.len > NAME_MAX) 802 return -ENAMETOOLONG; 803 804 err = ceph_wait_on_conflict_unlink(dentry); 805 if (err) 806 return err; 807 /* 808 * Do not truncate the file, since atomic_open is called before the 809 * permission check. The caller will do the truncation afterward. 810 */ 811 flags &= ~O_TRUNC; 812 813 dn = d_find_alias(dir); 814 if (!dn) { 815 try_async = false; 816 } else { 817 path = ceph_mdsc_build_path(mdsc, dn, &pathlen, &pathbase, 0); 818 if (IS_ERR(path)) { 819 try_async = false; 820 err = 0; 821 } else { 822 int fmode = ceph_flags_to_mode(flags); 823 824 mask = MAY_READ; 825 if (fmode & CEPH_FILE_MODE_WR) 826 mask |= MAY_WRITE; 827 err = ceph_mds_check_access(mdsc, path, mask); 828 } 829 ceph_mdsc_free_path(path, pathlen); 830 dput(dn); 831 832 /* For none EACCES cases will let the MDS do the mds auth check */ 833 if (err == -EACCES) { 834 return err; 835 } else if (err < 0) { 836 try_async = false; 837 err = 0; 838 } 839 } 840 841 retry: 842 if (flags & O_CREAT) { 843 if (ceph_quota_is_max_files_exceeded(dir)) 844 return -EDQUOT; 845 846 new_inode = ceph_new_inode(dir, dentry, &mode, &as_ctx); 847 if (IS_ERR(new_inode)) { 848 err = PTR_ERR(new_inode); 849 goto out_ctx; 850 } 851 /* Async create can't handle more than a page of xattrs */ 852 if (as_ctx.pagelist && 853 !list_is_singular(&as_ctx.pagelist->head)) 854 try_async = false; 855 } else if (!d_in_lookup(dentry)) { 856 /* If it's not being looked up, it's negative */ 857 return -ENOENT; 858 } 859 860 /* do the open */ 861 req = prepare_open_request(dir->i_sb, flags, mode); 862 if (IS_ERR(req)) { 863 err = PTR_ERR(req); 864 goto out_ctx; 865 } 866 req->r_dentry = dget(dentry); 867 req->r_num_caps = 2; 868 mask = CEPH_STAT_CAP_INODE | CEPH_CAP_AUTH_SHARED; 869 if (ceph_security_xattr_wanted(dir)) 870 mask |= CEPH_CAP_XATTR_SHARED; 871 req->r_args.open.mask = cpu_to_le32(mask); 872 req->r_parent = dir; 873 if (req->r_op == CEPH_MDS_OP_CREATE) 874 req->r_mnt_idmap = mnt_idmap_get(idmap); 875 ihold(dir); 876 if (IS_ENCRYPTED(dir)) { 877 set_bit(CEPH_MDS_R_FSCRYPT_FILE, &req->r_req_flags); 878 err = fscrypt_prepare_lookup_partial(dir, dentry); 879 if (err < 0) 880 goto out_req; 881 } 882 883 if (flags & O_CREAT) { 884 struct ceph_file_layout lo; 885 886 req->r_dentry_drop = CEPH_CAP_FILE_SHARED | CEPH_CAP_AUTH_EXCL | 887 CEPH_CAP_XATTR_EXCL; 888 req->r_dentry_unless = CEPH_CAP_FILE_EXCL; 889 890 ceph_as_ctx_to_req(req, &as_ctx); 891 892 if (try_async && (req->r_dir_caps = 893 try_prep_async_create(dir, dentry, &lo, 894 &req->r_deleg_ino))) { 895 struct ceph_vino vino = { .ino = req->r_deleg_ino, 896 .snap = CEPH_NOSNAP }; 897 struct ceph_dentry_info *di = ceph_dentry(dentry); 898 899 set_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags); 900 req->r_args.open.flags |= cpu_to_le32(CEPH_O_EXCL); 901 req->r_callback = ceph_async_create_cb; 902 903 /* Hash inode before RPC */ 904 new_inode = ceph_get_inode(dir->i_sb, vino, new_inode); 905 if (IS_ERR(new_inode)) { 906 err = PTR_ERR(new_inode); 907 new_inode = NULL; 908 goto out_req; 909 } 910 WARN_ON_ONCE(!(new_inode->i_state & I_NEW)); 911 912 spin_lock(&dentry->d_lock); 913 di->flags |= CEPH_DENTRY_ASYNC_CREATE; 914 spin_unlock(&dentry->d_lock); 915 916 err = ceph_mdsc_submit_request(mdsc, dir, req); 917 if (!err) { 918 err = ceph_finish_async_create(dir, new_inode, 919 dentry, file, 920 mode, req, 921 &as_ctx, &lo); 922 new_inode = NULL; 923 } else if (err == -EJUKEBOX) { 924 restore_deleg_ino(dir, req->r_deleg_ino); 925 ceph_mdsc_put_request(req); 926 discard_new_inode(new_inode); 927 ceph_release_acl_sec_ctx(&as_ctx); 928 memset(&as_ctx, 0, sizeof(as_ctx)); 929 new_inode = NULL; 930 try_async = false; 931 ceph_put_string(rcu_dereference_raw(lo.pool_ns)); 932 goto retry; 933 } 934 ceph_put_string(rcu_dereference_raw(lo.pool_ns)); 935 goto out_req; 936 } 937 } 938 939 set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags); 940 req->r_new_inode = new_inode; 941 new_inode = NULL; 942 err = ceph_mdsc_do_request(mdsc, (flags & O_CREAT) ? dir : NULL, req); 943 if (err == -ENOENT) { 944 dentry = ceph_handle_snapdir(req, dentry); 945 if (IS_ERR(dentry)) { 946 err = PTR_ERR(dentry); 947 goto out_req; 948 } 949 err = 0; 950 } 951 952 if (!err && (flags & O_CREAT) && !req->r_reply_info.head->is_dentry) 953 err = ceph_handle_notrace_create(dir, dentry); 954 955 if (d_in_lookup(dentry)) { 956 dn = ceph_finish_lookup(req, dentry, err); 957 if (IS_ERR(dn)) 958 err = PTR_ERR(dn); 959 } else { 960 /* we were given a hashed negative dentry */ 961 dn = NULL; 962 } 963 if (err) 964 goto out_req; 965 if (dn || d_really_is_negative(dentry) || d_is_symlink(dentry)) { 966 /* make vfs retry on splice, ENOENT, or symlink */ 967 doutc(cl, "finish_no_open on dn %p\n", dn); 968 err = finish_no_open(file, dn); 969 } else { 970 if (IS_ENCRYPTED(dir) && 971 !fscrypt_has_permitted_context(dir, d_inode(dentry))) { 972 pr_warn_client(cl, 973 "Inconsistent encryption context (parent %llx:%llx child %llx:%llx)\n", 974 ceph_vinop(dir), ceph_vinop(d_inode(dentry))); 975 goto out_req; 976 } 977 978 doutc(cl, "finish_open on dn %p\n", dn); 979 if (req->r_op == CEPH_MDS_OP_CREATE && req->r_reply_info.has_create_ino) { 980 struct inode *newino = d_inode(dentry); 981 982 cache_file_layout(dir, newino); 983 ceph_init_inode_acls(newino, &as_ctx); 984 file->f_mode |= FMODE_CREATED; 985 } 986 err = finish_open(file, dentry, ceph_open); 987 } 988 out_req: 989 ceph_mdsc_put_request(req); 990 iput(new_inode); 991 out_ctx: 992 ceph_release_acl_sec_ctx(&as_ctx); 993 doutc(cl, "result=%d\n", err); 994 return err; 995 } 996 997 int ceph_release(struct inode *inode, struct file *file) 998 { 999 struct ceph_client *cl = ceph_inode_to_client(inode); 1000 struct ceph_inode_info *ci = ceph_inode(inode); 1001 1002 if (S_ISDIR(inode->i_mode)) { 1003 struct ceph_dir_file_info *dfi = file->private_data; 1004 doutc(cl, "%p %llx.%llx dir file %p\n", inode, 1005 ceph_vinop(inode), file); 1006 WARN_ON(!list_empty(&dfi->file_info.rw_contexts)); 1007 1008 ceph_put_fmode(ci, dfi->file_info.fmode, 1); 1009 1010 if (dfi->last_readdir) 1011 ceph_mdsc_put_request(dfi->last_readdir); 1012 kfree(dfi->last_name); 1013 kfree(dfi->dir_info); 1014 kmem_cache_free(ceph_dir_file_cachep, dfi); 1015 } else { 1016 struct ceph_file_info *fi = file->private_data; 1017 doutc(cl, "%p %llx.%llx regular file %p\n", inode, 1018 ceph_vinop(inode), file); 1019 WARN_ON(!list_empty(&fi->rw_contexts)); 1020 1021 ceph_fscache_unuse_cookie(inode, file->f_mode & FMODE_WRITE); 1022 ceph_put_fmode(ci, fi->fmode, 1); 1023 1024 kmem_cache_free(ceph_file_cachep, fi); 1025 } 1026 1027 /* wake up anyone waiting for caps on this inode */ 1028 wake_up_all(&ci->i_cap_wq); 1029 return 0; 1030 } 1031 1032 enum { 1033 HAVE_RETRIED = 1, 1034 CHECK_EOF = 2, 1035 READ_INLINE = 3, 1036 }; 1037 1038 /* 1039 * Completely synchronous read and write methods. Direct from __user 1040 * buffer to osd, or directly to user pages (if O_DIRECT). 1041 * 1042 * If the read spans object boundary, just do multiple reads. (That's not 1043 * atomic, but good enough for now.) 1044 * 1045 * If we get a short result from the OSD, check against i_size; we need to 1046 * only return a short read to the caller if we hit EOF. 1047 */ 1048 ssize_t __ceph_sync_read(struct inode *inode, loff_t *ki_pos, 1049 struct iov_iter *to, int *retry_op, 1050 u64 *last_objver) 1051 { 1052 struct ceph_inode_info *ci = ceph_inode(inode); 1053 struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); 1054 struct ceph_client *cl = fsc->client; 1055 struct ceph_osd_client *osdc = &fsc->client->osdc; 1056 ssize_t ret; 1057 u64 off = *ki_pos; 1058 u64 len = iov_iter_count(to); 1059 u64 i_size = i_size_read(inode); 1060 bool sparse = IS_ENCRYPTED(inode) || ceph_test_mount_opt(fsc, SPARSEREAD); 1061 u64 objver = 0; 1062 1063 doutc(cl, "on inode %p %llx.%llx %llx~%llx\n", inode, 1064 ceph_vinop(inode), *ki_pos, len); 1065 1066 if (ceph_inode_is_shutdown(inode)) 1067 return -EIO; 1068 1069 if (!len) 1070 return 0; 1071 /* 1072 * flush any page cache pages in this range. this 1073 * will make concurrent normal and sync io slow, 1074 * but it will at least behave sensibly when they are 1075 * in sequence. 1076 */ 1077 ret = filemap_write_and_wait_range(inode->i_mapping, 1078 off, off + len - 1); 1079 if (ret < 0) 1080 return ret; 1081 1082 ret = 0; 1083 while ((len = iov_iter_count(to)) > 0) { 1084 struct ceph_osd_request *req; 1085 struct page **pages; 1086 int num_pages; 1087 size_t page_off; 1088 bool more; 1089 int idx; 1090 size_t left; 1091 struct ceph_osd_req_op *op; 1092 u64 read_off = off; 1093 u64 read_len = len; 1094 int extent_cnt; 1095 1096 /* determine new offset/length if encrypted */ 1097 ceph_fscrypt_adjust_off_and_len(inode, &read_off, &read_len); 1098 1099 doutc(cl, "orig %llu~%llu reading %llu~%llu", off, len, 1100 read_off, read_len); 1101 1102 req = ceph_osdc_new_request(osdc, &ci->i_layout, 1103 ci->i_vino, read_off, &read_len, 0, 1, 1104 sparse ? CEPH_OSD_OP_SPARSE_READ : 1105 CEPH_OSD_OP_READ, 1106 CEPH_OSD_FLAG_READ, 1107 NULL, ci->i_truncate_seq, 1108 ci->i_truncate_size, false); 1109 if (IS_ERR(req)) { 1110 ret = PTR_ERR(req); 1111 break; 1112 } 1113 1114 /* adjust len downward if the request truncated the len */ 1115 if (off + len > read_off + read_len) 1116 len = read_off + read_len - off; 1117 more = len < iov_iter_count(to); 1118 1119 num_pages = calc_pages_for(read_off, read_len); 1120 page_off = offset_in_page(off); 1121 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL); 1122 if (IS_ERR(pages)) { 1123 ceph_osdc_put_request(req); 1124 ret = PTR_ERR(pages); 1125 break; 1126 } 1127 1128 osd_req_op_extent_osd_data_pages(req, 0, pages, read_len, 1129 offset_in_page(read_off), 1130 false, false); 1131 1132 op = &req->r_ops[0]; 1133 if (sparse) { 1134 extent_cnt = __ceph_sparse_read_ext_count(inode, read_len); 1135 ret = ceph_alloc_sparse_ext_map(op, extent_cnt); 1136 if (ret) { 1137 ceph_osdc_put_request(req); 1138 break; 1139 } 1140 } 1141 1142 ceph_osdc_start_request(osdc, req); 1143 ret = ceph_osdc_wait_request(osdc, req); 1144 1145 ceph_update_read_metrics(&fsc->mdsc->metric, 1146 req->r_start_latency, 1147 req->r_end_latency, 1148 read_len, ret); 1149 1150 if (ret > 0) 1151 objver = req->r_version; 1152 1153 i_size = i_size_read(inode); 1154 doutc(cl, "%llu~%llu got %zd i_size %llu%s\n", off, len, 1155 ret, i_size, (more ? " MORE" : "")); 1156 1157 /* Fix it to go to end of extent map */ 1158 if (sparse && ret >= 0) 1159 ret = ceph_sparse_ext_map_end(op); 1160 else if (ret == -ENOENT) 1161 ret = 0; 1162 1163 if (ret > 0 && IS_ENCRYPTED(inode)) { 1164 int fret; 1165 1166 fret = ceph_fscrypt_decrypt_extents(inode, pages, 1167 read_off, op->extent.sparse_ext, 1168 op->extent.sparse_ext_cnt); 1169 if (fret < 0) { 1170 ret = fret; 1171 ceph_osdc_put_request(req); 1172 break; 1173 } 1174 1175 /* account for any partial block at the beginning */ 1176 fret -= (off - read_off); 1177 1178 /* 1179 * Short read after big offset adjustment? 1180 * Nothing is usable, just call it a zero 1181 * len read. 1182 */ 1183 fret = max(fret, 0); 1184 1185 /* account for partial block at the end */ 1186 ret = min_t(ssize_t, fret, len); 1187 } 1188 1189 ceph_osdc_put_request(req); 1190 1191 /* Short read but not EOF? Zero out the remainder. */ 1192 if (ret >= 0 && ret < len && (off + ret < i_size)) { 1193 int zlen = min(len - ret, i_size - off - ret); 1194 int zoff = page_off + ret; 1195 1196 doutc(cl, "zero gap %llu~%llu\n", off + ret, 1197 off + ret + zlen); 1198 ceph_zero_page_vector_range(zoff, zlen, pages); 1199 ret += zlen; 1200 } 1201 1202 idx = 0; 1203 if (ret <= 0) 1204 left = 0; 1205 else if (off + ret > i_size) 1206 left = i_size - off; 1207 else 1208 left = ret; 1209 while (left > 0) { 1210 size_t plen, copied; 1211 1212 plen = min_t(size_t, left, PAGE_SIZE - page_off); 1213 SetPageUptodate(pages[idx]); 1214 copied = copy_page_to_iter(pages[idx++], 1215 page_off, plen, to); 1216 off += copied; 1217 left -= copied; 1218 page_off = 0; 1219 if (copied < plen) { 1220 ret = -EFAULT; 1221 break; 1222 } 1223 } 1224 ceph_release_page_vector(pages, num_pages); 1225 1226 if (ret < 0) { 1227 if (ret == -EBLOCKLISTED) 1228 fsc->blocklisted = true; 1229 break; 1230 } 1231 1232 if (off >= i_size || !more) 1233 break; 1234 } 1235 1236 if (ret > 0) { 1237 if (off >= i_size) { 1238 *retry_op = CHECK_EOF; 1239 ret = i_size - *ki_pos; 1240 *ki_pos = i_size; 1241 } else { 1242 ret = off - *ki_pos; 1243 *ki_pos = off; 1244 } 1245 1246 if (last_objver) 1247 *last_objver = objver; 1248 } 1249 doutc(cl, "result %zd retry_op %d\n", ret, *retry_op); 1250 return ret; 1251 } 1252 1253 static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *to, 1254 int *retry_op) 1255 { 1256 struct file *file = iocb->ki_filp; 1257 struct inode *inode = file_inode(file); 1258 struct ceph_client *cl = ceph_inode_to_client(inode); 1259 1260 doutc(cl, "on file %p %llx~%zx %s\n", file, iocb->ki_pos, 1261 iov_iter_count(to), 1262 (file->f_flags & O_DIRECT) ? "O_DIRECT" : ""); 1263 1264 return __ceph_sync_read(inode, &iocb->ki_pos, to, retry_op, NULL); 1265 } 1266 1267 struct ceph_aio_request { 1268 struct kiocb *iocb; 1269 size_t total_len; 1270 bool write; 1271 bool should_dirty; 1272 int error; 1273 struct list_head osd_reqs; 1274 unsigned num_reqs; 1275 atomic_t pending_reqs; 1276 struct timespec64 mtime; 1277 struct ceph_cap_flush *prealloc_cf; 1278 }; 1279 1280 struct ceph_aio_work { 1281 struct work_struct work; 1282 struct ceph_osd_request *req; 1283 }; 1284 1285 static void ceph_aio_retry_work(struct work_struct *work); 1286 1287 static void ceph_aio_complete(struct inode *inode, 1288 struct ceph_aio_request *aio_req) 1289 { 1290 struct ceph_client *cl = ceph_inode_to_client(inode); 1291 struct ceph_inode_info *ci = ceph_inode(inode); 1292 int ret; 1293 1294 if (!atomic_dec_and_test(&aio_req->pending_reqs)) 1295 return; 1296 1297 if (aio_req->iocb->ki_flags & IOCB_DIRECT) 1298 inode_dio_end(inode); 1299 1300 ret = aio_req->error; 1301 if (!ret) 1302 ret = aio_req->total_len; 1303 1304 doutc(cl, "%p %llx.%llx rc %d\n", inode, ceph_vinop(inode), ret); 1305 1306 if (ret >= 0 && aio_req->write) { 1307 int dirty; 1308 1309 loff_t endoff = aio_req->iocb->ki_pos + aio_req->total_len; 1310 if (endoff > i_size_read(inode)) { 1311 if (ceph_inode_set_size(inode, endoff)) 1312 ceph_check_caps(ci, CHECK_CAPS_AUTHONLY); 1313 } 1314 1315 spin_lock(&ci->i_ceph_lock); 1316 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR, 1317 &aio_req->prealloc_cf); 1318 spin_unlock(&ci->i_ceph_lock); 1319 if (dirty) 1320 __mark_inode_dirty(inode, dirty); 1321 1322 } 1323 1324 ceph_put_cap_refs(ci, (aio_req->write ? CEPH_CAP_FILE_WR : 1325 CEPH_CAP_FILE_RD)); 1326 1327 aio_req->iocb->ki_complete(aio_req->iocb, ret); 1328 1329 ceph_free_cap_flush(aio_req->prealloc_cf); 1330 kfree(aio_req); 1331 } 1332 1333 static void ceph_aio_complete_req(struct ceph_osd_request *req) 1334 { 1335 int rc = req->r_result; 1336 struct inode *inode = req->r_inode; 1337 struct ceph_aio_request *aio_req = req->r_priv; 1338 struct ceph_osd_data *osd_data = osd_req_op_extent_osd_data(req, 0); 1339 struct ceph_osd_req_op *op = &req->r_ops[0]; 1340 struct ceph_client_metric *metric = &ceph_sb_to_mdsc(inode->i_sb)->metric; 1341 unsigned int len = osd_data->bvec_pos.iter.bi_size; 1342 bool sparse = (op->op == CEPH_OSD_OP_SPARSE_READ); 1343 struct ceph_client *cl = ceph_inode_to_client(inode); 1344 1345 BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_BVECS); 1346 BUG_ON(!osd_data->num_bvecs); 1347 1348 doutc(cl, "req %p inode %p %llx.%llx, rc %d bytes %u\n", req, 1349 inode, ceph_vinop(inode), rc, len); 1350 1351 if (rc == -EOLDSNAPC) { 1352 struct ceph_aio_work *aio_work; 1353 BUG_ON(!aio_req->write); 1354 1355 aio_work = kmalloc(sizeof(*aio_work), GFP_NOFS); 1356 if (aio_work) { 1357 INIT_WORK(&aio_work->work, ceph_aio_retry_work); 1358 aio_work->req = req; 1359 queue_work(ceph_inode_to_fs_client(inode)->inode_wq, 1360 &aio_work->work); 1361 return; 1362 } 1363 rc = -ENOMEM; 1364 } else if (!aio_req->write) { 1365 if (sparse && rc >= 0) 1366 rc = ceph_sparse_ext_map_end(op); 1367 if (rc == -ENOENT) 1368 rc = 0; 1369 if (rc >= 0 && len > rc) { 1370 struct iov_iter i; 1371 int zlen = len - rc; 1372 1373 /* 1374 * If read is satisfied by single OSD request, 1375 * it can pass EOF. Otherwise read is within 1376 * i_size. 1377 */ 1378 if (aio_req->num_reqs == 1) { 1379 loff_t i_size = i_size_read(inode); 1380 loff_t endoff = aio_req->iocb->ki_pos + rc; 1381 if (endoff < i_size) 1382 zlen = min_t(size_t, zlen, 1383 i_size - endoff); 1384 aio_req->total_len = rc + zlen; 1385 } 1386 1387 iov_iter_bvec(&i, ITER_DEST, osd_data->bvec_pos.bvecs, 1388 osd_data->num_bvecs, len); 1389 iov_iter_advance(&i, rc); 1390 iov_iter_zero(zlen, &i); 1391 } 1392 } 1393 1394 /* r_start_latency == 0 means the request was not submitted */ 1395 if (req->r_start_latency) { 1396 if (aio_req->write) 1397 ceph_update_write_metrics(metric, req->r_start_latency, 1398 req->r_end_latency, len, rc); 1399 else 1400 ceph_update_read_metrics(metric, req->r_start_latency, 1401 req->r_end_latency, len, rc); 1402 } 1403 1404 put_bvecs(osd_data->bvec_pos.bvecs, osd_data->num_bvecs, 1405 aio_req->should_dirty); 1406 ceph_osdc_put_request(req); 1407 1408 if (rc < 0) 1409 cmpxchg(&aio_req->error, 0, rc); 1410 1411 ceph_aio_complete(inode, aio_req); 1412 return; 1413 } 1414 1415 static void ceph_aio_retry_work(struct work_struct *work) 1416 { 1417 struct ceph_aio_work *aio_work = 1418 container_of(work, struct ceph_aio_work, work); 1419 struct ceph_osd_request *orig_req = aio_work->req; 1420 struct ceph_aio_request *aio_req = orig_req->r_priv; 1421 struct inode *inode = orig_req->r_inode; 1422 struct ceph_inode_info *ci = ceph_inode(inode); 1423 struct ceph_snap_context *snapc; 1424 struct ceph_osd_request *req; 1425 int ret; 1426 1427 spin_lock(&ci->i_ceph_lock); 1428 if (__ceph_have_pending_cap_snap(ci)) { 1429 struct ceph_cap_snap *capsnap = 1430 list_last_entry(&ci->i_cap_snaps, 1431 struct ceph_cap_snap, 1432 ci_item); 1433 snapc = ceph_get_snap_context(capsnap->context); 1434 } else { 1435 BUG_ON(!ci->i_head_snapc); 1436 snapc = ceph_get_snap_context(ci->i_head_snapc); 1437 } 1438 spin_unlock(&ci->i_ceph_lock); 1439 1440 req = ceph_osdc_alloc_request(orig_req->r_osdc, snapc, 1, 1441 false, GFP_NOFS); 1442 if (!req) { 1443 ret = -ENOMEM; 1444 req = orig_req; 1445 goto out; 1446 } 1447 1448 req->r_flags = /* CEPH_OSD_FLAG_ORDERSNAP | */ CEPH_OSD_FLAG_WRITE; 1449 ceph_oloc_copy(&req->r_base_oloc, &orig_req->r_base_oloc); 1450 ceph_oid_copy(&req->r_base_oid, &orig_req->r_base_oid); 1451 1452 req->r_ops[0] = orig_req->r_ops[0]; 1453 1454 req->r_mtime = aio_req->mtime; 1455 req->r_data_offset = req->r_ops[0].extent.offset; 1456 1457 ret = ceph_osdc_alloc_messages(req, GFP_NOFS); 1458 if (ret) { 1459 ceph_osdc_put_request(req); 1460 req = orig_req; 1461 goto out; 1462 } 1463 1464 ceph_osdc_put_request(orig_req); 1465 1466 req->r_callback = ceph_aio_complete_req; 1467 req->r_inode = inode; 1468 req->r_priv = aio_req; 1469 1470 ceph_osdc_start_request(req->r_osdc, req); 1471 out: 1472 if (ret < 0) { 1473 req->r_result = ret; 1474 ceph_aio_complete_req(req); 1475 } 1476 1477 ceph_put_snap_context(snapc); 1478 kfree(aio_work); 1479 } 1480 1481 static ssize_t 1482 ceph_direct_read_write(struct kiocb *iocb, struct iov_iter *iter, 1483 struct ceph_snap_context *snapc, 1484 struct ceph_cap_flush **pcf) 1485 { 1486 struct file *file = iocb->ki_filp; 1487 struct inode *inode = file_inode(file); 1488 struct ceph_inode_info *ci = ceph_inode(inode); 1489 struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); 1490 struct ceph_client *cl = fsc->client; 1491 struct ceph_client_metric *metric = &fsc->mdsc->metric; 1492 struct ceph_vino vino; 1493 struct ceph_osd_request *req; 1494 struct bio_vec *bvecs; 1495 struct ceph_aio_request *aio_req = NULL; 1496 int num_pages = 0; 1497 int flags; 1498 int ret = 0; 1499 struct timespec64 mtime = current_time(inode); 1500 size_t count = iov_iter_count(iter); 1501 loff_t pos = iocb->ki_pos; 1502 bool write = iov_iter_rw(iter) == WRITE; 1503 bool should_dirty = !write && user_backed_iter(iter); 1504 bool sparse = ceph_test_mount_opt(fsc, SPARSEREAD); 1505 1506 if (write && ceph_snap(file_inode(file)) != CEPH_NOSNAP) 1507 return -EROFS; 1508 1509 doutc(cl, "sync_direct_%s on file %p %lld~%u snapc %p seq %lld\n", 1510 (write ? "write" : "read"), file, pos, (unsigned)count, 1511 snapc, snapc ? snapc->seq : 0); 1512 1513 if (write) { 1514 int ret2; 1515 1516 ceph_fscache_invalidate(inode, true); 1517 1518 ret2 = invalidate_inode_pages2_range(inode->i_mapping, 1519 pos >> PAGE_SHIFT, 1520 (pos + count - 1) >> PAGE_SHIFT); 1521 if (ret2 < 0) 1522 doutc(cl, "invalidate_inode_pages2_range returned %d\n", 1523 ret2); 1524 1525 flags = /* CEPH_OSD_FLAG_ORDERSNAP | */ CEPH_OSD_FLAG_WRITE; 1526 } else { 1527 flags = CEPH_OSD_FLAG_READ; 1528 } 1529 1530 while (iov_iter_count(iter) > 0) { 1531 u64 size = iov_iter_count(iter); 1532 ssize_t len; 1533 struct ceph_osd_req_op *op; 1534 int readop = sparse ? CEPH_OSD_OP_SPARSE_READ : CEPH_OSD_OP_READ; 1535 int extent_cnt; 1536 1537 if (write) 1538 size = min_t(u64, size, fsc->mount_options->wsize); 1539 else 1540 size = min_t(u64, size, fsc->mount_options->rsize); 1541 1542 vino = ceph_vino(inode); 1543 req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout, 1544 vino, pos, &size, 0, 1545 1, 1546 write ? CEPH_OSD_OP_WRITE : readop, 1547 flags, snapc, 1548 ci->i_truncate_seq, 1549 ci->i_truncate_size, 1550 false); 1551 if (IS_ERR(req)) { 1552 ret = PTR_ERR(req); 1553 break; 1554 } 1555 1556 len = iter_get_bvecs_alloc(iter, size, &bvecs, &num_pages); 1557 if (len < 0) { 1558 ceph_osdc_put_request(req); 1559 ret = len; 1560 break; 1561 } 1562 if (len != size) 1563 osd_req_op_extent_update(req, 0, len); 1564 1565 /* 1566 * To simplify error handling, allow AIO when IO within i_size 1567 * or IO can be satisfied by single OSD request. 1568 */ 1569 if (pos == iocb->ki_pos && !is_sync_kiocb(iocb) && 1570 (len == count || pos + count <= i_size_read(inode))) { 1571 aio_req = kzalloc(sizeof(*aio_req), GFP_KERNEL); 1572 if (aio_req) { 1573 aio_req->iocb = iocb; 1574 aio_req->write = write; 1575 aio_req->should_dirty = should_dirty; 1576 INIT_LIST_HEAD(&aio_req->osd_reqs); 1577 if (write) { 1578 aio_req->mtime = mtime; 1579 swap(aio_req->prealloc_cf, *pcf); 1580 } 1581 } 1582 /* ignore error */ 1583 } 1584 1585 if (write) { 1586 /* 1587 * throw out any page cache pages in this range. this 1588 * may block. 1589 */ 1590 truncate_inode_pages_range(inode->i_mapping, pos, 1591 PAGE_ALIGN(pos + len) - 1); 1592 1593 req->r_mtime = mtime; 1594 } 1595 1596 osd_req_op_extent_osd_data_bvecs(req, 0, bvecs, num_pages, len); 1597 op = &req->r_ops[0]; 1598 if (sparse) { 1599 extent_cnt = __ceph_sparse_read_ext_count(inode, size); 1600 ret = ceph_alloc_sparse_ext_map(op, extent_cnt); 1601 if (ret) { 1602 ceph_osdc_put_request(req); 1603 break; 1604 } 1605 } 1606 1607 if (aio_req) { 1608 aio_req->total_len += len; 1609 aio_req->num_reqs++; 1610 atomic_inc(&aio_req->pending_reqs); 1611 1612 req->r_callback = ceph_aio_complete_req; 1613 req->r_inode = inode; 1614 req->r_priv = aio_req; 1615 list_add_tail(&req->r_private_item, &aio_req->osd_reqs); 1616 1617 pos += len; 1618 continue; 1619 } 1620 1621 ceph_osdc_start_request(req->r_osdc, req); 1622 ret = ceph_osdc_wait_request(&fsc->client->osdc, req); 1623 1624 if (write) 1625 ceph_update_write_metrics(metric, req->r_start_latency, 1626 req->r_end_latency, len, ret); 1627 else 1628 ceph_update_read_metrics(metric, req->r_start_latency, 1629 req->r_end_latency, len, ret); 1630 1631 size = i_size_read(inode); 1632 if (!write) { 1633 if (sparse && ret >= 0) 1634 ret = ceph_sparse_ext_map_end(op); 1635 else if (ret == -ENOENT) 1636 ret = 0; 1637 1638 if (ret >= 0 && ret < len && pos + ret < size) { 1639 struct iov_iter i; 1640 int zlen = min_t(size_t, len - ret, 1641 size - pos - ret); 1642 1643 iov_iter_bvec(&i, ITER_DEST, bvecs, num_pages, len); 1644 iov_iter_advance(&i, ret); 1645 iov_iter_zero(zlen, &i); 1646 ret += zlen; 1647 } 1648 if (ret >= 0) 1649 len = ret; 1650 } 1651 1652 put_bvecs(bvecs, num_pages, should_dirty); 1653 ceph_osdc_put_request(req); 1654 if (ret < 0) 1655 break; 1656 1657 pos += len; 1658 if (!write && pos >= size) 1659 break; 1660 1661 if (write && pos > size) { 1662 if (ceph_inode_set_size(inode, pos)) 1663 ceph_check_caps(ceph_inode(inode), 1664 CHECK_CAPS_AUTHONLY); 1665 } 1666 } 1667 1668 if (aio_req) { 1669 LIST_HEAD(osd_reqs); 1670 1671 if (aio_req->num_reqs == 0) { 1672 kfree(aio_req); 1673 return ret; 1674 } 1675 1676 ceph_get_cap_refs(ci, write ? CEPH_CAP_FILE_WR : 1677 CEPH_CAP_FILE_RD); 1678 1679 list_splice(&aio_req->osd_reqs, &osd_reqs); 1680 inode_dio_begin(inode); 1681 while (!list_empty(&osd_reqs)) { 1682 req = list_first_entry(&osd_reqs, 1683 struct ceph_osd_request, 1684 r_private_item); 1685 list_del_init(&req->r_private_item); 1686 if (ret >= 0) 1687 ceph_osdc_start_request(req->r_osdc, req); 1688 if (ret < 0) { 1689 req->r_result = ret; 1690 ceph_aio_complete_req(req); 1691 } 1692 } 1693 return -EIOCBQUEUED; 1694 } 1695 1696 if (ret != -EOLDSNAPC && pos > iocb->ki_pos) { 1697 ret = pos - iocb->ki_pos; 1698 iocb->ki_pos = pos; 1699 } 1700 return ret; 1701 } 1702 1703 /* 1704 * Synchronous write, straight from __user pointer or user pages. 1705 * 1706 * If write spans object boundary, just do multiple writes. (For a 1707 * correct atomic write, we should e.g. take write locks on all 1708 * objects, rollback on failure, etc.) 1709 */ 1710 static ssize_t 1711 ceph_sync_write(struct kiocb *iocb, struct iov_iter *from, loff_t pos, 1712 struct ceph_snap_context *snapc) 1713 { 1714 struct file *file = iocb->ki_filp; 1715 struct inode *inode = file_inode(file); 1716 struct ceph_inode_info *ci = ceph_inode(inode); 1717 struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); 1718 struct ceph_client *cl = fsc->client; 1719 struct ceph_osd_client *osdc = &fsc->client->osdc; 1720 struct ceph_osd_request *req; 1721 struct page **pages; 1722 u64 len; 1723 int num_pages; 1724 int written = 0; 1725 int ret; 1726 bool check_caps = false; 1727 struct timespec64 mtime = current_time(inode); 1728 size_t count = iov_iter_count(from); 1729 1730 if (ceph_snap(file_inode(file)) != CEPH_NOSNAP) 1731 return -EROFS; 1732 1733 doutc(cl, "on file %p %lld~%u snapc %p seq %lld\n", file, pos, 1734 (unsigned)count, snapc, snapc->seq); 1735 1736 ret = filemap_write_and_wait_range(inode->i_mapping, 1737 pos, pos + count - 1); 1738 if (ret < 0) 1739 return ret; 1740 1741 ceph_fscache_invalidate(inode, false); 1742 1743 while ((len = iov_iter_count(from)) > 0) { 1744 size_t left; 1745 int n; 1746 u64 write_pos = pos; 1747 u64 write_len = len; 1748 u64 objnum, objoff; 1749 u32 xlen; 1750 u64 assert_ver = 0; 1751 bool rmw; 1752 bool first, last; 1753 struct iov_iter saved_iter = *from; 1754 size_t off; 1755 1756 ceph_fscrypt_adjust_off_and_len(inode, &write_pos, &write_len); 1757 1758 /* clamp the length to the end of first object */ 1759 ceph_calc_file_object_mapping(&ci->i_layout, write_pos, 1760 write_len, &objnum, &objoff, 1761 &xlen); 1762 write_len = xlen; 1763 1764 /* adjust len downward if it goes beyond current object */ 1765 if (pos + len > write_pos + write_len) 1766 len = write_pos + write_len - pos; 1767 1768 /* 1769 * If we had to adjust the length or position to align with a 1770 * crypto block, then we must do a read/modify/write cycle. We 1771 * use a version assertion to redrive the thing if something 1772 * changes in between. 1773 */ 1774 first = pos != write_pos; 1775 last = (pos + len) != (write_pos + write_len); 1776 rmw = first || last; 1777 1778 doutc(cl, "ino %llx %lld~%llu adjusted %lld~%llu -- %srmw\n", 1779 ci->i_vino.ino, pos, len, write_pos, write_len, 1780 rmw ? "" : "no "); 1781 1782 /* 1783 * The data is emplaced into the page as it would be if it were 1784 * in an array of pagecache pages. 1785 */ 1786 num_pages = calc_pages_for(write_pos, write_len); 1787 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL); 1788 if (IS_ERR(pages)) { 1789 ret = PTR_ERR(pages); 1790 break; 1791 } 1792 1793 /* Do we need to preload the pages? */ 1794 if (rmw) { 1795 u64 first_pos = write_pos; 1796 u64 last_pos = (write_pos + write_len) - CEPH_FSCRYPT_BLOCK_SIZE; 1797 u64 read_len = CEPH_FSCRYPT_BLOCK_SIZE; 1798 struct ceph_osd_req_op *op; 1799 1800 /* We should only need to do this for encrypted inodes */ 1801 WARN_ON_ONCE(!IS_ENCRYPTED(inode)); 1802 1803 /* No need to do two reads if first and last blocks are same */ 1804 if (first && last_pos == first_pos) 1805 last = false; 1806 1807 /* 1808 * Allocate a read request for one or two extents, 1809 * depending on how the request was aligned. 1810 */ 1811 req = ceph_osdc_new_request(osdc, &ci->i_layout, 1812 ci->i_vino, first ? first_pos : last_pos, 1813 &read_len, 0, (first && last) ? 2 : 1, 1814 CEPH_OSD_OP_SPARSE_READ, CEPH_OSD_FLAG_READ, 1815 NULL, ci->i_truncate_seq, 1816 ci->i_truncate_size, false); 1817 if (IS_ERR(req)) { 1818 ceph_release_page_vector(pages, num_pages); 1819 ret = PTR_ERR(req); 1820 break; 1821 } 1822 1823 /* Something is misaligned! */ 1824 if (read_len != CEPH_FSCRYPT_BLOCK_SIZE) { 1825 ceph_osdc_put_request(req); 1826 ceph_release_page_vector(pages, num_pages); 1827 ret = -EIO; 1828 break; 1829 } 1830 1831 /* Add extent for first block? */ 1832 op = &req->r_ops[0]; 1833 1834 if (first) { 1835 osd_req_op_extent_osd_data_pages(req, 0, pages, 1836 CEPH_FSCRYPT_BLOCK_SIZE, 1837 offset_in_page(first_pos), 1838 false, false); 1839 /* We only expect a single extent here */ 1840 ret = __ceph_alloc_sparse_ext_map(op, 1); 1841 if (ret) { 1842 ceph_osdc_put_request(req); 1843 ceph_release_page_vector(pages, num_pages); 1844 break; 1845 } 1846 } 1847 1848 /* Add extent for last block */ 1849 if (last) { 1850 /* Init the other extent if first extent has been used */ 1851 if (first) { 1852 op = &req->r_ops[1]; 1853 osd_req_op_extent_init(req, 1, 1854 CEPH_OSD_OP_SPARSE_READ, 1855 last_pos, CEPH_FSCRYPT_BLOCK_SIZE, 1856 ci->i_truncate_size, 1857 ci->i_truncate_seq); 1858 } 1859 1860 ret = __ceph_alloc_sparse_ext_map(op, 1); 1861 if (ret) { 1862 ceph_osdc_put_request(req); 1863 ceph_release_page_vector(pages, num_pages); 1864 break; 1865 } 1866 1867 osd_req_op_extent_osd_data_pages(req, first ? 1 : 0, 1868 &pages[num_pages - 1], 1869 CEPH_FSCRYPT_BLOCK_SIZE, 1870 offset_in_page(last_pos), 1871 false, false); 1872 } 1873 1874 ceph_osdc_start_request(osdc, req); 1875 ret = ceph_osdc_wait_request(osdc, req); 1876 1877 /* FIXME: length field is wrong if there are 2 extents */ 1878 ceph_update_read_metrics(&fsc->mdsc->metric, 1879 req->r_start_latency, 1880 req->r_end_latency, 1881 read_len, ret); 1882 1883 /* Ok if object is not already present */ 1884 if (ret == -ENOENT) { 1885 /* 1886 * If there is no object, then we can't assert 1887 * on its version. Set it to 0, and we'll use an 1888 * exclusive create instead. 1889 */ 1890 ceph_osdc_put_request(req); 1891 ret = 0; 1892 1893 /* 1894 * zero out the soon-to-be uncopied parts of the 1895 * first and last pages. 1896 */ 1897 if (first) 1898 zero_user_segment(pages[0], 0, 1899 offset_in_page(first_pos)); 1900 if (last) 1901 zero_user_segment(pages[num_pages - 1], 1902 offset_in_page(last_pos), 1903 PAGE_SIZE); 1904 } else { 1905 if (ret < 0) { 1906 ceph_osdc_put_request(req); 1907 ceph_release_page_vector(pages, num_pages); 1908 break; 1909 } 1910 1911 op = &req->r_ops[0]; 1912 if (op->extent.sparse_ext_cnt == 0) { 1913 if (first) 1914 zero_user_segment(pages[0], 0, 1915 offset_in_page(first_pos)); 1916 else 1917 zero_user_segment(pages[num_pages - 1], 1918 offset_in_page(last_pos), 1919 PAGE_SIZE); 1920 } else if (op->extent.sparse_ext_cnt != 1 || 1921 ceph_sparse_ext_map_end(op) != 1922 CEPH_FSCRYPT_BLOCK_SIZE) { 1923 ret = -EIO; 1924 ceph_osdc_put_request(req); 1925 ceph_release_page_vector(pages, num_pages); 1926 break; 1927 } 1928 1929 if (first && last) { 1930 op = &req->r_ops[1]; 1931 if (op->extent.sparse_ext_cnt == 0) { 1932 zero_user_segment(pages[num_pages - 1], 1933 offset_in_page(last_pos), 1934 PAGE_SIZE); 1935 } else if (op->extent.sparse_ext_cnt != 1 || 1936 ceph_sparse_ext_map_end(op) != 1937 CEPH_FSCRYPT_BLOCK_SIZE) { 1938 ret = -EIO; 1939 ceph_osdc_put_request(req); 1940 ceph_release_page_vector(pages, num_pages); 1941 break; 1942 } 1943 } 1944 1945 /* Grab assert version. It must be non-zero. */ 1946 assert_ver = req->r_version; 1947 WARN_ON_ONCE(ret > 0 && assert_ver == 0); 1948 1949 ceph_osdc_put_request(req); 1950 if (first) { 1951 ret = ceph_fscrypt_decrypt_block_inplace(inode, 1952 pages[0], CEPH_FSCRYPT_BLOCK_SIZE, 1953 offset_in_page(first_pos), 1954 first_pos >> CEPH_FSCRYPT_BLOCK_SHIFT); 1955 if (ret < 0) { 1956 ceph_release_page_vector(pages, num_pages); 1957 break; 1958 } 1959 } 1960 if (last) { 1961 ret = ceph_fscrypt_decrypt_block_inplace(inode, 1962 pages[num_pages - 1], 1963 CEPH_FSCRYPT_BLOCK_SIZE, 1964 offset_in_page(last_pos), 1965 last_pos >> CEPH_FSCRYPT_BLOCK_SHIFT); 1966 if (ret < 0) { 1967 ceph_release_page_vector(pages, num_pages); 1968 break; 1969 } 1970 } 1971 } 1972 } 1973 1974 left = len; 1975 off = offset_in_page(pos); 1976 for (n = 0; n < num_pages; n++) { 1977 size_t plen = min_t(size_t, left, PAGE_SIZE - off); 1978 1979 /* copy the data */ 1980 ret = copy_page_from_iter(pages[n], off, plen, from); 1981 if (ret != plen) { 1982 ret = -EFAULT; 1983 break; 1984 } 1985 off = 0; 1986 left -= ret; 1987 } 1988 if (ret < 0) { 1989 doutc(cl, "write failed with %d\n", ret); 1990 ceph_release_page_vector(pages, num_pages); 1991 break; 1992 } 1993 1994 if (IS_ENCRYPTED(inode)) { 1995 ret = ceph_fscrypt_encrypt_pages(inode, pages, 1996 write_pos, write_len, 1997 GFP_KERNEL); 1998 if (ret < 0) { 1999 doutc(cl, "encryption failed with %d\n", ret); 2000 ceph_release_page_vector(pages, num_pages); 2001 break; 2002 } 2003 } 2004 2005 req = ceph_osdc_new_request(osdc, &ci->i_layout, 2006 ci->i_vino, write_pos, &write_len, 2007 rmw ? 1 : 0, rmw ? 2 : 1, 2008 CEPH_OSD_OP_WRITE, 2009 CEPH_OSD_FLAG_WRITE, 2010 snapc, ci->i_truncate_seq, 2011 ci->i_truncate_size, false); 2012 if (IS_ERR(req)) { 2013 ret = PTR_ERR(req); 2014 ceph_release_page_vector(pages, num_pages); 2015 break; 2016 } 2017 2018 doutc(cl, "write op %lld~%llu\n", write_pos, write_len); 2019 osd_req_op_extent_osd_data_pages(req, rmw ? 1 : 0, pages, write_len, 2020 offset_in_page(write_pos), false, 2021 true); 2022 req->r_inode = inode; 2023 req->r_mtime = mtime; 2024 2025 /* Set up the assertion */ 2026 if (rmw) { 2027 /* 2028 * Set up the assertion. If we don't have a version 2029 * number, then the object doesn't exist yet. Use an 2030 * exclusive create instead of a version assertion in 2031 * that case. 2032 */ 2033 if (assert_ver) { 2034 osd_req_op_init(req, 0, CEPH_OSD_OP_ASSERT_VER, 0); 2035 req->r_ops[0].assert_ver.ver = assert_ver; 2036 } else { 2037 osd_req_op_init(req, 0, CEPH_OSD_OP_CREATE, 2038 CEPH_OSD_OP_FLAG_EXCL); 2039 } 2040 } 2041 2042 ceph_osdc_start_request(osdc, req); 2043 ret = ceph_osdc_wait_request(osdc, req); 2044 2045 ceph_update_write_metrics(&fsc->mdsc->metric, req->r_start_latency, 2046 req->r_end_latency, len, ret); 2047 ceph_osdc_put_request(req); 2048 if (ret != 0) { 2049 doutc(cl, "osd write returned %d\n", ret); 2050 /* Version changed! Must re-do the rmw cycle */ 2051 if ((assert_ver && (ret == -ERANGE || ret == -EOVERFLOW)) || 2052 (!assert_ver && ret == -EEXIST)) { 2053 /* We should only ever see this on a rmw */ 2054 WARN_ON_ONCE(!rmw); 2055 2056 /* The version should never go backward */ 2057 WARN_ON_ONCE(ret == -EOVERFLOW); 2058 2059 *from = saved_iter; 2060 2061 /* FIXME: limit number of times we loop? */ 2062 continue; 2063 } 2064 ceph_set_error_write(ci); 2065 break; 2066 } 2067 2068 ceph_clear_error_write(ci); 2069 2070 /* 2071 * We successfully wrote to a range of the file. Declare 2072 * that region of the pagecache invalid. 2073 */ 2074 ret = invalidate_inode_pages2_range( 2075 inode->i_mapping, 2076 pos >> PAGE_SHIFT, 2077 (pos + len - 1) >> PAGE_SHIFT); 2078 if (ret < 0) { 2079 doutc(cl, "invalidate_inode_pages2_range returned %d\n", 2080 ret); 2081 ret = 0; 2082 } 2083 pos += len; 2084 written += len; 2085 doutc(cl, "written %d\n", written); 2086 if (pos > i_size_read(inode)) { 2087 check_caps = ceph_inode_set_size(inode, pos); 2088 if (check_caps) 2089 ceph_check_caps(ceph_inode(inode), 2090 CHECK_CAPS_AUTHONLY); 2091 } 2092 2093 } 2094 2095 if (ret != -EOLDSNAPC && written > 0) { 2096 ret = written; 2097 iocb->ki_pos = pos; 2098 } 2099 doutc(cl, "returning %d\n", ret); 2100 return ret; 2101 } 2102 2103 /* 2104 * Wrap generic_file_aio_read with checks for cap bits on the inode. 2105 * Atomically grab references, so that those bits are not released 2106 * back to the MDS mid-read. 2107 * 2108 * Hmm, the sync read case isn't actually async... should it be? 2109 */ 2110 static ssize_t ceph_read_iter(struct kiocb *iocb, struct iov_iter *to) 2111 { 2112 struct file *filp = iocb->ki_filp; 2113 struct ceph_file_info *fi = filp->private_data; 2114 size_t len = iov_iter_count(to); 2115 struct inode *inode = file_inode(filp); 2116 struct ceph_inode_info *ci = ceph_inode(inode); 2117 bool direct_lock = iocb->ki_flags & IOCB_DIRECT; 2118 struct ceph_client *cl = ceph_inode_to_client(inode); 2119 ssize_t ret; 2120 int want = 0, got = 0; 2121 int retry_op = 0, read = 0; 2122 2123 again: 2124 doutc(cl, "%llu~%u trying to get caps on %p %llx.%llx\n", 2125 iocb->ki_pos, (unsigned)len, inode, ceph_vinop(inode)); 2126 2127 if (ceph_inode_is_shutdown(inode)) 2128 return -ESTALE; 2129 2130 if (direct_lock) 2131 ceph_start_io_direct(inode); 2132 else 2133 ceph_start_io_read(inode); 2134 2135 if (!(fi->flags & CEPH_F_SYNC) && !direct_lock) 2136 want |= CEPH_CAP_FILE_CACHE; 2137 if (fi->fmode & CEPH_FILE_MODE_LAZY) 2138 want |= CEPH_CAP_FILE_LAZYIO; 2139 2140 ret = ceph_get_caps(filp, CEPH_CAP_FILE_RD, want, -1, &got); 2141 if (ret < 0) { 2142 if (direct_lock) 2143 ceph_end_io_direct(inode); 2144 else 2145 ceph_end_io_read(inode); 2146 return ret; 2147 } 2148 2149 if ((got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0 || 2150 (iocb->ki_flags & IOCB_DIRECT) || 2151 (fi->flags & CEPH_F_SYNC)) { 2152 2153 doutc(cl, "sync %p %llx.%llx %llu~%u got cap refs on %s\n", 2154 inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len, 2155 ceph_cap_string(got)); 2156 2157 if (!ceph_has_inline_data(ci)) { 2158 if (!retry_op && 2159 (iocb->ki_flags & IOCB_DIRECT) && 2160 !IS_ENCRYPTED(inode)) { 2161 ret = ceph_direct_read_write(iocb, to, 2162 NULL, NULL); 2163 if (ret >= 0 && ret < len) 2164 retry_op = CHECK_EOF; 2165 } else { 2166 ret = ceph_sync_read(iocb, to, &retry_op); 2167 } 2168 } else { 2169 retry_op = READ_INLINE; 2170 } 2171 } else { 2172 CEPH_DEFINE_RW_CONTEXT(rw_ctx, got); 2173 doutc(cl, "async %p %llx.%llx %llu~%u got cap refs on %s\n", 2174 inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len, 2175 ceph_cap_string(got)); 2176 ceph_add_rw_context(fi, &rw_ctx); 2177 ret = generic_file_read_iter(iocb, to); 2178 ceph_del_rw_context(fi, &rw_ctx); 2179 } 2180 2181 doutc(cl, "%p %llx.%llx dropping cap refs on %s = %d\n", 2182 inode, ceph_vinop(inode), ceph_cap_string(got), (int)ret); 2183 ceph_put_cap_refs(ci, got); 2184 2185 if (direct_lock) 2186 ceph_end_io_direct(inode); 2187 else 2188 ceph_end_io_read(inode); 2189 2190 if (retry_op > HAVE_RETRIED && ret >= 0) { 2191 int statret; 2192 struct page *page = NULL; 2193 loff_t i_size; 2194 int mask = CEPH_STAT_CAP_SIZE; 2195 if (retry_op == READ_INLINE) { 2196 page = __page_cache_alloc(GFP_KERNEL); 2197 if (!page) 2198 return -ENOMEM; 2199 2200 mask = CEPH_STAT_CAP_INLINE_DATA; 2201 } 2202 2203 statret = __ceph_do_getattr(inode, page, mask, !!page); 2204 if (statret < 0) { 2205 if (page) 2206 __free_page(page); 2207 if (statret == -ENODATA) { 2208 BUG_ON(retry_op != READ_INLINE); 2209 goto again; 2210 } 2211 return statret; 2212 } 2213 2214 i_size = i_size_read(inode); 2215 if (retry_op == READ_INLINE) { 2216 BUG_ON(ret > 0 || read > 0); 2217 if (iocb->ki_pos < i_size && 2218 iocb->ki_pos < PAGE_SIZE) { 2219 loff_t end = min_t(loff_t, i_size, 2220 iocb->ki_pos + len); 2221 end = min_t(loff_t, end, PAGE_SIZE); 2222 if (statret < end) 2223 zero_user_segment(page, statret, end); 2224 ret = copy_page_to_iter(page, 2225 iocb->ki_pos & ~PAGE_MASK, 2226 end - iocb->ki_pos, to); 2227 iocb->ki_pos += ret; 2228 read += ret; 2229 } 2230 if (iocb->ki_pos < i_size && read < len) { 2231 size_t zlen = min_t(size_t, len - read, 2232 i_size - iocb->ki_pos); 2233 ret = iov_iter_zero(zlen, to); 2234 iocb->ki_pos += ret; 2235 read += ret; 2236 } 2237 __free_pages(page, 0); 2238 return read; 2239 } 2240 2241 /* hit EOF or hole? */ 2242 if (retry_op == CHECK_EOF && iocb->ki_pos < i_size && 2243 ret < len) { 2244 doutc(cl, "may hit hole, ppos %lld < size %lld, reading more\n", 2245 iocb->ki_pos, i_size); 2246 2247 read += ret; 2248 len -= ret; 2249 retry_op = HAVE_RETRIED; 2250 goto again; 2251 } 2252 } 2253 2254 if (ret >= 0) 2255 ret += read; 2256 2257 return ret; 2258 } 2259 2260 /* 2261 * Wrap filemap_splice_read with checks for cap bits on the inode. 2262 * Atomically grab references, so that those bits are not released 2263 * back to the MDS mid-read. 2264 */ 2265 static ssize_t ceph_splice_read(struct file *in, loff_t *ppos, 2266 struct pipe_inode_info *pipe, 2267 size_t len, unsigned int flags) 2268 { 2269 struct ceph_file_info *fi = in->private_data; 2270 struct inode *inode = file_inode(in); 2271 struct ceph_inode_info *ci = ceph_inode(inode); 2272 ssize_t ret; 2273 int want = 0, got = 0; 2274 CEPH_DEFINE_RW_CONTEXT(rw_ctx, 0); 2275 2276 dout("splice_read %p %llx.%llx %llu~%zu trying to get caps on %p\n", 2277 inode, ceph_vinop(inode), *ppos, len, inode); 2278 2279 if (ceph_inode_is_shutdown(inode)) 2280 return -ESTALE; 2281 2282 if (ceph_has_inline_data(ci) || 2283 (fi->flags & CEPH_F_SYNC)) 2284 return copy_splice_read(in, ppos, pipe, len, flags); 2285 2286 ceph_start_io_read(inode); 2287 2288 want = CEPH_CAP_FILE_CACHE; 2289 if (fi->fmode & CEPH_FILE_MODE_LAZY) 2290 want |= CEPH_CAP_FILE_LAZYIO; 2291 2292 ret = ceph_get_caps(in, CEPH_CAP_FILE_RD, want, -1, &got); 2293 if (ret < 0) 2294 goto out_end; 2295 2296 if ((got & (CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO)) == 0) { 2297 dout("splice_read/sync %p %llx.%llx %llu~%zu got cap refs on %s\n", 2298 inode, ceph_vinop(inode), *ppos, len, 2299 ceph_cap_string(got)); 2300 2301 ceph_put_cap_refs(ci, got); 2302 ceph_end_io_read(inode); 2303 return copy_splice_read(in, ppos, pipe, len, flags); 2304 } 2305 2306 dout("splice_read %p %llx.%llx %llu~%zu got cap refs on %s\n", 2307 inode, ceph_vinop(inode), *ppos, len, ceph_cap_string(got)); 2308 2309 rw_ctx.caps = got; 2310 ceph_add_rw_context(fi, &rw_ctx); 2311 ret = filemap_splice_read(in, ppos, pipe, len, flags); 2312 ceph_del_rw_context(fi, &rw_ctx); 2313 2314 dout("splice_read %p %llx.%llx dropping cap refs on %s = %zd\n", 2315 inode, ceph_vinop(inode), ceph_cap_string(got), ret); 2316 2317 ceph_put_cap_refs(ci, got); 2318 out_end: 2319 ceph_end_io_read(inode); 2320 return ret; 2321 } 2322 2323 /* 2324 * Take cap references to avoid releasing caps to MDS mid-write. 2325 * 2326 * If we are synchronous, and write with an old snap context, the OSD 2327 * may return EOLDSNAPC. In that case, retry the write.. _after_ 2328 * dropping our cap refs and allowing the pending snap to logically 2329 * complete _before_ this write occurs. 2330 * 2331 * If we are near ENOSPC, write synchronously. 2332 */ 2333 static ssize_t ceph_write_iter(struct kiocb *iocb, struct iov_iter *from) 2334 { 2335 struct file *file = iocb->ki_filp; 2336 struct ceph_file_info *fi = file->private_data; 2337 struct inode *inode = file_inode(file); 2338 struct ceph_inode_info *ci = ceph_inode(inode); 2339 struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); 2340 struct ceph_client *cl = fsc->client; 2341 struct ceph_osd_client *osdc = &fsc->client->osdc; 2342 struct ceph_cap_flush *prealloc_cf; 2343 ssize_t count, written = 0; 2344 int err, want = 0, got; 2345 bool direct_lock = false; 2346 u32 map_flags; 2347 u64 pool_flags; 2348 loff_t pos; 2349 loff_t limit = max(i_size_read(inode), fsc->max_file_size); 2350 2351 if (ceph_inode_is_shutdown(inode)) 2352 return -ESTALE; 2353 2354 if (ceph_snap(inode) != CEPH_NOSNAP) 2355 return -EROFS; 2356 2357 prealloc_cf = ceph_alloc_cap_flush(); 2358 if (!prealloc_cf) 2359 return -ENOMEM; 2360 2361 if ((iocb->ki_flags & (IOCB_DIRECT | IOCB_APPEND)) == IOCB_DIRECT) 2362 direct_lock = true; 2363 2364 retry_snap: 2365 if (direct_lock) 2366 ceph_start_io_direct(inode); 2367 else 2368 ceph_start_io_write(inode); 2369 2370 if (iocb->ki_flags & IOCB_APPEND) { 2371 err = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE, false); 2372 if (err < 0) 2373 goto out; 2374 } 2375 2376 err = generic_write_checks(iocb, from); 2377 if (err <= 0) 2378 goto out; 2379 2380 pos = iocb->ki_pos; 2381 if (unlikely(pos >= limit)) { 2382 err = -EFBIG; 2383 goto out; 2384 } else { 2385 iov_iter_truncate(from, limit - pos); 2386 } 2387 2388 count = iov_iter_count(from); 2389 if (ceph_quota_is_max_bytes_exceeded(inode, pos + count)) { 2390 err = -EDQUOT; 2391 goto out; 2392 } 2393 2394 down_read(&osdc->lock); 2395 map_flags = osdc->osdmap->flags; 2396 pool_flags = ceph_pg_pool_flags(osdc->osdmap, ci->i_layout.pool_id); 2397 up_read(&osdc->lock); 2398 if ((map_flags & CEPH_OSDMAP_FULL) || 2399 (pool_flags & CEPH_POOL_FLAG_FULL)) { 2400 err = -ENOSPC; 2401 goto out; 2402 } 2403 2404 err = file_remove_privs(file); 2405 if (err) 2406 goto out; 2407 2408 doutc(cl, "%p %llx.%llx %llu~%zd getting caps. i_size %llu\n", 2409 inode, ceph_vinop(inode), pos, count, 2410 i_size_read(inode)); 2411 if (!(fi->flags & CEPH_F_SYNC) && !direct_lock) 2412 want |= CEPH_CAP_FILE_BUFFER; 2413 if (fi->fmode & CEPH_FILE_MODE_LAZY) 2414 want |= CEPH_CAP_FILE_LAZYIO; 2415 got = 0; 2416 err = ceph_get_caps(file, CEPH_CAP_FILE_WR, want, pos + count, &got); 2417 if (err < 0) 2418 goto out; 2419 2420 err = file_update_time(file); 2421 if (err) 2422 goto out_caps; 2423 2424 inode_inc_iversion_raw(inode); 2425 2426 doutc(cl, "%p %llx.%llx %llu~%zd got cap refs on %s\n", 2427 inode, ceph_vinop(inode), pos, count, ceph_cap_string(got)); 2428 2429 if ((got & (CEPH_CAP_FILE_BUFFER|CEPH_CAP_FILE_LAZYIO)) == 0 || 2430 (iocb->ki_flags & IOCB_DIRECT) || (fi->flags & CEPH_F_SYNC) || 2431 (ci->i_ceph_flags & CEPH_I_ERROR_WRITE)) { 2432 struct ceph_snap_context *snapc; 2433 struct iov_iter data; 2434 2435 spin_lock(&ci->i_ceph_lock); 2436 if (__ceph_have_pending_cap_snap(ci)) { 2437 struct ceph_cap_snap *capsnap = 2438 list_last_entry(&ci->i_cap_snaps, 2439 struct ceph_cap_snap, 2440 ci_item); 2441 snapc = ceph_get_snap_context(capsnap->context); 2442 } else { 2443 BUG_ON(!ci->i_head_snapc); 2444 snapc = ceph_get_snap_context(ci->i_head_snapc); 2445 } 2446 spin_unlock(&ci->i_ceph_lock); 2447 2448 /* we might need to revert back to that point */ 2449 data = *from; 2450 if ((iocb->ki_flags & IOCB_DIRECT) && !IS_ENCRYPTED(inode)) 2451 written = ceph_direct_read_write(iocb, &data, snapc, 2452 &prealloc_cf); 2453 else 2454 written = ceph_sync_write(iocb, &data, pos, snapc); 2455 if (direct_lock) 2456 ceph_end_io_direct(inode); 2457 else 2458 ceph_end_io_write(inode); 2459 if (written > 0) 2460 iov_iter_advance(from, written); 2461 ceph_put_snap_context(snapc); 2462 } else { 2463 /* 2464 * No need to acquire the i_truncate_mutex. Because 2465 * the MDS revokes Fwb caps before sending truncate 2466 * message to us. We can't get Fwb cap while there 2467 * are pending vmtruncate. So write and vmtruncate 2468 * can not run at the same time 2469 */ 2470 written = generic_perform_write(iocb, from); 2471 ceph_end_io_write(inode); 2472 } 2473 2474 if (written >= 0) { 2475 int dirty; 2476 2477 spin_lock(&ci->i_ceph_lock); 2478 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR, 2479 &prealloc_cf); 2480 spin_unlock(&ci->i_ceph_lock); 2481 if (dirty) 2482 __mark_inode_dirty(inode, dirty); 2483 if (ceph_quota_is_max_bytes_approaching(inode, iocb->ki_pos)) 2484 ceph_check_caps(ci, CHECK_CAPS_FLUSH); 2485 } 2486 2487 doutc(cl, "%p %llx.%llx %llu~%u dropping cap refs on %s\n", 2488 inode, ceph_vinop(inode), pos, (unsigned)count, 2489 ceph_cap_string(got)); 2490 ceph_put_cap_refs(ci, got); 2491 2492 if (written == -EOLDSNAPC) { 2493 doutc(cl, "%p %llx.%llx %llu~%u" "got EOLDSNAPC, retrying\n", 2494 inode, ceph_vinop(inode), pos, (unsigned)count); 2495 goto retry_snap; 2496 } 2497 2498 if (written >= 0) { 2499 if ((map_flags & CEPH_OSDMAP_NEARFULL) || 2500 (pool_flags & CEPH_POOL_FLAG_NEARFULL)) 2501 iocb->ki_flags |= IOCB_DSYNC; 2502 written = generic_write_sync(iocb, written); 2503 } 2504 2505 goto out_unlocked; 2506 out_caps: 2507 ceph_put_cap_refs(ci, got); 2508 out: 2509 if (direct_lock) 2510 ceph_end_io_direct(inode); 2511 else 2512 ceph_end_io_write(inode); 2513 out_unlocked: 2514 ceph_free_cap_flush(prealloc_cf); 2515 return written ? written : err; 2516 } 2517 2518 /* 2519 * llseek. be sure to verify file size on SEEK_END. 2520 */ 2521 static loff_t ceph_llseek(struct file *file, loff_t offset, int whence) 2522 { 2523 if (whence == SEEK_END || whence == SEEK_DATA || whence == SEEK_HOLE) { 2524 struct inode *inode = file_inode(file); 2525 int ret; 2526 2527 ret = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE, false); 2528 if (ret < 0) 2529 return ret; 2530 } 2531 return generic_file_llseek(file, offset, whence); 2532 } 2533 2534 static inline void ceph_zero_partial_page( 2535 struct inode *inode, loff_t offset, unsigned size) 2536 { 2537 struct page *page; 2538 pgoff_t index = offset >> PAGE_SHIFT; 2539 2540 page = find_lock_page(inode->i_mapping, index); 2541 if (page) { 2542 wait_on_page_writeback(page); 2543 zero_user(page, offset & (PAGE_SIZE - 1), size); 2544 unlock_page(page); 2545 put_page(page); 2546 } 2547 } 2548 2549 static void ceph_zero_pagecache_range(struct inode *inode, loff_t offset, 2550 loff_t length) 2551 { 2552 loff_t nearly = round_up(offset, PAGE_SIZE); 2553 if (offset < nearly) { 2554 loff_t size = nearly - offset; 2555 if (length < size) 2556 size = length; 2557 ceph_zero_partial_page(inode, offset, size); 2558 offset += size; 2559 length -= size; 2560 } 2561 if (length >= PAGE_SIZE) { 2562 loff_t size = round_down(length, PAGE_SIZE); 2563 truncate_pagecache_range(inode, offset, offset + size - 1); 2564 offset += size; 2565 length -= size; 2566 } 2567 if (length) 2568 ceph_zero_partial_page(inode, offset, length); 2569 } 2570 2571 static int ceph_zero_partial_object(struct inode *inode, 2572 loff_t offset, loff_t *length) 2573 { 2574 struct ceph_inode_info *ci = ceph_inode(inode); 2575 struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); 2576 struct ceph_osd_request *req; 2577 int ret = 0; 2578 loff_t zero = 0; 2579 int op; 2580 2581 if (ceph_inode_is_shutdown(inode)) 2582 return -EIO; 2583 2584 if (!length) { 2585 op = offset ? CEPH_OSD_OP_DELETE : CEPH_OSD_OP_TRUNCATE; 2586 length = &zero; 2587 } else { 2588 op = CEPH_OSD_OP_ZERO; 2589 } 2590 2591 req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout, 2592 ceph_vino(inode), 2593 offset, length, 2594 0, 1, op, 2595 CEPH_OSD_FLAG_WRITE, 2596 NULL, 0, 0, false); 2597 if (IS_ERR(req)) { 2598 ret = PTR_ERR(req); 2599 goto out; 2600 } 2601 2602 req->r_mtime = inode_get_mtime(inode); 2603 ceph_osdc_start_request(&fsc->client->osdc, req); 2604 ret = ceph_osdc_wait_request(&fsc->client->osdc, req); 2605 if (ret == -ENOENT) 2606 ret = 0; 2607 ceph_osdc_put_request(req); 2608 2609 out: 2610 return ret; 2611 } 2612 2613 static int ceph_zero_objects(struct inode *inode, loff_t offset, loff_t length) 2614 { 2615 int ret = 0; 2616 struct ceph_inode_info *ci = ceph_inode(inode); 2617 s32 stripe_unit = ci->i_layout.stripe_unit; 2618 s32 stripe_count = ci->i_layout.stripe_count; 2619 s32 object_size = ci->i_layout.object_size; 2620 u64 object_set_size = object_size * stripe_count; 2621 u64 nearly, t; 2622 2623 /* round offset up to next period boundary */ 2624 nearly = offset + object_set_size - 1; 2625 t = nearly; 2626 nearly -= do_div(t, object_set_size); 2627 2628 while (length && offset < nearly) { 2629 loff_t size = length; 2630 ret = ceph_zero_partial_object(inode, offset, &size); 2631 if (ret < 0) 2632 return ret; 2633 offset += size; 2634 length -= size; 2635 } 2636 while (length >= object_set_size) { 2637 int i; 2638 loff_t pos = offset; 2639 for (i = 0; i < stripe_count; ++i) { 2640 ret = ceph_zero_partial_object(inode, pos, NULL); 2641 if (ret < 0) 2642 return ret; 2643 pos += stripe_unit; 2644 } 2645 offset += object_set_size; 2646 length -= object_set_size; 2647 } 2648 while (length) { 2649 loff_t size = length; 2650 ret = ceph_zero_partial_object(inode, offset, &size); 2651 if (ret < 0) 2652 return ret; 2653 offset += size; 2654 length -= size; 2655 } 2656 return ret; 2657 } 2658 2659 static long ceph_fallocate(struct file *file, int mode, 2660 loff_t offset, loff_t length) 2661 { 2662 struct ceph_file_info *fi = file->private_data; 2663 struct inode *inode = file_inode(file); 2664 struct ceph_inode_info *ci = ceph_inode(inode); 2665 struct ceph_cap_flush *prealloc_cf; 2666 struct ceph_client *cl = ceph_inode_to_client(inode); 2667 int want, got = 0; 2668 int dirty; 2669 int ret = 0; 2670 loff_t endoff = 0; 2671 loff_t size; 2672 2673 doutc(cl, "%p %llx.%llx mode %x, offset %llu length %llu\n", 2674 inode, ceph_vinop(inode), mode, offset, length); 2675 2676 if (mode != (FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE)) 2677 return -EOPNOTSUPP; 2678 2679 if (!S_ISREG(inode->i_mode)) 2680 return -EOPNOTSUPP; 2681 2682 if (IS_ENCRYPTED(inode)) 2683 return -EOPNOTSUPP; 2684 2685 prealloc_cf = ceph_alloc_cap_flush(); 2686 if (!prealloc_cf) 2687 return -ENOMEM; 2688 2689 inode_lock(inode); 2690 2691 if (ceph_snap(inode) != CEPH_NOSNAP) { 2692 ret = -EROFS; 2693 goto unlock; 2694 } 2695 2696 size = i_size_read(inode); 2697 2698 /* Are we punching a hole beyond EOF? */ 2699 if (offset >= size) 2700 goto unlock; 2701 if ((offset + length) > size) 2702 length = size - offset; 2703 2704 if (fi->fmode & CEPH_FILE_MODE_LAZY) 2705 want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO; 2706 else 2707 want = CEPH_CAP_FILE_BUFFER; 2708 2709 ret = ceph_get_caps(file, CEPH_CAP_FILE_WR, want, endoff, &got); 2710 if (ret < 0) 2711 goto unlock; 2712 2713 ret = file_modified(file); 2714 if (ret) 2715 goto put_caps; 2716 2717 filemap_invalidate_lock(inode->i_mapping); 2718 ceph_fscache_invalidate(inode, false); 2719 ceph_zero_pagecache_range(inode, offset, length); 2720 ret = ceph_zero_objects(inode, offset, length); 2721 2722 if (!ret) { 2723 spin_lock(&ci->i_ceph_lock); 2724 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR, 2725 &prealloc_cf); 2726 spin_unlock(&ci->i_ceph_lock); 2727 if (dirty) 2728 __mark_inode_dirty(inode, dirty); 2729 } 2730 filemap_invalidate_unlock(inode->i_mapping); 2731 2732 put_caps: 2733 ceph_put_cap_refs(ci, got); 2734 unlock: 2735 inode_unlock(inode); 2736 ceph_free_cap_flush(prealloc_cf); 2737 return ret; 2738 } 2739 2740 /* 2741 * This function tries to get FILE_WR capabilities for dst_ci and FILE_RD for 2742 * src_ci. Two attempts are made to obtain both caps, and an error is return if 2743 * this fails; zero is returned on success. 2744 */ 2745 static int get_rd_wr_caps(struct file *src_filp, int *src_got, 2746 struct file *dst_filp, 2747 loff_t dst_endoff, int *dst_got) 2748 { 2749 int ret = 0; 2750 bool retrying = false; 2751 2752 retry_caps: 2753 ret = ceph_get_caps(dst_filp, CEPH_CAP_FILE_WR, CEPH_CAP_FILE_BUFFER, 2754 dst_endoff, dst_got); 2755 if (ret < 0) 2756 return ret; 2757 2758 /* 2759 * Since we're already holding the FILE_WR capability for the dst file, 2760 * we would risk a deadlock by using ceph_get_caps. Thus, we'll do some 2761 * retry dance instead to try to get both capabilities. 2762 */ 2763 ret = ceph_try_get_caps(file_inode(src_filp), 2764 CEPH_CAP_FILE_RD, CEPH_CAP_FILE_SHARED, 2765 false, src_got); 2766 if (ret <= 0) { 2767 /* Start by dropping dst_ci caps and getting src_ci caps */ 2768 ceph_put_cap_refs(ceph_inode(file_inode(dst_filp)), *dst_got); 2769 if (retrying) { 2770 if (!ret) 2771 /* ceph_try_get_caps masks EAGAIN */ 2772 ret = -EAGAIN; 2773 return ret; 2774 } 2775 ret = ceph_get_caps(src_filp, CEPH_CAP_FILE_RD, 2776 CEPH_CAP_FILE_SHARED, -1, src_got); 2777 if (ret < 0) 2778 return ret; 2779 /*... drop src_ci caps too, and retry */ 2780 ceph_put_cap_refs(ceph_inode(file_inode(src_filp)), *src_got); 2781 retrying = true; 2782 goto retry_caps; 2783 } 2784 return ret; 2785 } 2786 2787 static void put_rd_wr_caps(struct ceph_inode_info *src_ci, int src_got, 2788 struct ceph_inode_info *dst_ci, int dst_got) 2789 { 2790 ceph_put_cap_refs(src_ci, src_got); 2791 ceph_put_cap_refs(dst_ci, dst_got); 2792 } 2793 2794 /* 2795 * This function does several size-related checks, returning an error if: 2796 * - source file is smaller than off+len 2797 * - destination file size is not OK (inode_newsize_ok()) 2798 * - max bytes quotas is exceeded 2799 */ 2800 static int is_file_size_ok(struct inode *src_inode, struct inode *dst_inode, 2801 loff_t src_off, loff_t dst_off, size_t len) 2802 { 2803 struct ceph_client *cl = ceph_inode_to_client(src_inode); 2804 loff_t size, endoff; 2805 2806 size = i_size_read(src_inode); 2807 /* 2808 * Don't copy beyond source file EOF. Instead of simply setting length 2809 * to (size - src_off), just drop to VFS default implementation, as the 2810 * local i_size may be stale due to other clients writing to the source 2811 * inode. 2812 */ 2813 if (src_off + len > size) { 2814 doutc(cl, "Copy beyond EOF (%llu + %zu > %llu)\n", src_off, 2815 len, size); 2816 return -EOPNOTSUPP; 2817 } 2818 size = i_size_read(dst_inode); 2819 2820 endoff = dst_off + len; 2821 if (inode_newsize_ok(dst_inode, endoff)) 2822 return -EOPNOTSUPP; 2823 2824 if (ceph_quota_is_max_bytes_exceeded(dst_inode, endoff)) 2825 return -EDQUOT; 2826 2827 return 0; 2828 } 2829 2830 static struct ceph_osd_request * 2831 ceph_alloc_copyfrom_request(struct ceph_osd_client *osdc, 2832 u64 src_snapid, 2833 struct ceph_object_id *src_oid, 2834 struct ceph_object_locator *src_oloc, 2835 struct ceph_object_id *dst_oid, 2836 struct ceph_object_locator *dst_oloc, 2837 u32 truncate_seq, u64 truncate_size) 2838 { 2839 struct ceph_osd_request *req; 2840 int ret; 2841 u32 src_fadvise_flags = 2842 CEPH_OSD_OP_FLAG_FADVISE_SEQUENTIAL | 2843 CEPH_OSD_OP_FLAG_FADVISE_NOCACHE; 2844 u32 dst_fadvise_flags = 2845 CEPH_OSD_OP_FLAG_FADVISE_SEQUENTIAL | 2846 CEPH_OSD_OP_FLAG_FADVISE_DONTNEED; 2847 2848 req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_KERNEL); 2849 if (!req) 2850 return ERR_PTR(-ENOMEM); 2851 2852 req->r_flags = CEPH_OSD_FLAG_WRITE; 2853 2854 ceph_oloc_copy(&req->r_t.base_oloc, dst_oloc); 2855 ceph_oid_copy(&req->r_t.base_oid, dst_oid); 2856 2857 ret = osd_req_op_copy_from_init(req, src_snapid, 0, 2858 src_oid, src_oloc, 2859 src_fadvise_flags, 2860 dst_fadvise_flags, 2861 truncate_seq, 2862 truncate_size, 2863 CEPH_OSD_COPY_FROM_FLAG_TRUNCATE_SEQ); 2864 if (ret) 2865 goto out; 2866 2867 ret = ceph_osdc_alloc_messages(req, GFP_KERNEL); 2868 if (ret) 2869 goto out; 2870 2871 return req; 2872 2873 out: 2874 ceph_osdc_put_request(req); 2875 return ERR_PTR(ret); 2876 } 2877 2878 static ssize_t ceph_do_objects_copy(struct ceph_inode_info *src_ci, u64 *src_off, 2879 struct ceph_inode_info *dst_ci, u64 *dst_off, 2880 struct ceph_fs_client *fsc, 2881 size_t len, unsigned int flags) 2882 { 2883 struct ceph_object_locator src_oloc, dst_oloc; 2884 struct ceph_object_id src_oid, dst_oid; 2885 struct ceph_osd_client *osdc; 2886 struct ceph_osd_request *req; 2887 size_t bytes = 0; 2888 u64 src_objnum, src_objoff, dst_objnum, dst_objoff; 2889 u32 src_objlen, dst_objlen; 2890 u32 object_size = src_ci->i_layout.object_size; 2891 struct ceph_client *cl = fsc->client; 2892 int ret; 2893 2894 src_oloc.pool = src_ci->i_layout.pool_id; 2895 src_oloc.pool_ns = ceph_try_get_string(src_ci->i_layout.pool_ns); 2896 dst_oloc.pool = dst_ci->i_layout.pool_id; 2897 dst_oloc.pool_ns = ceph_try_get_string(dst_ci->i_layout.pool_ns); 2898 osdc = &fsc->client->osdc; 2899 2900 while (len >= object_size) { 2901 ceph_calc_file_object_mapping(&src_ci->i_layout, *src_off, 2902 object_size, &src_objnum, 2903 &src_objoff, &src_objlen); 2904 ceph_calc_file_object_mapping(&dst_ci->i_layout, *dst_off, 2905 object_size, &dst_objnum, 2906 &dst_objoff, &dst_objlen); 2907 ceph_oid_init(&src_oid); 2908 ceph_oid_printf(&src_oid, "%llx.%08llx", 2909 src_ci->i_vino.ino, src_objnum); 2910 ceph_oid_init(&dst_oid); 2911 ceph_oid_printf(&dst_oid, "%llx.%08llx", 2912 dst_ci->i_vino.ino, dst_objnum); 2913 /* Do an object remote copy */ 2914 req = ceph_alloc_copyfrom_request(osdc, src_ci->i_vino.snap, 2915 &src_oid, &src_oloc, 2916 &dst_oid, &dst_oloc, 2917 dst_ci->i_truncate_seq, 2918 dst_ci->i_truncate_size); 2919 if (IS_ERR(req)) 2920 ret = PTR_ERR(req); 2921 else { 2922 ceph_osdc_start_request(osdc, req); 2923 ret = ceph_osdc_wait_request(osdc, req); 2924 ceph_update_copyfrom_metrics(&fsc->mdsc->metric, 2925 req->r_start_latency, 2926 req->r_end_latency, 2927 object_size, ret); 2928 ceph_osdc_put_request(req); 2929 } 2930 if (ret) { 2931 if (ret == -EOPNOTSUPP) { 2932 fsc->have_copy_from2 = false; 2933 pr_notice_client(cl, 2934 "OSDs don't support copy-from2; disabling copy offload\n"); 2935 } 2936 doutc(cl, "returned %d\n", ret); 2937 if (!bytes) 2938 bytes = ret; 2939 goto out; 2940 } 2941 len -= object_size; 2942 bytes += object_size; 2943 *src_off += object_size; 2944 *dst_off += object_size; 2945 } 2946 2947 out: 2948 ceph_oloc_destroy(&src_oloc); 2949 ceph_oloc_destroy(&dst_oloc); 2950 return bytes; 2951 } 2952 2953 static ssize_t __ceph_copy_file_range(struct file *src_file, loff_t src_off, 2954 struct file *dst_file, loff_t dst_off, 2955 size_t len, unsigned int flags) 2956 { 2957 struct inode *src_inode = file_inode(src_file); 2958 struct inode *dst_inode = file_inode(dst_file); 2959 struct ceph_inode_info *src_ci = ceph_inode(src_inode); 2960 struct ceph_inode_info *dst_ci = ceph_inode(dst_inode); 2961 struct ceph_cap_flush *prealloc_cf; 2962 struct ceph_fs_client *src_fsc = ceph_inode_to_fs_client(src_inode); 2963 struct ceph_client *cl = src_fsc->client; 2964 loff_t size; 2965 ssize_t ret = -EIO, bytes; 2966 u64 src_objnum, dst_objnum, src_objoff, dst_objoff; 2967 u32 src_objlen, dst_objlen; 2968 int src_got = 0, dst_got = 0, err, dirty; 2969 2970 if (src_inode->i_sb != dst_inode->i_sb) { 2971 struct ceph_fs_client *dst_fsc = ceph_inode_to_fs_client(dst_inode); 2972 2973 if (ceph_fsid_compare(&src_fsc->client->fsid, 2974 &dst_fsc->client->fsid)) { 2975 dout("Copying files across clusters: src: %pU dst: %pU\n", 2976 &src_fsc->client->fsid, &dst_fsc->client->fsid); 2977 return -EXDEV; 2978 } 2979 } 2980 if (ceph_snap(dst_inode) != CEPH_NOSNAP) 2981 return -EROFS; 2982 2983 /* 2984 * Some of the checks below will return -EOPNOTSUPP, which will force a 2985 * fallback to the default VFS copy_file_range implementation. This is 2986 * desirable in several cases (for ex, the 'len' is smaller than the 2987 * size of the objects, or in cases where that would be more 2988 * efficient). 2989 */ 2990 2991 if (ceph_test_mount_opt(src_fsc, NOCOPYFROM)) 2992 return -EOPNOTSUPP; 2993 2994 if (!src_fsc->have_copy_from2) 2995 return -EOPNOTSUPP; 2996 2997 /* 2998 * Striped file layouts require that we copy partial objects, but the 2999 * OSD copy-from operation only supports full-object copies. Limit 3000 * this to non-striped file layouts for now. 3001 */ 3002 if ((src_ci->i_layout.stripe_unit != dst_ci->i_layout.stripe_unit) || 3003 (src_ci->i_layout.stripe_count != 1) || 3004 (dst_ci->i_layout.stripe_count != 1) || 3005 (src_ci->i_layout.object_size != dst_ci->i_layout.object_size)) { 3006 doutc(cl, "Invalid src/dst files layout\n"); 3007 return -EOPNOTSUPP; 3008 } 3009 3010 /* Every encrypted inode gets its own key, so we can't offload them */ 3011 if (IS_ENCRYPTED(src_inode) || IS_ENCRYPTED(dst_inode)) 3012 return -EOPNOTSUPP; 3013 3014 if (len < src_ci->i_layout.object_size) 3015 return -EOPNOTSUPP; /* no remote copy will be done */ 3016 3017 prealloc_cf = ceph_alloc_cap_flush(); 3018 if (!prealloc_cf) 3019 return -ENOMEM; 3020 3021 /* Start by sync'ing the source and destination files */ 3022 ret = file_write_and_wait_range(src_file, src_off, (src_off + len)); 3023 if (ret < 0) { 3024 doutc(cl, "failed to write src file (%zd)\n", ret); 3025 goto out; 3026 } 3027 ret = file_write_and_wait_range(dst_file, dst_off, (dst_off + len)); 3028 if (ret < 0) { 3029 doutc(cl, "failed to write dst file (%zd)\n", ret); 3030 goto out; 3031 } 3032 3033 /* 3034 * We need FILE_WR caps for dst_ci and FILE_RD for src_ci as other 3035 * clients may have dirty data in their caches. And OSDs know nothing 3036 * about caps, so they can't safely do the remote object copies. 3037 */ 3038 err = get_rd_wr_caps(src_file, &src_got, 3039 dst_file, (dst_off + len), &dst_got); 3040 if (err < 0) { 3041 doutc(cl, "get_rd_wr_caps returned %d\n", err); 3042 ret = -EOPNOTSUPP; 3043 goto out; 3044 } 3045 3046 ret = is_file_size_ok(src_inode, dst_inode, src_off, dst_off, len); 3047 if (ret < 0) 3048 goto out_caps; 3049 3050 /* Drop dst file cached pages */ 3051 ceph_fscache_invalidate(dst_inode, false); 3052 ret = invalidate_inode_pages2_range(dst_inode->i_mapping, 3053 dst_off >> PAGE_SHIFT, 3054 (dst_off + len) >> PAGE_SHIFT); 3055 if (ret < 0) { 3056 doutc(cl, "Failed to invalidate inode pages (%zd)\n", 3057 ret); 3058 ret = 0; /* XXX */ 3059 } 3060 ceph_calc_file_object_mapping(&src_ci->i_layout, src_off, 3061 src_ci->i_layout.object_size, 3062 &src_objnum, &src_objoff, &src_objlen); 3063 ceph_calc_file_object_mapping(&dst_ci->i_layout, dst_off, 3064 dst_ci->i_layout.object_size, 3065 &dst_objnum, &dst_objoff, &dst_objlen); 3066 /* object-level offsets need to the same */ 3067 if (src_objoff != dst_objoff) { 3068 ret = -EOPNOTSUPP; 3069 goto out_caps; 3070 } 3071 3072 /* 3073 * Do a manual copy if the object offset isn't object aligned. 3074 * 'src_objlen' contains the bytes left until the end of the object, 3075 * starting at the src_off 3076 */ 3077 if (src_objoff) { 3078 doutc(cl, "Initial partial copy of %u bytes\n", src_objlen); 3079 3080 /* 3081 * we need to temporarily drop all caps as we'll be calling 3082 * {read,write}_iter, which will get caps again. 3083 */ 3084 put_rd_wr_caps(src_ci, src_got, dst_ci, dst_got); 3085 ret = splice_file_range(src_file, &src_off, dst_file, &dst_off, 3086 src_objlen); 3087 /* Abort on short copies or on error */ 3088 if (ret < (long)src_objlen) { 3089 doutc(cl, "Failed partial copy (%zd)\n", ret); 3090 goto out; 3091 } 3092 len -= ret; 3093 err = get_rd_wr_caps(src_file, &src_got, 3094 dst_file, (dst_off + len), &dst_got); 3095 if (err < 0) 3096 goto out; 3097 err = is_file_size_ok(src_inode, dst_inode, 3098 src_off, dst_off, len); 3099 if (err < 0) 3100 goto out_caps; 3101 } 3102 3103 size = i_size_read(dst_inode); 3104 bytes = ceph_do_objects_copy(src_ci, &src_off, dst_ci, &dst_off, 3105 src_fsc, len, flags); 3106 if (bytes <= 0) { 3107 if (!ret) 3108 ret = bytes; 3109 goto out_caps; 3110 } 3111 doutc(cl, "Copied %zu bytes out of %zu\n", bytes, len); 3112 len -= bytes; 3113 ret += bytes; 3114 3115 file_update_time(dst_file); 3116 inode_inc_iversion_raw(dst_inode); 3117 3118 if (dst_off > size) { 3119 /* Let the MDS know about dst file size change */ 3120 if (ceph_inode_set_size(dst_inode, dst_off) || 3121 ceph_quota_is_max_bytes_approaching(dst_inode, dst_off)) 3122 ceph_check_caps(dst_ci, CHECK_CAPS_AUTHONLY | CHECK_CAPS_FLUSH); 3123 } 3124 /* Mark Fw dirty */ 3125 spin_lock(&dst_ci->i_ceph_lock); 3126 dirty = __ceph_mark_dirty_caps(dst_ci, CEPH_CAP_FILE_WR, &prealloc_cf); 3127 spin_unlock(&dst_ci->i_ceph_lock); 3128 if (dirty) 3129 __mark_inode_dirty(dst_inode, dirty); 3130 3131 out_caps: 3132 put_rd_wr_caps(src_ci, src_got, dst_ci, dst_got); 3133 3134 /* 3135 * Do the final manual copy if we still have some bytes left, unless 3136 * there were errors in remote object copies (len >= object_size). 3137 */ 3138 if (len && (len < src_ci->i_layout.object_size)) { 3139 doutc(cl, "Final partial copy of %zu bytes\n", len); 3140 bytes = splice_file_range(src_file, &src_off, dst_file, 3141 &dst_off, len); 3142 if (bytes > 0) 3143 ret += bytes; 3144 else 3145 doutc(cl, "Failed partial copy (%zd)\n", bytes); 3146 } 3147 3148 out: 3149 ceph_free_cap_flush(prealloc_cf); 3150 3151 return ret; 3152 } 3153 3154 static ssize_t ceph_copy_file_range(struct file *src_file, loff_t src_off, 3155 struct file *dst_file, loff_t dst_off, 3156 size_t len, unsigned int flags) 3157 { 3158 ssize_t ret; 3159 3160 ret = __ceph_copy_file_range(src_file, src_off, dst_file, dst_off, 3161 len, flags); 3162 3163 if (ret == -EOPNOTSUPP || ret == -EXDEV) 3164 ret = splice_copy_file_range(src_file, src_off, dst_file, 3165 dst_off, len); 3166 return ret; 3167 } 3168 3169 const struct file_operations ceph_file_fops = { 3170 .open = ceph_open, 3171 .release = ceph_release, 3172 .llseek = ceph_llseek, 3173 .read_iter = ceph_read_iter, 3174 .write_iter = ceph_write_iter, 3175 .mmap = ceph_mmap, 3176 .fsync = ceph_fsync, 3177 .lock = ceph_lock, 3178 .setlease = simple_nosetlease, 3179 .flock = ceph_flock, 3180 .splice_read = ceph_splice_read, 3181 .splice_write = iter_file_splice_write, 3182 .unlocked_ioctl = ceph_ioctl, 3183 .compat_ioctl = compat_ptr_ioctl, 3184 .fallocate = ceph_fallocate, 3185 .copy_file_range = ceph_copy_file_range, 3186 }; 3187