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/aio.h>
  11 #include <linux/falloc.h>
  12
  13 #include "super.h"
  14 #include "mds_client.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
  73         switch (inode->i_mode & S_IFMT) {
  74         case S_IFREG:
  75         case S_IFDIR:
  76                 dout("init_file %p %p 0%o (regular)\n", inode, file,
  77                      inode->i_mode);
  78                 cf = kmem_cache_alloc(ceph_file_cachep, GFP_NOFS | __GFP_ZERO);
  79                 if (cf == NULL) {
  80                         ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */
  81                         return -ENOMEM;
  82                 }
  83                 cf->fmode = fmode;
  84                 cf->next_offset = 2;
  85                 file->private_data = cf;
  86                 BUG_ON(inode->i_fop->release != ceph_release);
  87                 break;
  88
  89         case S_IFLNK:
  90                 dout("init_file %p %p 0%o (symlink)\n", inode, file,
  91                      inode->i_mode);
  92                 ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */
  93                 break;
  94
  95         default:
  96                 dout("init_file %p %p 0%o (special)\n", inode, file,
  97                      inode->i_mode);
  98                 /*
  99                  * we need to drop the open ref now, since we don't
 100                  * have .release set to ceph_release.
 101                  */
 102                 ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */
 103                 BUG_ON(inode->i_fop->release == ceph_release);
 104
 105                 /* call the proper open fop */
 106                 ret = inode->i_fop->open(inode, file);
 107         }
 108         return ret;
 109 }
 110
 111 /*
 112  * If we already have the requisite capabilities, we can satisfy
 113  * the open request locally (no need to request new caps from the
 114  * MDS).  We do, however, need to inform the MDS (asynchronously)
 115  * if our wanted caps set expands.
 116  */
 117 int ceph_open(struct inode *inode, struct file *file)
 118 {
 119         struct ceph_inode_info *ci = ceph_inode(inode);
 120         struct ceph_fs_client *fsc = ceph_sb_to_client(inode->i_sb);
 121         struct ceph_mds_client *mdsc = fsc->mdsc;
 122         struct ceph_mds_request *req;
 123         struct ceph_file_info *cf = file->private_data;
 124         struct inode *parent_inode = NULL;
 125         int err;
 126         int flags, fmode, wanted;
 127
 128         if (cf) {
 129                 dout("open file %p is already opened\n", file);
 130                 return 0;
 131         }
 132
 133         /* filter out O_CREAT|O_EXCL; vfs did that already.  yuck. */
 134         flags = file->f_flags & ~(O_CREAT|O_EXCL);
 135         if (S_ISDIR(inode->i_mode))
 136                 flags = O_DIRECTORY;  /* mds likes to know */
 137
 138         dout("open inode %p ino %llx.%llx file %p flags %d (%d)\n", inode,
 139              ceph_vinop(inode), file, flags, file->f_flags);
 140         fmode = ceph_flags_to_mode(flags);
 141         wanted = ceph_caps_for_mode(fmode);
 142
 143         /* snapped files are read-only */
 144         if (ceph_snap(inode) != CEPH_NOSNAP && (file->f_mode & FMODE_WRITE))
 145                 return -EROFS;
 146
 147         /* trivially open snapdir */
 148         if (ceph_snap(inode) == CEPH_SNAPDIR) {
 149                 spin_lock(&ci->i_ceph_lock);
 150                 __ceph_get_fmode(ci, fmode);
 151                 spin_unlock(&ci->i_ceph_lock);
 152                 return ceph_init_file(inode, file, fmode);
 153         }
 154
 155         /*
 156          * No need to block if we have caps on the auth MDS (for
 157          * write) or any MDS (for read).  Update wanted set
 158          * asynchronously.
 159          */
 160         spin_lock(&ci->i_ceph_lock);
 161         if (__ceph_is_any_real_caps(ci) &&
 162             (((fmode & CEPH_FILE_MODE_WR) == 0) || ci->i_auth_cap)) {
 163                 int mds_wanted = __ceph_caps_mds_wanted(ci);
 164                 int issued = __ceph_caps_issued(ci, NULL);
 165
 166                 dout("open %p fmode %d want %s issued %s using existing\n",
 167                      inode, fmode, ceph_cap_string(wanted),
 168                      ceph_cap_string(issued));
 169                 __ceph_get_fmode(ci, fmode);
 170                 spin_unlock(&ci->i_ceph_lock);
 171
 172                 /* adjust wanted? */
 173                 if ((issued & wanted) != wanted &&
 174                     (mds_wanted & wanted) != wanted &&
 175                     ceph_snap(inode) != CEPH_SNAPDIR)
 176                         ceph_check_caps(ci, 0, NULL);
 177
 178                 return ceph_init_file(inode, file, fmode);
 179         } else if (ceph_snap(inode) != CEPH_NOSNAP &&
 180                    (ci->i_snap_caps & wanted) == wanted) {
 181                 __ceph_get_fmode(ci, fmode);
 182                 spin_unlock(&ci->i_ceph_lock);
 183                 return ceph_init_file(inode, file, fmode);
 184         }
 185         spin_unlock(&ci->i_ceph_lock);
 186
 187         dout("open fmode %d wants %s\n", fmode, ceph_cap_string(wanted));
 188         req = prepare_open_request(inode->i_sb, flags, 0);
 189         if (IS_ERR(req)) {
 190                 err = PTR_ERR(req);
 191                 goto out;
 192         }
 193         req->r_inode = inode;
 194         ihold(inode);
 195         req->r_num_caps = 1;
 196         if (flags & (O_CREAT|O_TRUNC))
 197                 parent_inode = ceph_get_dentry_parent_inode(file->f_dentry);
 198         err = ceph_mdsc_do_request(mdsc, parent_inode, req);
 199         iput(parent_inode);
 200         if (!err)
 201                 err = ceph_init_file(inode, file, req->r_fmode);
 202         ceph_mdsc_put_request(req);
 203         dout("open result=%d on %llx.%llx\n", err, ceph_vinop(inode));
 204 out:
 205         return err;
 206 }
 207
 208
 209 /*
 210  * Do a lookup + open with a single request.  If we get a non-existent
 211  * file or symlink, return 1 so the VFS can retry.
 212  */
 213 int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
 214                      struct file *file, unsigned flags, umode_t mode,
 215                      int *opened)
 216 {
 217         struct ceph_fs_client *fsc = ceph_sb_to_client(dir->i_sb);
 218         struct ceph_mds_client *mdsc = fsc->mdsc;
 219         struct ceph_mds_request *req;
 220         struct dentry *dn;
 221         int err;
 222
 223         dout("atomic_open %p dentry %p '%.*s' %s flags %d mode 0%o\n",
 224              dir, dentry, dentry->d_name.len, dentry->d_name.name,
 225              d_unhashed(dentry) ? "unhashed" : "hashed", flags, mode);
 226
 227         if (dentry->d_name.len > NAME_MAX)
 228                 return -ENAMETOOLONG;
 229
 230         err = ceph_init_dentry(dentry);
 231         if (err < 0)
 232                 return err;
 233
 234         /* do the open */
 235         req = prepare_open_request(dir->i_sb, flags, mode);
 236         if (IS_ERR(req))
 237                 return PTR_ERR(req);
 238         req->r_dentry = dget(dentry);
 239         req->r_num_caps = 2;
 240         if (flags & O_CREAT) {
 241                 req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
 242                 req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
 243         }
 244         req->r_locked_dir = dir;           /* caller holds dir->i_mutex */
 245         err = ceph_mdsc_do_request(mdsc,
 246                                    (flags & (O_CREAT|O_TRUNC)) ? dir : NULL,
 247                                    req);
 248         if (err)
 249                 goto out_err;
 250
 251         err = ceph_handle_snapdir(req, dentry, err);
 252         if (err == 0 && (flags & O_CREAT) && !req->r_reply_info.head->is_dentry)
 253                 err = ceph_handle_notrace_create(dir, dentry);
 254
 255         if (d_unhashed(dentry)) {
 256                 dn = ceph_finish_lookup(req, dentry, err);
 257                 if (IS_ERR(dn))
 258                         err = PTR_ERR(dn);
 259         } else {
 260                 /* we were given a hashed negative dentry */
 261                 dn = NULL;
 262         }
 263         if (err)
 264                 goto out_err;
 265         if (dn || dentry->d_inode == NULL || S_ISLNK(dentry->d_inode->i_mode)) {
 266                 /* make vfs retry on splice, ENOENT, or symlink */
 267                 dout("atomic_open finish_no_open on dn %p\n", dn);
 268                 err = finish_no_open(file, dn);
 269         } else {
 270                 dout("atomic_open finish_open on dn %p\n", dn);
 271                 if (req->r_op == CEPH_MDS_OP_CREATE && req->r_reply_info.has_create_ino) {
 272                         *opened |= FILE_CREATED;
 273                 }
 274                 err = finish_open(file, dentry, ceph_open, opened);
 275         }
 276
 277 out_err:
 278         ceph_mdsc_put_request(req);
 279         dout("atomic_open result=%d\n", err);
 280         return err;
 281 }
 282
 283 int ceph_release(struct inode *inode, struct file *file)
 284 {
 285         struct ceph_inode_info *ci = ceph_inode(inode);
 286         struct ceph_file_info *cf = file->private_data;
 287
 288         dout("release inode %p file %p\n", inode, file);
 289         ceph_put_fmode(ci, cf->fmode);
 290         if (cf->last_readdir)
 291                 ceph_mdsc_put_request(cf->last_readdir);
 292         kfree(cf->last_name);
 293         kfree(cf->dir_info);
 294         dput(cf->dentry);
 295         kmem_cache_free(ceph_file_cachep, cf);
 296
 297         /* wake up anyone waiting for caps on this inode */
 298         wake_up_all(&ci->i_cap_wq);
 299         return 0;
 300 }
 301
 302 /*
 303  * Read a range of bytes striped over one or more objects.  Iterate over
 304  * objects we stripe over.  (That's not atomic, but good enough for now.)
 305  *
 306  * If we get a short result from the OSD, check against i_size; we need to
 307  * only return a short read to the caller if we hit EOF.
 308  */
 309 static int striped_read(struct inode *inode,
 310                         u64 off, u64 len,
 311                         struct page **pages, int num_pages,
 312                         int *checkeof, bool o_direct,
 313                         unsigned long buf_align)
 314 {
 315         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
 316         struct ceph_inode_info *ci = ceph_inode(inode);
 317         u64 pos, this_len, left;
 318         int io_align, page_align;
 319         int pages_left;
 320         int read;
 321         struct page **page_pos;
 322         int ret;
 323         bool hit_stripe, was_short;
 324
 325         /*
 326          * we may need to do multiple reads.  not atomic, unfortunately.
 327          */
 328         pos = off;
 329         left = len;
 330         page_pos = pages;
 331         pages_left = num_pages;
 332         read = 0;
 333         io_align = off & ~PAGE_MASK;
 334
 335 more:
 336         if (o_direct)
 337                 page_align = (pos - io_align + buf_align) & ~PAGE_MASK;
 338         else
 339                 page_align = pos & ~PAGE_MASK;
 340         this_len = left;
 341         ret = ceph_osdc_readpages(&fsc->client->osdc, ceph_vino(inode),
 342                                   &ci->i_layout, pos, &this_len,
 343                                   ci->i_truncate_seq,
 344                                   ci->i_truncate_size,
 345                                   page_pos, pages_left, page_align);
 346         if (ret == -ENOENT)
 347                 ret = 0;
 348         hit_stripe = this_len < left;
 349         was_short = ret >= 0 && ret < this_len;
 350         dout("striped_read %llu~%llu (read %u) got %d%s%s\n", pos, left, read,
 351              ret, hit_stripe ? " HITSTRIPE" : "", was_short ? " SHORT" : "");
 352
 353         if (ret > 0) {
 354                 int didpages = (page_align + ret) >> PAGE_CACHE_SHIFT;
 355
 356                 if (read < pos - off) {
 357                         dout(" zero gap %llu to %llu\n", off + read, pos);
 358                         ceph_zero_page_vector_range(page_align + read,
 359                                                     pos - off - read, pages);
 360                 }
 361                 pos += ret;
 362                 read = pos - off;
 363                 left -= ret;
 364                 page_pos += didpages;
 365                 pages_left -= didpages;
 366
 367                 /* hit stripe? */
 368                 if (left && hit_stripe)
 369                         goto more;
 370         }
 371
 372         if (was_short) {
 373                 /* did we bounce off eof? */
 374                 if (pos + left > inode->i_size)
 375                         *checkeof = 1;
 376
 377                 /* zero trailing bytes (inside i_size) */
 378                 if (left > 0 && pos < inode->i_size) {
 379                         if (pos + left > inode->i_size)
 380                                 left = inode->i_size - pos;
 381
 382                         dout("zero tail %llu\n", left);
 383                         ceph_zero_page_vector_range(page_align + read, left,
 384                                                     pages);
 385                         read += left;
 386                 }
 387         }
 388
 389         if (ret >= 0)
 390                 ret = read;
 391         dout("striped_read returns %d\n", ret);
 392         return ret;
 393 }
 394
 395 /*
 396  * Completely synchronous read and write methods.  Direct from __user
 397  * buffer to osd, or directly to user pages (if O_DIRECT).
 398  *
 399  * If the read spans object boundary, just do multiple reads.
 400  */
 401 static ssize_t ceph_sync_read(struct file *file, char __user *data,
 402                               unsigned len, loff_t *poff, int *checkeof)
 403 {
 404         struct inode *inode = file_inode(file);
 405         struct page **pages;
 406         u64 off = *poff;
 407         int num_pages, ret;
 408
 409         dout("sync_read on file %p %llu~%u %s\n", file, off, len,
 410              (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
 411
 412         if (file->f_flags & O_DIRECT) {
 413                 num_pages = calc_pages_for((unsigned long)data, len);
 414                 pages = ceph_get_direct_page_vector(data, num_pages, true);
 415         } else {
 416                 num_pages = calc_pages_for(off, len);
 417                 pages = ceph_alloc_page_vector(num_pages, GFP_NOFS);
 418         }
 419         if (IS_ERR(pages))
 420                 return PTR_ERR(pages);
 421
 422         /*
 423          * flush any page cache pages in this range.  this
 424          * will make concurrent normal and sync io slow,
 425          * but it will at least behave sensibly when they are
 426          * in sequence.
 427          */
 428         ret = filemap_write_and_wait(inode->i_mapping);
 429         if (ret < 0)
 430                 goto done;
 431
 432         ret = striped_read(inode, off, len, pages, num_pages, checkeof,
 433                            file->f_flags & O_DIRECT,
 434                            (unsigned long)data & ~PAGE_MASK);
 435
 436         if (ret >= 0 && (file->f_flags & O_DIRECT) == 0)
 437                 ret = ceph_copy_page_vector_to_user(pages, data, off, ret);
 438         if (ret >= 0)
 439                 *poff = off + ret;
 440
 441 done:
 442         if (file->f_flags & O_DIRECT)
 443                 ceph_put_page_vector(pages, num_pages, true);
 444         else
 445                 ceph_release_page_vector(pages, num_pages);
 446         dout("sync_read result %d\n", ret);
 447         return ret;
 448 }
 449
 450 /*
 451  * Write commit request unsafe callback, called to tell us when a
 452  * request is unsafe (that is, in flight--has been handed to the
 453  * messenger to send to its target osd).  It is called again when
 454  * we've received a response message indicating the request is
 455  * "safe" (its CEPH_OSD_FLAG_ONDISK flag is set), or when a request
 456  * is completed early (and unsuccessfully) due to a timeout or
 457  * interrupt.
 458  *
 459  * This is used if we requested both an ACK and ONDISK commit reply
 460  * from the OSD.
 461  */
 462 static void ceph_sync_write_unsafe(struct ceph_osd_request *req, bool unsafe)
 463 {
 464         struct ceph_inode_info *ci = ceph_inode(req->r_inode);
 465
 466         dout("%s %p tid %llu %ssafe\n", __func__, req, req->r_tid,
 467                 unsafe ? "un" : "");
 468         if (unsafe) {
 469                 ceph_get_cap_refs(ci, CEPH_CAP_FILE_WR);
 470                 spin_lock(&ci->i_unsafe_lock);
 471                 list_add_tail(&req->r_unsafe_item,
 472                               &ci->i_unsafe_writes);
 473                 spin_unlock(&ci->i_unsafe_lock);
 474         } else {
 475                 spin_lock(&ci->i_unsafe_lock);
 476                 list_del_init(&req->r_unsafe_item);
 477                 spin_unlock(&ci->i_unsafe_lock);
 478                 ceph_put_cap_refs(ci, CEPH_CAP_FILE_WR);
 479         }
 480 }
 481
 482 /*
 483  * Synchronous write, straight from __user pointer or user pages (if
 484  * O_DIRECT).
 485  *
 486  * If write spans object boundary, just do multiple writes.  (For a
 487  * correct atomic write, we should e.g. take write locks on all
 488  * objects, rollback on failure, etc.)
 489  */
 490 static ssize_t ceph_sync_write(struct file *file, const char __user *data,
 491                                size_t left, loff_t pos, loff_t *ppos)
 492 {
 493         struct inode *inode = file_inode(file);
 494         struct ceph_inode_info *ci = ceph_inode(inode);
 495         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
 496         struct ceph_snap_context *snapc;
 497         struct ceph_vino vino;
 498         struct ceph_osd_request *req;
 499         int num_ops = 1;
 500         struct page **pages;
 501         int num_pages;
 502         u64 len;
 503         int written = 0;
 504         int flags;
 505         int check_caps = 0;
 506         int page_align, io_align;
 507         unsigned long buf_align;
 508         int ret;
 509         struct timespec mtime = CURRENT_TIME;
 510         bool own_pages = false;
 511
 512         if (ceph_snap(file_inode(file)) != CEPH_NOSNAP)
 513                 return -EROFS;
 514
 515         dout("sync_write on file %p %lld~%u %s\n", file, pos,
 516              (unsigned)left, (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
 517
 518         ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + left);
 519         if (ret < 0)
 520                 return ret;
 521
 522         ret = invalidate_inode_pages2_range(inode->i_mapping,
 523                                             pos >> PAGE_CACHE_SHIFT,
 524                                             (pos + left) >> PAGE_CACHE_SHIFT);
 525         if (ret < 0)
 526                 dout("invalidate_inode_pages2_range returned %d\n", ret);
 527
 528         flags = CEPH_OSD_FLAG_ORDERSNAP |
 529                 CEPH_OSD_FLAG_ONDISK |
 530                 CEPH_OSD_FLAG_WRITE;
 531         if ((file->f_flags & (O_SYNC|O_DIRECT)) == 0)
 532                 flags |= CEPH_OSD_FLAG_ACK;
 533         else
 534                 num_ops++;      /* Also include a 'startsync' command. */
 535
 536         /*
 537          * we may need to do multiple writes here if we span an object
 538          * boundary.  this isn't atomic, unfortunately.  :(
 539          */
 540 more:
 541         io_align = pos & ~PAGE_MASK;
 542         buf_align = (unsigned long)data & ~PAGE_MASK;
 543         len = left;
 544
 545         snapc = ci->i_snap_realm->cached_context;
 546         vino = ceph_vino(inode);
 547         req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
 548                                     vino, pos, &len, num_ops,
 549                                     CEPH_OSD_OP_WRITE, flags, snapc,
 550                                     ci->i_truncate_seq, ci->i_truncate_size,
 551                                     false);
 552         if (IS_ERR(req))
 553                 return PTR_ERR(req);
 554
 555         /* write from beginning of first page, regardless of io alignment */
 556         page_align = file->f_flags & O_DIRECT ? buf_align : io_align;
 557         num_pages = calc_pages_for(page_align, len);
 558         if (file->f_flags & O_DIRECT) {
 559                 pages = ceph_get_direct_page_vector(data, num_pages, false);
 560                 if (IS_ERR(pages)) {
 561                         ret = PTR_ERR(pages);
 562                         goto out;
 563                 }
 564
 565                 /*
 566                  * throw out any page cache pages in this range. this
 567                  * may block.
 568                  */
 569                 truncate_inode_pages_range(inode->i_mapping, pos,
 570                                            (pos+len) | (PAGE_CACHE_SIZE-1));
 571         } else {
 572                 pages = ceph_alloc_page_vector(num_pages, GFP_NOFS);
 573                 if (IS_ERR(pages)) {
 574                         ret = PTR_ERR(pages);
 575                         goto out;
 576                 }
 577                 ret = ceph_copy_user_to_page_vector(pages, data, pos, len);
 578                 if (ret < 0) {
 579                         ceph_release_page_vector(pages, num_pages);
 580                         goto out;
 581                 }
 582
 583                 if ((file->f_flags & O_SYNC) == 0) {
 584                         /* get a second commit callback */
 585                         req->r_unsafe_callback = ceph_sync_write_unsafe;
 586                         req->r_inode = inode;
 587                         own_pages = true;
 588                 }
 589         }
 590         osd_req_op_extent_osd_data_pages(req, 0, pages, len, page_align,
 591                                         false, own_pages);
 592
 593         /* BUG_ON(vino.snap != CEPH_NOSNAP); */
 594         ceph_osdc_build_request(req, pos, snapc, vino.snap, &mtime);
 595
 596         ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
 597         if (!ret)
 598                 ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
 599
 600         if (file->f_flags & O_DIRECT)
 601                 ceph_put_page_vector(pages, num_pages, false);
 602         else if (file->f_flags & O_SYNC)
 603                 ceph_release_page_vector(pages, num_pages);
 604
 605 out:
 606         ceph_osdc_put_request(req);
 607         if (ret == 0) {
 608                 pos += len;
 609                 written += len;
 610                 left -= len;
 611                 data += len;
 612                 if (left)
 613                         goto more;
 614
 615                 ret = written;
 616                 *ppos = pos;
 617                 if (pos > i_size_read(inode))
 618                         check_caps = ceph_inode_set_size(inode, pos);
 619                 if (check_caps)
 620                         ceph_check_caps(ceph_inode(inode), CHECK_CAPS_AUTHONLY,
 621                                         NULL);
 622         }
 623         return ret;
 624 }
 625
 626 /*
 627  * Wrap generic_file_aio_read with checks for cap bits on the inode.
 628  * Atomically grab references, so that those bits are not released
 629  * back to the MDS mid-read.
 630  *
 631  * Hmm, the sync read case isn't actually async... should it be?
 632  */
 633 static ssize_t ceph_aio_read(struct kiocb *iocb, const struct iovec *iov,
 634                              unsigned long nr_segs, loff_t pos)
 635 {
 636         struct file *filp = iocb->ki_filp;
 637         struct ceph_file_info *fi = filp->private_data;
 638         loff_t *ppos = &iocb->ki_pos;
 639         size_t len = iov->iov_len;
 640         struct inode *inode = file_inode(filp);
 641         struct ceph_inode_info *ci = ceph_inode(inode);
 642         void __user *base = iov->iov_base;
 643         ssize_t ret;
 644         int want, got = 0;
 645         int checkeof = 0, read = 0;
 646
 647         dout("aio_read %p %llx.%llx %llu~%u trying to get caps on %p\n",
 648              inode, ceph_vinop(inode), pos, (unsigned)len, inode);
 649 again:
 650         if (fi->fmode & CEPH_FILE_MODE_LAZY)
 651                 want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
 652         else
 653                 want = CEPH_CAP_FILE_CACHE;
 654         ret = ceph_get_caps(ci, CEPH_CAP_FILE_RD, want, &got, -1);
 655         if (ret < 0)
 656                 goto out;
 657         dout("aio_read %p %llx.%llx %llu~%u got cap refs on %s\n",
 658              inode, ceph_vinop(inode), pos, (unsigned)len,
 659              ceph_cap_string(got));
 660
 661         if ((got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0 ||
 662             (iocb->ki_filp->f_flags & O_DIRECT) ||
 663             (fi->flags & CEPH_F_SYNC))
 664                 /* hmm, this isn't really async... */
 665                 ret = ceph_sync_read(filp, base, len, ppos, &checkeof);
 666         else
 667                 ret = generic_file_aio_read(iocb, iov, nr_segs, pos);
 668
 669 out:
 670         dout("aio_read %p %llx.%llx dropping cap refs on %s = %d\n",
 671              inode, ceph_vinop(inode), ceph_cap_string(got), (int)ret);
 672         ceph_put_cap_refs(ci, got);
 673
 674         if (checkeof && ret >= 0) {
 675                 int statret = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE);
 676
 677                 /* hit EOF or hole? */
 678                 if (statret == 0 && *ppos < inode->i_size) {
 679                         dout("aio_read sync_read hit hole, ppos %lld < size %lld, reading more\n", *ppos, inode->i_size);
 680                         read += ret;
 681                         base += ret;
 682                         len -= ret;
 683                         checkeof = 0;
 684                         goto again;
 685                 }
 686         }
 687         if (ret >= 0)
 688                 ret += read;
 689
 690         return ret;
 691 }
 692
 693 /*
 694  * Take cap references to avoid releasing caps to MDS mid-write.
 695  *
 696  * If we are synchronous, and write with an old snap context, the OSD
 697  * may return EOLDSNAPC.  In that case, retry the write.. _after_
 698  * dropping our cap refs and allowing the pending snap to logically
 699  * complete _before_ this write occurs.
 700  *
 701  * If we are near ENOSPC, write synchronously.
 702  */
 703 static ssize_t ceph_aio_write(struct kiocb *iocb, const struct iovec *iov,
 704                        unsigned long nr_segs, loff_t pos)
 705 {
 706         struct file *file = iocb->ki_filp;
 707         struct ceph_file_info *fi = file->private_data;
 708         struct inode *inode = file_inode(file);
 709         struct ceph_inode_info *ci = ceph_inode(inode);
 710         struct ceph_osd_client *osdc =
 711                 &ceph_sb_to_client(inode->i_sb)->client->osdc;
 712         ssize_t count, written = 0;
 713         int err, want, got;
 714
 715         if (ceph_snap(inode) != CEPH_NOSNAP)
 716                 return -EROFS;
 717
 718         mutex_lock(&inode->i_mutex);
 719
 720         err = generic_segment_checks(iov, &nr_segs, &count, VERIFY_READ);
 721         if (err)
 722                 goto out;
 723
 724         /* We can write back this queue in page reclaim */
 725         current->backing_dev_info = file->f_mapping->backing_dev_info;
 726
 727         err = generic_write_checks(file, &pos, &count, S_ISBLK(inode->i_mode));
 728         if (err)
 729                 goto out;
 730
 731         if (count == 0)
 732                 goto out;
 733
 734         err = file_remove_suid(file);
 735         if (err)
 736                 goto out;
 737
 738         err = file_update_time(file);
 739         if (err)
 740                 goto out;
 741
 742 retry_snap:
 743         if (ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_FULL)) {
 744                 err = -ENOSPC;
 745                 goto out;
 746         }
 747
 748         dout("aio_write %p %llx.%llx %llu~%zd getting caps. i_size %llu\n",
 749              inode, ceph_vinop(inode), pos, count, inode->i_size);
 750         if (fi->fmode & CEPH_FILE_MODE_LAZY)
 751                 want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
 752         else
 753                 want = CEPH_CAP_FILE_BUFFER;
 754         got = 0;
 755         err = ceph_get_caps(ci, CEPH_CAP_FILE_WR, want, &got, pos + count);
 756         if (err < 0)
 757                 goto out;
 758
 759         dout("aio_write %p %llx.%llx %llu~%zd got cap refs on %s\n",
 760              inode, ceph_vinop(inode), pos, count, ceph_cap_string(got));
 761
 762         if ((got & (CEPH_CAP_FILE_BUFFER|CEPH_CAP_FILE_LAZYIO)) == 0 ||
 763             (iocb->ki_filp->f_flags & O_DIRECT) ||
 764             (fi->flags & CEPH_F_SYNC)) {
 765                 mutex_unlock(&inode->i_mutex);
 766                 written = ceph_sync_write(file, iov->iov_base, count,
 767                                           pos, &iocb->ki_pos);
 768                 if (written == -EOLDSNAPC) {
 769                         dout("aio_write %p %llx.%llx %llu~%u"
 770                                 "got EOLDSNAPC, retrying\n",
 771                                 inode, ceph_vinop(inode),
 772                                 pos, (unsigned)iov->iov_len);
 773                         mutex_lock(&inode->i_mutex);
 774                         goto retry_snap;
 775                 }
 776         } else {
 777                 /*
 778                  * No need to acquire the i_truncate_mutex. Because
 779                  * the MDS revokes Fwb caps before sending truncate
 780                  * message to us. We can't get Fwb cap while there
 781                  * are pending vmtruncate. So write and vmtruncate
 782                  * can not run at the same time
 783                  */
 784                 written = generic_file_buffered_write(iocb, iov, nr_segs,
 785                                                       pos, &iocb->ki_pos,
 786                                                       count, 0);
 787                 mutex_unlock(&inode->i_mutex);
 788         }
 789
 790         if (written >= 0) {
 791                 int dirty;
 792                 spin_lock(&ci->i_ceph_lock);
 793                 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR);
 794                 spin_unlock(&ci->i_ceph_lock);
 795                 if (dirty)
 796                         __mark_inode_dirty(inode, dirty);
 797         }
 798
 799         dout("aio_write %p %llx.%llx %llu~%u  dropping cap refs on %s\n",
 800              inode, ceph_vinop(inode), pos, (unsigned)iov->iov_len,
 801              ceph_cap_string(got));
 802         ceph_put_cap_refs(ci, got);
 803
 804         if (written >= 0 &&
 805             ((file->f_flags & O_SYNC) || IS_SYNC(file->f_mapping->host) ||
 806              ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_NEARFULL))) {
 807                 err = vfs_fsync_range(file, pos, pos + written - 1, 1);
 808                 if (err < 0)
 809                         written = err;
 810         }
 811
 812         goto out_unlocked;
 813
 814 out:
 815         mutex_unlock(&inode->i_mutex);
 816 out_unlocked:
 817         current->backing_dev_info = NULL;
 818         return written ? written : err;
 819 }
 820
 821 /*
 822  * llseek.  be sure to verify file size on SEEK_END.
 823  */
 824 static loff_t ceph_llseek(struct file *file, loff_t offset, int whence)
 825 {
 826         struct inode *inode = file->f_mapping->host;
 827         int ret;
 828
 829         mutex_lock(&inode->i_mutex);
 830
 831         if (whence == SEEK_END || whence == SEEK_DATA || whence == SEEK_HOLE) {
 832                 ret = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE);
 833                 if (ret < 0) {
 834                         offset = ret;
 835                         goto out;
 836                 }
 837         }
 838
 839         switch (whence) {
 840         case SEEK_END:
 841                 offset += inode->i_size;
 842                 break;
 843         case SEEK_CUR:
 844                 /*
 845                  * Here we special-case the lseek(fd, 0, SEEK_CUR)
 846                  * position-querying operation.  Avoid rewriting the "same"
 847                  * f_pos value back to the file because a concurrent read(),
 848                  * write() or lseek() might have altered it
 849                  */
 850                 if (offset == 0) {
 851                         offset = file->f_pos;
 852                         goto out;
 853                 }
 854                 offset += file->f_pos;
 855                 break;
 856         case SEEK_DATA:
 857                 if (offset >= inode->i_size) {
 858                         ret = -ENXIO;
 859                         goto out;
 860                 }
 861                 break;
 862         case SEEK_HOLE:
 863                 if (offset >= inode->i_size) {
 864                         ret = -ENXIO;
 865                         goto out;
 866                 }
 867                 offset = inode->i_size;
 868                 break;
 869         }
 870
 871         offset = vfs_setpos(file, offset, inode->i_sb->s_maxbytes);
 872
 873 out:
 874         mutex_unlock(&inode->i_mutex);
 875         return offset;
 876 }
 877
 878 static inline void ceph_zero_partial_page(
 879         struct inode *inode, loff_t offset, unsigned size)
 880 {
 881         struct page *page;
 882         pgoff_t index = offset >> PAGE_CACHE_SHIFT;
 883
 884         page = find_lock_page(inode->i_mapping, index);
 885         if (page) {
 886                 wait_on_page_writeback(page);
 887                 zero_user(page, offset & (PAGE_CACHE_SIZE - 1), size);
 888                 unlock_page(page);
 889                 page_cache_release(page);
 890         }
 891 }
 892
 893 static void ceph_zero_pagecache_range(struct inode *inode, loff_t offset,
 894                                       loff_t length)
 895 {
 896         loff_t nearly = round_up(offset, PAGE_CACHE_SIZE);
 897         if (offset < nearly) {
 898                 loff_t size = nearly - offset;
 899                 if (length < size)
 900                         size = length;
 901                 ceph_zero_partial_page(inode, offset, size);
 902                 offset += size;
 903                 length -= size;
 904         }
 905         if (length >= PAGE_CACHE_SIZE) {
 906                 loff_t size = round_down(length, PAGE_CACHE_SIZE);
 907                 truncate_pagecache_range(inode, offset, offset + size - 1);
 908                 offset += size;
 909                 length -= size;
 910         }
 911         if (length)
 912                 ceph_zero_partial_page(inode, offset, length);
 913 }
 914
 915 static int ceph_zero_partial_object(struct inode *inode,
 916                                     loff_t offset, loff_t *length)
 917 {
 918         struct ceph_inode_info *ci = ceph_inode(inode);
 919         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
 920         struct ceph_osd_request *req;
 921         int ret = 0;
 922         loff_t zero = 0;
 923         int op;
 924
 925         if (!length) {
 926                 op = offset ? CEPH_OSD_OP_DELETE : CEPH_OSD_OP_TRUNCATE;
 927                 length = &zero;
 928         } else {
 929                 op = CEPH_OSD_OP_ZERO;
 930         }
 931
 932         req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
 933                                         ceph_vino(inode),
 934                                         offset, length,
 935                                         1, op,
 936                                         CEPH_OSD_FLAG_WRITE |
 937                                         CEPH_OSD_FLAG_ONDISK,
 938                                         NULL, 0, 0, false);
 939         if (IS_ERR(req)) {
 940                 ret = PTR_ERR(req);
 941                 goto out;
 942         }
 943
 944         ceph_osdc_build_request(req, offset, NULL, ceph_vino(inode).snap,
 945                                 &inode->i_mtime);
 946
 947         ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
 948         if (!ret) {
 949                 ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
 950                 if (ret == -ENOENT)
 951                         ret = 0;
 952         }
 953         ceph_osdc_put_request(req);
 954
 955 out:
 956         return ret;
 957 }
 958
 959 static int ceph_zero_objects(struct inode *inode, loff_t offset, loff_t length)
 960 {
 961         int ret = 0;
 962         struct ceph_inode_info *ci = ceph_inode(inode);
 963         __s32 stripe_unit = ceph_file_layout_su(ci->i_layout);
 964         __s32 stripe_count = ceph_file_layout_stripe_count(ci->i_layout);
 965         __s32 object_size = ceph_file_layout_object_size(ci->i_layout);
 966         loff_t object_set_size = (loff_t)object_size * stripe_count;
 967
 968         loff_t nearly = (offset + object_set_size - 1)
 969                         / object_set_size * object_set_size;
 970         while (length && offset < nearly) {
 971                 loff_t size = length;
 972                 ret = ceph_zero_partial_object(inode, offset, &size);
 973                 if (ret < 0)
 974                         return ret;
 975                 offset += size;
 976                 length -= size;
 977         }
 978         while (length >= object_set_size) {
 979                 int i;
 980                 loff_t pos = offset;
 981                 for (i = 0; i < stripe_count; ++i) {
 982                         ret = ceph_zero_partial_object(inode, pos, NULL);
 983                         if (ret < 0)
 984                                 return ret;
 985                         pos += stripe_unit;
 986                 }
 987                 offset += object_set_size;
 988                 length -= object_set_size;
 989         }
 990         while (length) {
 991                 loff_t size = length;
 992                 ret = ceph_zero_partial_object(inode, offset, &size);
 993                 if (ret < 0)
 994                         return ret;
 995                 offset += size;
 996                 length -= size;
 997         }
 998         return ret;
 999 }
1000
1001 static long ceph_fallocate(struct file *file, int mode,
1002                                 loff_t offset, loff_t length)
1003 {
1004         struct ceph_file_info *fi = file->private_data;
1005         struct inode *inode = file->f_dentry->d_inode;
1006         struct ceph_inode_info *ci = ceph_inode(inode);
1007         struct ceph_osd_client *osdc =
1008                 &ceph_inode_to_client(inode)->client->osdc;
1009         int want, got = 0;
1010         int dirty;
1011         int ret = 0;
1012         loff_t endoff = 0;
1013         loff_t size;
1014
1015         if (!S_ISREG(inode->i_mode))
1016                 return -EOPNOTSUPP;
1017
1018         if (IS_SWAPFILE(inode))
1019                 return -ETXTBSY;
1020
1021         mutex_lock(&inode->i_mutex);
1022
1023         if (ceph_snap(inode) != CEPH_NOSNAP) {
1024                 ret = -EROFS;
1025                 goto unlock;
1026         }
1027
1028         if (ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_FULL) &&
1029                 !(mode & FALLOC_FL_PUNCH_HOLE)) {
1030                 ret = -ENOSPC;
1031                 goto unlock;
1032         }
1033
1034         size = i_size_read(inode);
1035         if (!(mode & FALLOC_FL_KEEP_SIZE))
1036                 endoff = offset + length;
1037
1038         if (fi->fmode & CEPH_FILE_MODE_LAZY)
1039                 want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
1040         else
1041                 want = CEPH_CAP_FILE_BUFFER;
1042
1043         ret = ceph_get_caps(ci, CEPH_CAP_FILE_WR, want, &got, endoff);
1044         if (ret < 0)
1045                 goto unlock;
1046
1047         if (mode & FALLOC_FL_PUNCH_HOLE) {
1048                 if (offset < size)
1049                         ceph_zero_pagecache_range(inode, offset, length);
1050                 ret = ceph_zero_objects(inode, offset, length);
1051         } else if (endoff > size) {
1052                 truncate_pagecache_range(inode, size, -1);
1053                 if (ceph_inode_set_size(inode, endoff))
1054                         ceph_check_caps(ceph_inode(inode),
1055                                 CHECK_CAPS_AUTHONLY, NULL);
1056         }
1057
1058         if (!ret) {
1059                 spin_lock(&ci->i_ceph_lock);
1060                 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR);
1061                 spin_unlock(&ci->i_ceph_lock);
1062                 if (dirty)
1063                         __mark_inode_dirty(inode, dirty);
1064         }
1065
1066         ceph_put_cap_refs(ci, got);
1067 unlock:
1068         mutex_unlock(&inode->i_mutex);
1069         return ret;
1070 }
1071
1072 const struct file_operations ceph_file_fops = {
1073         .open = ceph_open,
1074         .release = ceph_release,
1075         .llseek = ceph_llseek,
1076         .read = do_sync_read,
1077         .write = do_sync_write,
1078         .aio_read = ceph_aio_read,
1079         .aio_write = ceph_aio_write,
1080         .mmap = ceph_mmap,
1081         .fsync = ceph_fsync,
1082         .lock = ceph_lock,
1083         .flock = ceph_flock,
1084         .splice_read = generic_file_splice_read,
1085         .splice_write = generic_file_splice_write,
1086         .unlocked_ioctl = ceph_ioctl,
1087         .compat_ioctl   = ceph_ioctl,
1088         .fallocate      = ceph_fallocate,
1089 };
1090