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