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