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