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