fs/ceph/file.c

   1 #include <linux/ceph/ceph_debug.h>
   2
   3 #include <linux/module.h>
   4 #include <linux/sched.h>
   5 #include <linux/slab.h>
   6 #include <linux/file.h>
   7 #include <linux/mount.h>
   8 #include <linux/namei.h>
   9 #include <linux/writeback.h>
  10 #include <linux/falloc.h>
  11
  12 #include "super.h"
  13 #include "mds_client.h"
  14 #include "cache.h"
  15
  16 /*
  17  * Ceph file operations
  18  *
  19  * Implement basic open/close functionality, and implement
  20  * read/write.
  21  *
  22  * We implement three modes of file I/O:
  23  *  - buffered uses the generic_file_aio_{read,write} helpers
  24  *
  25  *  - synchronous is used when there is multi-client read/write
  26  *    sharing, avoids the page cache, and synchronously waits for an
  27  *    ack from the OSD.
  28  *
  29  *  - direct io takes the variant of the sync path that references
  30  *    user pages directly.
  31  *
  32  * fsync() flushes and waits on dirty pages, but just queues metadata
  33  * for writeback: since the MDS can recover size and mtime there is no
  34  * need to wait for MDS acknowledgement.
  35  */
  36
  37
  38 /*
  39  * Prepare an open request.  Preallocate ceph_cap to avoid an
  40  * inopportune ENOMEM later.
  41  */
  42 static struct ceph_mds_request *
  43 prepare_open_request(struct super_block *sb, int flags, int create_mode)
  44 {
  45         struct ceph_fs_client *fsc = ceph_sb_to_client(sb);
  46         struct ceph_mds_client *mdsc = fsc->mdsc;
  47         struct ceph_mds_request *req;
  48         int want_auth = USE_ANY_MDS;
  49         int op = (flags & O_CREAT) ? CEPH_MDS_OP_CREATE : CEPH_MDS_OP_OPEN;
  50
  51         if (flags & (O_WRONLY|O_RDWR|O_CREAT|O_TRUNC))
  52                 want_auth = USE_AUTH_MDS;
  53
  54         req = ceph_mdsc_create_request(mdsc, op, want_auth);
  55         if (IS_ERR(req))
  56                 goto out;
  57         req->r_fmode = ceph_flags_to_mode(flags);
  58         req->r_args.open.flags = cpu_to_le32(flags);
  59         req->r_args.open.mode = cpu_to_le32(create_mode);
  60 out:
  61         return req;
  62 }
  63
  64 /*
  65  * initialize private struct file data.
  66  * if we fail, clean up by dropping fmode reference on the ceph_inode
  67  */
  68 static int ceph_init_file(struct inode *inode, struct file *file, int fmode)
  69 {
  70         struct ceph_file_info *cf;
  71         int ret = 0;
  72         struct ceph_inode_info *ci = ceph_inode(inode);
  73         struct ceph_fs_client *fsc = ceph_sb_to_client(inode->i_sb);
  74         struct ceph_mds_client *mdsc = fsc->mdsc;
  75
  76         switch (inode->i_mode & S_IFMT) {
  77         case S_IFREG:
  78                 /* First file open request creates the cookie, we want to keep
  79                  * this cookie around for the filetime of the inode as not to
  80                  * have to worry about fscache register / revoke / operation
  81                  * races.
  82                  *
  83                  * Also, if we know the operation is going to invalidate data
  84                  * (non readonly) just nuke the cache right away.
  85                  */
  86                 ceph_fscache_register_inode_cookie(mdsc->fsc, ci);
  87                 if ((fmode & CEPH_FILE_MODE_WR))
  88                         ceph_fscache_invalidate(inode);
  89         case S_IFDIR:
  90                 dout("init_file %p %p 0%o (regular)\n", inode, file,
  91                      inode->i_mode);
  92                 cf = kmem_cache_alloc(ceph_file_cachep, GFP_KERNEL | __GFP_ZERO);
  93                 if (cf == NULL) {
  94                         ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */
  95                         return -ENOMEM;
  96                 }
  97                 cf->fmode = fmode;
  98                 cf->next_offset = 2;
  99                 cf->readdir_cache_idx = -1;
 100                 file->private_data = cf;
 101                 BUG_ON(inode->i_fop->release != ceph_release);
 102                 break;
 103
 104         case S_IFLNK:
 105                 dout("init_file %p %p 0%o (symlink)\n", inode, file,
 106                      inode->i_mode);
 107                 ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */
 108                 break;
 109
 110         default:
 111                 dout("init_file %p %p 0%o (special)\n", inode, file,
 112                      inode->i_mode);
 113                 /*
 114                  * we need to drop the open ref now, since we don't
 115                  * have .release set to ceph_release.
 116                  */
 117                 ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */
 118                 BUG_ON(inode->i_fop->release == ceph_release);
 119
 120                 /* call the proper open fop */
 121                 ret = inode->i_fop->open(inode, file);
 122         }
 123         return ret;
 124 }
 125
 126 /*
 127  * If we already have the requisite capabilities, we can satisfy
 128  * the open request locally (no need to request new caps from the
 129  * MDS).  We do, however, need to inform the MDS (asynchronously)
 130  * if our wanted caps set expands.
 131  */
 132 int ceph_open(struct inode *inode, struct file *file)
 133 {
 134         struct ceph_inode_info *ci = ceph_inode(inode);
 135         struct ceph_fs_client *fsc = ceph_sb_to_client(inode->i_sb);
 136         struct ceph_mds_client *mdsc = fsc->mdsc;
 137         struct ceph_mds_request *req;
 138         struct ceph_file_info *cf = file->private_data;
 139         int err;
 140         int flags, fmode, wanted;
 141
 142         if (cf) {
 143                 dout("open file %p is already opened\n", file);
 144                 return 0;
 145         }
 146
 147         /* filter out O_CREAT|O_EXCL; vfs did that already.  yuck. */
 148         flags = file->f_flags & ~(O_CREAT|O_EXCL);
 149         if (S_ISDIR(inode->i_mode))
 150                 flags = O_DIRECTORY;  /* mds likes to know */
 151
 152         dout("open inode %p ino %llx.%llx file %p flags %d (%d)\n", inode,
 153              ceph_vinop(inode), file, flags, file->f_flags);
 154         fmode = ceph_flags_to_mode(flags);
 155         wanted = ceph_caps_for_mode(fmode);
 156
 157         /* snapped files are read-only */
 158         if (ceph_snap(inode) != CEPH_NOSNAP && (file->f_mode & FMODE_WRITE))
 159                 return -EROFS;
 160
 161         /* trivially open snapdir */
 162         if (ceph_snap(inode) == CEPH_SNAPDIR) {
 163                 spin_lock(&ci->i_ceph_lock);
 164                 __ceph_get_fmode(ci, fmode);
 165                 spin_unlock(&ci->i_ceph_lock);
 166                 return ceph_init_file(inode, file, fmode);
 167         }
 168
 169         /*
 170          * No need to block if we have caps on the auth MDS (for
 171          * write) or any MDS (for read).  Update wanted set
 172          * asynchronously.
 173          */
 174         spin_lock(&ci->i_ceph_lock);
 175         if (__ceph_is_any_real_caps(ci) &&
 176             (((fmode & CEPH_FILE_MODE_WR) == 0) || ci->i_auth_cap)) {
 177                 int mds_wanted = __ceph_caps_mds_wanted(ci);
 178                 int issued = __ceph_caps_issued(ci, NULL);
 179
 180                 dout("open %p fmode %d want %s issued %s using existing\n",
 181                      inode, fmode, ceph_cap_string(wanted),
 182                      ceph_cap_string(issued));
 183                 __ceph_get_fmode(ci, fmode);
 184                 spin_unlock(&ci->i_ceph_lock);
 185
 186                 /* adjust wanted? */
 187                 if ((issued & wanted) != wanted &&
 188                     (mds_wanted & wanted) != wanted &&
 189                     ceph_snap(inode) != CEPH_SNAPDIR)
 190                         ceph_check_caps(ci, 0, NULL);
 191
 192                 return ceph_init_file(inode, file, fmode);
 193         } else if (ceph_snap(inode) != CEPH_NOSNAP &&
 194                    (ci->i_snap_caps & wanted) == wanted) {
 195                 __ceph_get_fmode(ci, fmode);
 196                 spin_unlock(&ci->i_ceph_lock);
 197                 return ceph_init_file(inode, file, fmode);
 198         }
 199
 200         spin_unlock(&ci->i_ceph_lock);
 201
 202         dout("open fmode %d wants %s\n", fmode, ceph_cap_string(wanted));
 203         req = prepare_open_request(inode->i_sb, flags, 0);
 204         if (IS_ERR(req)) {
 205                 err = PTR_ERR(req);
 206                 goto out;
 207         }
 208         req->r_inode = inode;
 209         ihold(inode);
 210
 211         req->r_num_caps = 1;
 212         err = ceph_mdsc_do_request(mdsc, NULL, req);
 213         if (!err)
 214                 err = ceph_init_file(inode, file, req->r_fmode);
 215         ceph_mdsc_put_request(req);
 216         dout("open result=%d on %llx.%llx\n", err, ceph_vinop(inode));
 217 out:
 218         return err;
 219 }
 220
 221
 222 /*
 223  * Do a lookup + open with a single request.  If we get a non-existent
 224  * file or symlink, return 1 so the VFS can retry.
 225  */
 226 int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
 227                      struct file *file, unsigned flags, umode_t mode,
 228                      int *opened)
 229 {
 230         struct ceph_fs_client *fsc = ceph_sb_to_client(dir->i_sb);
 231         struct ceph_mds_client *mdsc = fsc->mdsc;
 232         struct ceph_mds_request *req;
 233         struct dentry *dn;
 234         struct ceph_acls_info acls = {};
 235         int err;
 236
 237         dout("atomic_open %p dentry %p '%pd' %s flags %d mode 0%o\n",
 238              dir, dentry, dentry,
 239              d_unhashed(dentry) ? "unhashed" : "hashed", flags, mode);
 240
 241         if (dentry->d_name.len > NAME_MAX)
 242                 return -ENAMETOOLONG;
 243
 244         err = ceph_init_dentry(dentry);
 245         if (err < 0)
 246                 return err;
 247
 248         if (flags & O_CREAT) {
 249                 err = ceph_pre_init_acls(dir, &mode, &acls);
 250                 if (err < 0)
 251                         return err;
 252         }
 253
 254         /* do the open */
 255         req = prepare_open_request(dir->i_sb, flags, mode);
 256         if (IS_ERR(req)) {
 257                 err = PTR_ERR(req);
 258                 goto out_acl;
 259         }
 260         req->r_dentry = dget(dentry);
 261         req->r_num_caps = 2;
 262         if (flags & O_CREAT) {
 263                 req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
 264                 req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
 265                 if (acls.pagelist) {
 266                         req->r_pagelist = acls.pagelist;
 267                         acls.pagelist = NULL;
 268                 }
 269         }
 270         req->r_locked_dir = dir;           /* caller holds dir->i_mutex */
 271         err = ceph_mdsc_do_request(mdsc,
 272                                    (flags & (O_CREAT|O_TRUNC)) ? dir : NULL,
 273                                    req);
 274         err = ceph_handle_snapdir(req, dentry, err);
 275         if (err)
 276                 goto out_req;
 277
 278         if ((flags & O_CREAT) && !req->r_reply_info.head->is_dentry)
 279                 err = ceph_handle_notrace_create(dir, dentry);
 280
 281         if (d_unhashed(dentry)) {
 282                 dn = ceph_finish_lookup(req, dentry, err);
 283                 if (IS_ERR(dn))
 284                         err = PTR_ERR(dn);
 285         } else {
 286                 /* we were given a hashed negative dentry */
 287                 dn = NULL;
 288         }
 289         if (err)
 290                 goto out_req;
 291         if (dn || d_really_is_negative(dentry) || d_is_symlink(dentry)) {
 292                 /* make vfs retry on splice, ENOENT, or symlink */
 293                 dout("atomic_open finish_no_open on dn %p\n", dn);
 294                 err = finish_no_open(file, dn);
 295         } else {
 296                 dout("atomic_open finish_open on dn %p\n", dn);
 297                 if (req->r_op == CEPH_MDS_OP_CREATE && req->r_reply_info.has_create_ino) {
 298                         ceph_init_inode_acls(d_inode(dentry), &acls);
 299                         *opened |= FILE_CREATED;
 300                 }
 301                 err = finish_open(file, dentry, ceph_open, opened);
 302         }
 303 out_req:
 304         if (!req->r_err && req->r_target_inode)
 305                 ceph_put_fmode(ceph_inode(req->r_target_inode), req->r_fmode);
 306         ceph_mdsc_put_request(req);
 307 out_acl:
 308         ceph_release_acls_info(&acls);
 309         dout("atomic_open result=%d\n", err);
 310         return err;
 311 }
 312
 313 int ceph_release(struct inode *inode, struct file *file)
 314 {
 315         struct ceph_inode_info *ci = ceph_inode(inode);
 316         struct ceph_file_info *cf = file->private_data;
 317
 318         dout("release inode %p file %p\n", inode, file);
 319         ceph_put_fmode(ci, cf->fmode);
 320         if (cf->last_readdir)
 321                 ceph_mdsc_put_request(cf->last_readdir);
 322         kfree(cf->last_name);
 323         kfree(cf->dir_info);
 324         kmem_cache_free(ceph_file_cachep, cf);
 325
 326         /* wake up anyone waiting for caps on this inode */
 327         wake_up_all(&ci->i_cap_wq);
 328         return 0;
 329 }
 330
 331 enum {
 332         CHECK_EOF = 1,
 333         READ_INLINE = 2,
 334 };
 335
 336 /*
 337  * Read a range of bytes striped over one or more objects.  Iterate over
 338  * objects we stripe over.  (That's not atomic, but good enough for now.)
 339  *
 340  * If we get a short result from the OSD, check against i_size; we need to
 341  * only return a short read to the caller if we hit EOF.
 342  */
 343 static int striped_read(struct inode *inode,
 344                         u64 off, u64 len,
 345                         struct page **pages, int num_pages,
 346                         int *checkeof, bool o_direct,
 347                         unsigned long buf_align)
 348 {
 349         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
 350         struct ceph_inode_info *ci = ceph_inode(inode);
 351         u64 pos, this_len, left;
 352         int io_align, page_align;
 353         int pages_left;
 354         int read;
 355         struct page **page_pos;
 356         int ret;
 357         bool hit_stripe, was_short;
 358
 359         /*
 360          * we may need to do multiple reads.  not atomic, unfortunately.
 361          */
 362         pos = off;
 363         left = len;
 364         page_pos = pages;
 365         pages_left = num_pages;
 366         read = 0;
 367         io_align = off & ~PAGE_MASK;
 368
 369 more:
 370         if (o_direct)
 371                 page_align = (pos - io_align + buf_align) & ~PAGE_MASK;
 372         else
 373                 page_align = pos & ~PAGE_MASK;
 374         this_len = left;
 375         ret = ceph_osdc_readpages(&fsc->client->osdc, ceph_vino(inode),
 376                                   &ci->i_layout, pos, &this_len,
 377                                   ci->i_truncate_seq,
 378                                   ci->i_truncate_size,
 379                                   page_pos, pages_left, page_align);
 380         if (ret == -ENOENT)
 381                 ret = 0;
 382         hit_stripe = this_len < left;
 383         was_short = ret >= 0 && ret < this_len;
 384         dout("striped_read %llu~%llu (read %u) got %d%s%s\n", pos, left, read,
 385              ret, hit_stripe ? " HITSTRIPE" : "", was_short ? " SHORT" : "");
 386
 387         if (ret >= 0) {
 388                 int didpages;
 389                 if (was_short && (pos + ret < inode->i_size)) {
 390                         int zlen = min(this_len - ret,
 391                                        inode->i_size - pos - ret);
 392                         int zoff = (o_direct ? buf_align : io_align) +
 393                                     read + ret;
 394                         dout(" zero gap %llu to %llu\n",
 395                                 pos + ret, pos + ret + zlen);
 396                         ceph_zero_page_vector_range(zoff, zlen, pages);
 397                         ret += zlen;
 398                 }
 399
 400                 didpages = (page_align + ret) >> PAGE_CACHE_SHIFT;
 401                 pos += ret;
 402                 read = pos - off;
 403                 left -= ret;
 404                 page_pos += didpages;
 405                 pages_left -= didpages;
 406
 407                 /* hit stripe and need continue*/
 408                 if (left && hit_stripe && pos < inode->i_size)
 409                         goto more;
 410         }
 411
 412         if (read > 0) {
 413                 ret = read;
 414                 /* did we bounce off eof? */
 415                 if (pos + left > inode->i_size)
 416                         *checkeof = CHECK_EOF;
 417         }
 418
 419         dout("striped_read returns %d\n", ret);
 420         return ret;
 421 }
 422
 423 /*
 424  * Completely synchronous read and write methods.  Direct from __user
 425  * buffer to osd, or directly to user pages (if O_DIRECT).
 426  *
 427  * If the read spans object boundary, just do multiple reads.
 428  */
 429 static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *i,
 430                                 int *checkeof)
 431 {
 432         struct file *file = iocb->ki_filp;
 433         struct inode *inode = file_inode(file);
 434         struct page **pages;
 435         u64 off = iocb->ki_pos;
 436         int num_pages, ret;
 437         size_t len = iov_iter_count(i);
 438
 439         dout("sync_read on file %p %llu~%u %s\n", file, off,
 440              (unsigned)len,
 441              (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
 442
 443         if (!len)
 444                 return 0;
 445         /*
 446          * flush any page cache pages in this range.  this
 447          * will make concurrent normal and sync io slow,
 448          * but it will at least behave sensibly when they are
 449          * in sequence.
 450          */
 451         ret = filemap_write_and_wait_range(inode->i_mapping, off,
 452                                                 off + len);
 453         if (ret < 0)
 454                 return ret;
 455
 456         if (iocb->ki_flags & IOCB_DIRECT) {
 457                 while (iov_iter_count(i)) {
 458                         size_t start;
 459                         ssize_t n;
 460
 461                         n = iov_iter_get_pages_alloc(i, &pages, INT_MAX, &start);
 462                         if (n < 0)
 463                                 return n;
 464
 465                         num_pages = (n + start + PAGE_SIZE - 1) / PAGE_SIZE;
 466
 467                         ret = striped_read(inode, off, n,
 468                                            pages, num_pages, checkeof,
 469                                            1, start);
 470
 471                         ceph_put_page_vector(pages, num_pages, true);
 472
 473                         if (ret <= 0)
 474                                 break;
 475                         off += ret;
 476                         iov_iter_advance(i, ret);
 477                         if (ret < n)
 478                                 break;
 479                 }
 480         } else {
 481                 num_pages = calc_pages_for(off, len);
 482                 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
 483                 if (IS_ERR(pages))
 484                         return PTR_ERR(pages);
 485                 ret = striped_read(inode, off, len, pages,
 486                                         num_pages, checkeof, 0, 0);
 487                 if (ret > 0) {
 488                         int l, k = 0;
 489                         size_t left = ret;
 490
 491                         while (left) {
 492                                 size_t page_off = off & ~PAGE_MASK;
 493                                 size_t copy = min_t(size_t,
 494                                                     PAGE_SIZE - page_off, left);
 495                                 l = copy_page_to_iter(pages[k++], page_off,
 496                                                       copy, i);
 497                                 off += l;
 498                                 left -= l;
 499                                 if (l < copy)
 500                                         break;
 501                         }
 502                 }
 503                 ceph_release_page_vector(pages, num_pages);
 504         }
 505
 506         if (off > iocb->ki_pos) {
 507                 ret = off - iocb->ki_pos;
 508                 iocb->ki_pos = off;
 509         }
 510
 511         dout("sync_read result %d\n", ret);
 512         return ret;
 513 }
 514
 515 /*
 516  * Write commit request unsafe callback, called to tell us when a
 517  * request is unsafe (that is, in flight--has been handed to the
 518  * messenger to send to its target osd).  It is called again when
 519  * we've received a response message indicating the request is
 520  * "safe" (its CEPH_OSD_FLAG_ONDISK flag is set), or when a request
 521  * is completed early (and unsuccessfully) due to a timeout or
 522  * interrupt.
 523  *
 524  * This is used if we requested both an ACK and ONDISK commit reply
 525  * from the OSD.
 526  */
 527 static void ceph_sync_write_unsafe(struct ceph_osd_request *req, bool unsafe)
 528 {
 529         struct ceph_inode_info *ci = ceph_inode(req->r_inode);
 530
 531         dout("%s %p tid %llu %ssafe\n", __func__, req, req->r_tid,
 532                 unsafe ? "un" : "");
 533         if (unsafe) {
 534                 ceph_get_cap_refs(ci, CEPH_CAP_FILE_WR);
 535                 spin_lock(&ci->i_unsafe_lock);
 536                 list_add_tail(&req->r_unsafe_item,
 537                               &ci->i_unsafe_writes);
 538                 spin_unlock(&ci->i_unsafe_lock);
 539         } else {
 540                 spin_lock(&ci->i_unsafe_lock);
 541                 list_del_init(&req->r_unsafe_item);
 542                 spin_unlock(&ci->i_unsafe_lock);
 543                 ceph_put_cap_refs(ci, CEPH_CAP_FILE_WR);
 544         }
 545 }
 546
 547
 548 /*
 549  * Synchronous write, straight from __user pointer or user pages.
 550  *
 551  * If write spans object boundary, just do multiple writes.  (For a
 552  * correct atomic write, we should e.g. take write locks on all
 553  * objects, rollback on failure, etc.)
 554  */
 555 static ssize_t
 556 ceph_sync_direct_write(struct kiocb *iocb, struct iov_iter *from, loff_t pos,
 557                        struct ceph_snap_context *snapc)
 558 {
 559         struct file *file = iocb->ki_filp;
 560         struct inode *inode = file_inode(file);
 561         struct ceph_inode_info *ci = ceph_inode(inode);
 562         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
 563         struct ceph_vino vino;
 564         struct ceph_osd_request *req;
 565         struct page **pages;
 566         int num_pages;
 567         int written = 0;
 568         int flags;
 569         int check_caps = 0;
 570         int ret;
 571         struct timespec mtime = CURRENT_TIME;
 572         size_t count = iov_iter_count(from);
 573
 574         if (ceph_snap(file_inode(file)) != CEPH_NOSNAP)
 575                 return -EROFS;
 576
 577         dout("sync_direct_write on file %p %lld~%u\n", file, pos,
 578              (unsigned)count);
 579
 580         ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + count);
 581         if (ret < 0)
 582                 return ret;
 583
 584         ret = invalidate_inode_pages2_range(inode->i_mapping,
 585                                             pos >> PAGE_CACHE_SHIFT,
 586                                             (pos + count) >> PAGE_CACHE_SHIFT);
 587         if (ret < 0)
 588                 dout("invalidate_inode_pages2_range returned %d\n", ret);
 589
 590         flags = CEPH_OSD_FLAG_ORDERSNAP |
 591                 CEPH_OSD_FLAG_ONDISK |
 592                 CEPH_OSD_FLAG_WRITE;
 593
 594         while (iov_iter_count(from) > 0) {
 595                 u64 len = iov_iter_single_seg_count(from);
 596                 size_t start;
 597                 ssize_t n;
 598
 599                 vino = ceph_vino(inode);
 600                 req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
 601                                             vino, pos, &len, 0,
 602                                             2,/*include a 'startsync' command*/
 603                                             CEPH_OSD_OP_WRITE, flags, snapc,
 604                                             ci->i_truncate_seq,
 605                                             ci->i_truncate_size,
 606                                             false);
 607                 if (IS_ERR(req)) {
 608                         ret = PTR_ERR(req);
 609                         break;
 610                 }
 611
 612                 osd_req_op_init(req, 1, CEPH_OSD_OP_STARTSYNC, 0);
 613
 614                 n = iov_iter_get_pages_alloc(from, &pages, len, &start);
 615                 if (unlikely(n < 0)) {
 616                         ret = n;
 617                         ceph_osdc_put_request(req);
 618                         break;
 619                 }
 620
 621                 num_pages = (n + start + PAGE_SIZE - 1) / PAGE_SIZE;
 622                 /*
 623                  * throw out any page cache pages in this range. this
 624                  * may block.
 625                  */
 626                 truncate_inode_pages_range(inode->i_mapping, pos,
 627                                    (pos+n) | (PAGE_CACHE_SIZE-1));
 628                 osd_req_op_extent_osd_data_pages(req, 0, pages, n, start,
 629                                                 false, false);
 630
 631                 /* BUG_ON(vino.snap != CEPH_NOSNAP); */
 632                 ceph_osdc_build_request(req, pos, snapc, vino.snap, &mtime);
 633
 634                 ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
 635                 if (!ret)
 636                         ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
 637
 638                 ceph_put_page_vector(pages, num_pages, false);
 639
 640                 ceph_osdc_put_request(req);
 641                 if (ret)
 642                         break;
 643                 pos += n;
 644                 written += n;
 645                 iov_iter_advance(from, n);
 646
 647                 if (pos > i_size_read(inode)) {
 648                         check_caps = ceph_inode_set_size(inode, pos);
 649                         if (check_caps)
 650                                 ceph_check_caps(ceph_inode(inode),
 651                                                 CHECK_CAPS_AUTHONLY,
 652                                                 NULL);
 653                 }
 654         }
 655
 656         if (ret != -EOLDSNAPC && written > 0) {
 657                 iocb->ki_pos = pos;
 658                 ret = written;
 659         }
 660         return ret;
 661 }
 662
 663
 664 /*
 665  * Synchronous write, straight from __user pointer or user pages.
 666  *
 667  * If write spans object boundary, just do multiple writes.  (For a
 668  * correct atomic write, we should e.g. take write locks on all
 669  * objects, rollback on failure, etc.)
 670  */
 671 static ssize_t
 672 ceph_sync_write(struct kiocb *iocb, struct iov_iter *from, loff_t pos,
 673                 struct ceph_snap_context *snapc)
 674 {
 675         struct file *file = iocb->ki_filp;
 676         struct inode *inode = file_inode(file);
 677         struct ceph_inode_info *ci = ceph_inode(inode);
 678         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
 679         struct ceph_vino vino;
 680         struct ceph_osd_request *req;
 681         struct page **pages;
 682         u64 len;
 683         int num_pages;
 684         int written = 0;
 685         int flags;
 686         int check_caps = 0;
 687         int ret;
 688         struct timespec mtime = CURRENT_TIME;
 689         size_t count = iov_iter_count(from);
 690
 691         if (ceph_snap(file_inode(file)) != CEPH_NOSNAP)
 692                 return -EROFS;
 693
 694         dout("sync_write on file %p %lld~%u\n", file, pos, (unsigned)count);
 695
 696         ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + count);
 697         if (ret < 0)
 698                 return ret;
 699
 700         ret = invalidate_inode_pages2_range(inode->i_mapping,
 701                                             pos >> PAGE_CACHE_SHIFT,
 702                                             (pos + count) >> PAGE_CACHE_SHIFT);
 703         if (ret < 0)
 704                 dout("invalidate_inode_pages2_range returned %d\n", ret);
 705
 706         flags = CEPH_OSD_FLAG_ORDERSNAP |
 707                 CEPH_OSD_FLAG_ONDISK |
 708                 CEPH_OSD_FLAG_WRITE |
 709                 CEPH_OSD_FLAG_ACK;
 710
 711         while ((len = iov_iter_count(from)) > 0) {
 712                 size_t left;
 713                 int n;
 714
 715                 vino = ceph_vino(inode);
 716                 req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
 717                                             vino, pos, &len, 0, 1,
 718                                             CEPH_OSD_OP_WRITE, flags, snapc,
 719                                             ci->i_truncate_seq,
 720                                             ci->i_truncate_size,
 721                                             false);
 722                 if (IS_ERR(req)) {
 723                         ret = PTR_ERR(req);
 724                         break;
 725                 }
 726
 727                 /*
 728                  * write from beginning of first page,
 729                  * regardless of io alignment
 730                  */
 731                 num_pages = (len + PAGE_CACHE_SIZE - 1) >> PAGE_CACHE_SHIFT;
 732
 733                 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
 734                 if (IS_ERR(pages)) {
 735                         ret = PTR_ERR(pages);
 736                         goto out;
 737                 }
 738
 739                 left = len;
 740                 for (n = 0; n < num_pages; n++) {
 741                         size_t plen = min_t(size_t, left, PAGE_SIZE);
 742                         ret = copy_page_from_iter(pages[n], 0, plen, from);
 743                         if (ret != plen) {
 744                                 ret = -EFAULT;
 745                                 break;
 746                         }
 747                         left -= ret;
 748                 }
 749
 750                 if (ret < 0) {
 751                         ceph_release_page_vector(pages, num_pages);
 752                         goto out;
 753                 }
 754
 755                 /* get a second commit callback */
 756                 req->r_unsafe_callback = ceph_sync_write_unsafe;
 757                 req->r_inode = inode;
 758
 759                 osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0,
 760                                                 false, true);
 761
 762                 /* BUG_ON(vino.snap != CEPH_NOSNAP); */
 763                 ceph_osdc_build_request(req, pos, snapc, vino.snap, &mtime);
 764
 765                 ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
 766                 if (!ret)
 767                         ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
 768
 769 out:
 770                 ceph_osdc_put_request(req);
 771                 if (ret == 0) {
 772                         pos += len;
 773                         written += len;
 774
 775                         if (pos > i_size_read(inode)) {
 776                                 check_caps = ceph_inode_set_size(inode, pos);
 777                                 if (check_caps)
 778                                         ceph_check_caps(ceph_inode(inode),
 779                                                         CHECK_CAPS_AUTHONLY,
 780                                                         NULL);
 781                         }
 782                 } else
 783                         break;
 784         }
 785
 786         if (ret != -EOLDSNAPC && written > 0) {
 787                 ret = written;
 788                 iocb->ki_pos = pos;
 789         }
 790         return ret;
 791 }
 792
 793 /*
 794  * Wrap generic_file_aio_read with checks for cap bits on the inode.
 795  * Atomically grab references, so that those bits are not released
 796  * back to the MDS mid-read.
 797  *
 798  * Hmm, the sync read case isn't actually async... should it be?
 799  */
 800 static ssize_t ceph_read_iter(struct kiocb *iocb, struct iov_iter *to)
 801 {
 802         struct file *filp = iocb->ki_filp;
 803         struct ceph_file_info *fi = filp->private_data;
 804         size_t len = iov_iter_count(to);
 805         struct inode *inode = file_inode(filp);
 806         struct ceph_inode_info *ci = ceph_inode(inode);
 807         struct page *pinned_page = NULL;
 808         ssize_t ret;
 809         int want, got = 0;
 810         int retry_op = 0, read = 0;
 811
 812 again:
 813         dout("aio_read %p %llx.%llx %llu~%u trying to get caps on %p\n",
 814              inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len, inode);
 815
 816         if (fi->fmode & CEPH_FILE_MODE_LAZY)
 817                 want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
 818         else
 819                 want = CEPH_CAP_FILE_CACHE;
 820         ret = ceph_get_caps(ci, CEPH_CAP_FILE_RD, want, -1, &got, &pinned_page);
 821         if (ret < 0)
 822                 return ret;
 823
 824         if ((got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0 ||
 825             (iocb->ki_flags & IOCB_DIRECT) ||
 826             (fi->flags & CEPH_F_SYNC)) {
 827
 828                 dout("aio_sync_read %p %llx.%llx %llu~%u got cap refs on %s\n",
 829                      inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len,
 830                      ceph_cap_string(got));
 831
 832                 if (ci->i_inline_version == CEPH_INLINE_NONE) {
 833                         /* hmm, this isn't really async... */
 834                         ret = ceph_sync_read(iocb, to, &retry_op);
 835                 } else {
 836                         retry_op = READ_INLINE;
 837                 }
 838         } else {
 839                 dout("aio_read %p %llx.%llx %llu~%u got cap refs on %s\n",
 840                      inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len,
 841                      ceph_cap_string(got));
 842
 843                 ret = generic_file_read_iter(iocb, to);
 844         }
 845         dout("aio_read %p %llx.%llx dropping cap refs on %s = %d\n",
 846              inode, ceph_vinop(inode), ceph_cap_string(got), (int)ret);
 847         if (pinned_page) {
 848                 page_cache_release(pinned_page);
 849                 pinned_page = NULL;
 850         }
 851         ceph_put_cap_refs(ci, got);
 852         if (retry_op && ret >= 0) {
 853                 int statret;
 854                 struct page *page = NULL;
 855                 loff_t i_size;
 856                 if (retry_op == READ_INLINE) {
 857                         page = __page_cache_alloc(GFP_KERNEL);
 858                         if (!page)
 859                                 return -ENOMEM;
 860                 }
 861
 862                 statret = __ceph_do_getattr(inode, page,
 863                                             CEPH_STAT_CAP_INLINE_DATA, !!page);
 864                 if (statret < 0) {
 865                          __free_page(page);
 866                         if (statret == -ENODATA) {
 867                                 BUG_ON(retry_op != READ_INLINE);
 868                                 goto again;
 869                         }
 870                         return statret;
 871                 }
 872
 873                 i_size = i_size_read(inode);
 874                 if (retry_op == READ_INLINE) {
 875                         BUG_ON(ret > 0 || read > 0);
 876                         if (iocb->ki_pos < i_size &&
 877                             iocb->ki_pos < PAGE_CACHE_SIZE) {
 878                                 loff_t end = min_t(loff_t, i_size,
 879                                                    iocb->ki_pos + len);
 880                                 end = min_t(loff_t, end, PAGE_CACHE_SIZE);
 881                                 if (statret < end)
 882                                         zero_user_segment(page, statret, end);
 883                                 ret = copy_page_to_iter(page,
 884                                                 iocb->ki_pos & ~PAGE_MASK,
 885                                                 end - iocb->ki_pos, to);
 886                                 iocb->ki_pos += ret;
 887                                 read += ret;
 888                         }
 889                         if (iocb->ki_pos < i_size && read < len) {
 890                                 size_t zlen = min_t(size_t, len - read,
 891                                                     i_size - iocb->ki_pos);
 892                                 ret = iov_iter_zero(zlen, to);
 893                                 iocb->ki_pos += ret;
 894                                 read += ret;
 895                         }
 896                         __free_pages(page, 0);
 897                         return read;
 898                 }
 899
 900                 /* hit EOF or hole? */
 901                 if (retry_op == CHECK_EOF && iocb->ki_pos < i_size &&
 902                     ret < len) {
 903                         dout("sync_read hit hole, ppos %lld < size %lld"
 904                              ", reading more\n", iocb->ki_pos,
 905                              inode->i_size);
 906
 907                         read += ret;
 908                         len -= ret;
 909                         retry_op = 0;
 910                         goto again;
 911                 }
 912         }
 913
 914         if (ret >= 0)
 915                 ret += read;
 916
 917         return ret;
 918 }
 919
 920 /*
 921  * Take cap references to avoid releasing caps to MDS mid-write.
 922  *
 923  * If we are synchronous, and write with an old snap context, the OSD
 924  * may return EOLDSNAPC.  In that case, retry the write.. _after_
 925  * dropping our cap refs and allowing the pending snap to logically
 926  * complete _before_ this write occurs.
 927  *
 928  * If we are near ENOSPC, write synchronously.
 929  */
 930 static ssize_t ceph_write_iter(struct kiocb *iocb, struct iov_iter *from)
 931 {
 932         struct file *file = iocb->ki_filp;
 933         struct ceph_file_info *fi = file->private_data;
 934         struct inode *inode = file_inode(file);
 935         struct ceph_inode_info *ci = ceph_inode(inode);
 936         struct ceph_osd_client *osdc =
 937                 &ceph_sb_to_client(inode->i_sb)->client->osdc;
 938         struct ceph_cap_flush *prealloc_cf;
 939         ssize_t count, written = 0;
 940         int err, want, got;
 941         loff_t pos;
 942
 943         if (ceph_snap(inode) != CEPH_NOSNAP)
 944                 return -EROFS;
 945
 946         prealloc_cf = ceph_alloc_cap_flush();
 947         if (!prealloc_cf)
 948                 return -ENOMEM;
 949
 950         mutex_lock(&inode->i_mutex);
 951
 952         /* We can write back this queue in page reclaim */
 953         current->backing_dev_info = inode_to_bdi(inode);
 954
 955         if (iocb->ki_flags & IOCB_APPEND) {
 956                 err = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE, false);
 957                 if (err < 0)
 958                         goto out;
 959         }
 960
 961         err = generic_write_checks(iocb, from);
 962         if (err <= 0)
 963                 goto out;
 964
 965         pos = iocb->ki_pos;
 966         count = iov_iter_count(from);
 967         err = file_remove_privs(file);
 968         if (err)
 969                 goto out;
 970
 971         err = file_update_time(file);
 972         if (err)
 973                 goto out;
 974
 975         if (ci->i_inline_version != CEPH_INLINE_NONE) {
 976                 err = ceph_uninline_data(file, NULL);
 977                 if (err < 0)
 978                         goto out;
 979         }
 980
 981 retry_snap:
 982         if (ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_FULL)) {
 983                 err = -ENOSPC;
 984                 goto out;
 985         }
 986
 987         dout("aio_write %p %llx.%llx %llu~%zd getting caps. i_size %llu\n",
 988              inode, ceph_vinop(inode), pos, count, inode->i_size);
 989         if (fi->fmode & CEPH_FILE_MODE_LAZY)
 990                 want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
 991         else
 992                 want = CEPH_CAP_FILE_BUFFER;
 993         got = 0;
 994         err = ceph_get_caps(ci, CEPH_CAP_FILE_WR, want, pos + count,
 995                             &got, NULL);
 996         if (err < 0)
 997                 goto out;
 998
 999         dout("aio_write %p %llx.%llx %llu~%zd got cap refs on %s\n",
1000              inode, ceph_vinop(inode), pos, count, ceph_cap_string(got));
1001
1002         if ((got & (CEPH_CAP_FILE_BUFFER|CEPH_CAP_FILE_LAZYIO)) == 0 ||
1003             (iocb->ki_flags & IOCB_DIRECT) || (fi->flags & CEPH_F_SYNC)) {
1004                 struct ceph_snap_context *snapc;
1005                 struct iov_iter data;
1006                 mutex_unlock(&inode->i_mutex);
1007
1008                 spin_lock(&ci->i_ceph_lock);
1009                 if (__ceph_have_pending_cap_snap(ci)) {
1010                         struct ceph_cap_snap *capsnap =
1011                                         list_last_entry(&ci->i_cap_snaps,
1012                                                         struct ceph_cap_snap,
1013                                                         ci_item);
1014                         snapc = ceph_get_snap_context(capsnap->context);
1015                 } else {
1016                         BUG_ON(!ci->i_head_snapc);
1017                         snapc = ceph_get_snap_context(ci->i_head_snapc);
1018                 }
1019                 spin_unlock(&ci->i_ceph_lock);
1020
1021                 /* we might need to revert back to that point */
1022                 data = *from;
1023                 if (iocb->ki_flags & IOCB_DIRECT)
1024                         written = ceph_sync_direct_write(iocb, &data, pos,
1025                                                          snapc);
1026                 else
1027                         written = ceph_sync_write(iocb, &data, pos, snapc);
1028                 if (written == -EOLDSNAPC) {
1029                         dout("aio_write %p %llx.%llx %llu~%u"
1030                                 "got EOLDSNAPC, retrying\n",
1031                                 inode, ceph_vinop(inode),
1032                                 pos, (unsigned)count);
1033                         mutex_lock(&inode->i_mutex);
1034                         goto retry_snap;
1035                 }
1036                 if (written > 0)
1037                         iov_iter_advance(from, written);
1038                 ceph_put_snap_context(snapc);
1039         } else {
1040                 loff_t old_size = inode->i_size;
1041                 /*
1042                  * No need to acquire the i_truncate_mutex. Because
1043                  * the MDS revokes Fwb caps before sending truncate
1044                  * message to us. We can't get Fwb cap while there
1045                  * are pending vmtruncate. So write and vmtruncate
1046                  * can not run at the same time
1047                  */
1048                 written = generic_perform_write(file, from, pos);
1049                 if (likely(written >= 0))
1050                         iocb->ki_pos = pos + written;
1051                 if (inode->i_size > old_size)
1052                         ceph_fscache_update_objectsize(inode);
1053                 mutex_unlock(&inode->i_mutex);
1054         }
1055
1056         if (written >= 0) {
1057                 int dirty;
1058                 spin_lock(&ci->i_ceph_lock);
1059                 ci->i_inline_version = CEPH_INLINE_NONE;
1060                 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
1061                                                &prealloc_cf);
1062                 spin_unlock(&ci->i_ceph_lock);
1063                 if (dirty)
1064                         __mark_inode_dirty(inode, dirty);
1065         }
1066
1067         dout("aio_write %p %llx.%llx %llu~%u  dropping cap refs on %s\n",
1068              inode, ceph_vinop(inode), pos, (unsigned)count,
1069              ceph_cap_string(got));
1070         ceph_put_cap_refs(ci, got);
1071
1072         if (written >= 0 &&
1073             ((file->f_flags & O_SYNC) || IS_SYNC(file->f_mapping->host) ||
1074              ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_NEARFULL))) {
1075                 err = vfs_fsync_range(file, pos, pos + written - 1, 1);
1076                 if (err < 0)
1077                         written = err;
1078         }
1079
1080         goto out_unlocked;
1081
1082 out:
1083         mutex_unlock(&inode->i_mutex);
1084 out_unlocked:
1085         ceph_free_cap_flush(prealloc_cf);
1086         current->backing_dev_info = NULL;
1087         return written ? written : err;
1088 }
1089
1090 /*
1091  * llseek.  be sure to verify file size on SEEK_END.
1092  */
1093 static loff_t ceph_llseek(struct file *file, loff_t offset, int whence)
1094 {
1095         struct inode *inode = file->f_mapping->host;
1096         int ret;
1097
1098         mutex_lock(&inode->i_mutex);
1099
1100         if (whence == SEEK_END || whence == SEEK_DATA || whence == SEEK_HOLE) {
1101                 ret = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE, false);
1102                 if (ret < 0) {
1103                         offset = ret;
1104                         goto out;
1105                 }
1106         }
1107
1108         switch (whence) {
1109         case SEEK_END:
1110                 offset += inode->i_size;
1111                 break;
1112         case SEEK_CUR:
1113                 /*
1114                  * Here we special-case the lseek(fd, 0, SEEK_CUR)
1115                  * position-querying operation.  Avoid rewriting the "same"
1116                  * f_pos value back to the file because a concurrent read(),
1117                  * write() or lseek() might have altered it
1118                  */
1119                 if (offset == 0) {
1120                         offset = file->f_pos;
1121                         goto out;
1122                 }
1123                 offset += file->f_pos;
1124                 break;
1125         case SEEK_DATA:
1126                 if (offset >= inode->i_size) {
1127                         ret = -ENXIO;
1128                         goto out;
1129                 }
1130                 break;
1131         case SEEK_HOLE:
1132                 if (offset >= inode->i_size) {
1133                         ret = -ENXIO;
1134                         goto out;
1135                 }
1136                 offset = inode->i_size;
1137                 break;
1138         }
1139
1140         offset = vfs_setpos(file, offset, inode->i_sb->s_maxbytes);
1141
1142 out:
1143         mutex_unlock(&inode->i_mutex);
1144         return offset;
1145 }
1146
1147 static inline void ceph_zero_partial_page(
1148         struct inode *inode, loff_t offset, unsigned size)
1149 {
1150         struct page *page;
1151         pgoff_t index = offset >> PAGE_CACHE_SHIFT;
1152
1153         page = find_lock_page(inode->i_mapping, index);
1154         if (page) {
1155                 wait_on_page_writeback(page);
1156                 zero_user(page, offset & (PAGE_CACHE_SIZE - 1), size);
1157                 unlock_page(page);
1158                 page_cache_release(page);
1159         }
1160 }
1161
1162 static void ceph_zero_pagecache_range(struct inode *inode, loff_t offset,
1163                                       loff_t length)
1164 {
1165         loff_t nearly = round_up(offset, PAGE_CACHE_SIZE);
1166         if (offset < nearly) {
1167                 loff_t size = nearly - offset;
1168                 if (length < size)
1169                         size = length;
1170                 ceph_zero_partial_page(inode, offset, size);
1171                 offset += size;
1172                 length -= size;
1173         }
1174         if (length >= PAGE_CACHE_SIZE) {
1175                 loff_t size = round_down(length, PAGE_CACHE_SIZE);
1176                 truncate_pagecache_range(inode, offset, offset + size - 1);
1177                 offset += size;
1178                 length -= size;
1179         }
1180         if (length)
1181                 ceph_zero_partial_page(inode, offset, length);
1182 }
1183
1184 static int ceph_zero_partial_object(struct inode *inode,
1185                                     loff_t offset, loff_t *length)
1186 {
1187         struct ceph_inode_info *ci = ceph_inode(inode);
1188         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
1189         struct ceph_osd_request *req;
1190         int ret = 0;
1191         loff_t zero = 0;
1192         int op;
1193
1194         if (!length) {
1195                 op = offset ? CEPH_OSD_OP_DELETE : CEPH_OSD_OP_TRUNCATE;
1196                 length = &zero;
1197         } else {
1198                 op = CEPH_OSD_OP_ZERO;
1199         }
1200
1201         req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
1202                                         ceph_vino(inode),
1203                                         offset, length,
1204                                         0, 1, op,
1205                                         CEPH_OSD_FLAG_WRITE |
1206                                         CEPH_OSD_FLAG_ONDISK,
1207                                         NULL, 0, 0, false);
1208         if (IS_ERR(req)) {
1209                 ret = PTR_ERR(req);
1210                 goto out;
1211         }
1212
1213         ceph_osdc_build_request(req, offset, NULL, ceph_vino(inode).snap,
1214                                 &inode->i_mtime);
1215
1216         ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
1217         if (!ret) {
1218                 ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
1219                 if (ret == -ENOENT)
1220                         ret = 0;
1221         }
1222         ceph_osdc_put_request(req);
1223
1224 out:
1225         return ret;
1226 }
1227
1228 static int ceph_zero_objects(struct inode *inode, loff_t offset, loff_t length)
1229 {
1230         int ret = 0;
1231         struct ceph_inode_info *ci = ceph_inode(inode);
1232         s32 stripe_unit = ceph_file_layout_su(ci->i_layout);
1233         s32 stripe_count = ceph_file_layout_stripe_count(ci->i_layout);
1234         s32 object_size = ceph_file_layout_object_size(ci->i_layout);
1235         u64 object_set_size = object_size * stripe_count;
1236         u64 nearly, t;
1237
1238         /* round offset up to next period boundary */
1239         nearly = offset + object_set_size - 1;
1240         t = nearly;
1241         nearly -= do_div(t, object_set_size);
1242
1243         while (length && offset < nearly) {
1244                 loff_t size = length;
1245                 ret = ceph_zero_partial_object(inode, offset, &size);
1246                 if (ret < 0)
1247                         return ret;
1248                 offset += size;
1249                 length -= size;
1250         }
1251         while (length >= object_set_size) {
1252                 int i;
1253                 loff_t pos = offset;
1254                 for (i = 0; i < stripe_count; ++i) {
1255                         ret = ceph_zero_partial_object(inode, pos, NULL);
1256                         if (ret < 0)
1257                                 return ret;
1258                         pos += stripe_unit;
1259                 }
1260                 offset += object_set_size;
1261                 length -= object_set_size;
1262         }
1263         while (length) {
1264                 loff_t size = length;
1265                 ret = ceph_zero_partial_object(inode, offset, &size);
1266                 if (ret < 0)
1267                         return ret;
1268                 offset += size;
1269                 length -= size;
1270         }
1271         return ret;
1272 }
1273
1274 static long ceph_fallocate(struct file *file, int mode,
1275                                 loff_t offset, loff_t length)
1276 {
1277         struct ceph_file_info *fi = file->private_data;
1278         struct inode *inode = file_inode(file);
1279         struct ceph_inode_info *ci = ceph_inode(inode);
1280         struct ceph_osd_client *osdc =
1281                 &ceph_inode_to_client(inode)->client->osdc;
1282         struct ceph_cap_flush *prealloc_cf;
1283         int want, got = 0;
1284         int dirty;
1285         int ret = 0;
1286         loff_t endoff = 0;
1287         loff_t size;
1288
1289         if (mode & ~(FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE))
1290                 return -EOPNOTSUPP;
1291
1292         if (!S_ISREG(inode->i_mode))
1293                 return -EOPNOTSUPP;
1294
1295         prealloc_cf = ceph_alloc_cap_flush();
1296         if (!prealloc_cf)
1297                 return -ENOMEM;
1298
1299         mutex_lock(&inode->i_mutex);
1300
1301         if (ceph_snap(inode) != CEPH_NOSNAP) {
1302                 ret = -EROFS;
1303                 goto unlock;
1304         }
1305
1306         if (ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_FULL) &&
1307                 !(mode & FALLOC_FL_PUNCH_HOLE)) {
1308                 ret = -ENOSPC;
1309                 goto unlock;
1310         }
1311
1312         if (ci->i_inline_version != CEPH_INLINE_NONE) {
1313                 ret = ceph_uninline_data(file, NULL);
1314                 if (ret < 0)
1315                         goto unlock;
1316         }
1317
1318         size = i_size_read(inode);
1319         if (!(mode & FALLOC_FL_KEEP_SIZE))
1320                 endoff = offset + length;
1321
1322         if (fi->fmode & CEPH_FILE_MODE_LAZY)
1323                 want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
1324         else
1325                 want = CEPH_CAP_FILE_BUFFER;
1326
1327         ret = ceph_get_caps(ci, CEPH_CAP_FILE_WR, want, endoff, &got, NULL);
1328         if (ret < 0)
1329                 goto unlock;
1330
1331         if (mode & FALLOC_FL_PUNCH_HOLE) {
1332                 if (offset < size)
1333                         ceph_zero_pagecache_range(inode, offset, length);
1334                 ret = ceph_zero_objects(inode, offset, length);
1335         } else if (endoff > size) {
1336                 truncate_pagecache_range(inode, size, -1);
1337                 if (ceph_inode_set_size(inode, endoff))
1338                         ceph_check_caps(ceph_inode(inode),
1339                                 CHECK_CAPS_AUTHONLY, NULL);
1340         }
1341
1342         if (!ret) {
1343                 spin_lock(&ci->i_ceph_lock);
1344                 ci->i_inline_version = CEPH_INLINE_NONE;
1345                 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
1346                                                &prealloc_cf);
1347                 spin_unlock(&ci->i_ceph_lock);
1348                 if (dirty)
1349                         __mark_inode_dirty(inode, dirty);
1350         }
1351
1352         ceph_put_cap_refs(ci, got);
1353 unlock:
1354         mutex_unlock(&inode->i_mutex);
1355         ceph_free_cap_flush(prealloc_cf);
1356         return ret;
1357 }
1358
1359 const struct file_operations ceph_file_fops = {
1360         .open = ceph_open,
1361         .release = ceph_release,
1362         .llseek = ceph_llseek,
1363         .read_iter = ceph_read_iter,
1364         .write_iter = ceph_write_iter,
1365         .mmap = ceph_mmap,
1366         .fsync = ceph_fsync,
1367         .lock = ceph_lock,
1368         .flock = ceph_flock,
1369         .splice_read = generic_file_splice_read,
1370         .splice_write = iter_file_splice_write,
1371         .unlocked_ioctl = ceph_ioctl,
1372         .compat_ioctl   = ceph_ioctl,
1373         .fallocate      = ceph_fallocate,
1374 };
1375