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/namei.h> 8 #include <linux/writeback.h> 9 10 #include "super.h" 11 #include "mds_client.h" 12 13 /* 14 * Ceph file operations 15 * 16 * Implement basic open/close functionality, and implement 17 * read/write. 18 * 19 * We implement three modes of file I/O: 20 * - buffered uses the generic_file_aio_{read,write} helpers 21 * 22 * - synchronous is used when there is multi-client read/write 23 * sharing, avoids the page cache, and synchronously waits for an 24 * ack from the OSD. 25 * 26 * - direct io takes the variant of the sync path that references 27 * user pages directly. 28 * 29 * fsync() flushes and waits on dirty pages, but just queues metadata 30 * for writeback: since the MDS can recover size and mtime there is no 31 * need to wait for MDS acknowledgement. 32 */ 33 34 35 /* 36 * Prepare an open request. Preallocate ceph_cap to avoid an 37 * inopportune ENOMEM later. 38 */ 39 static struct ceph_mds_request * 40 prepare_open_request(struct super_block *sb, int flags, int create_mode) 41 { 42 struct ceph_fs_client *fsc = ceph_sb_to_client(sb); 43 struct ceph_mds_client *mdsc = fsc->mdsc; 44 struct ceph_mds_request *req; 45 int want_auth = USE_ANY_MDS; 46 int op = (flags & O_CREAT) ? CEPH_MDS_OP_CREATE : CEPH_MDS_OP_OPEN; 47 48 if (flags & (O_WRONLY|O_RDWR|O_CREAT|O_TRUNC)) 49 want_auth = USE_AUTH_MDS; 50 51 req = ceph_mdsc_create_request(mdsc, op, want_auth); 52 if (IS_ERR(req)) 53 goto out; 54 req->r_fmode = ceph_flags_to_mode(flags); 55 req->r_args.open.flags = cpu_to_le32(flags); 56 req->r_args.open.mode = cpu_to_le32(create_mode); 57 req->r_args.open.preferred = cpu_to_le32(-1); 58 out: 59 return req; 60 } 61 62 /* 63 * initialize private struct file data. 64 * if we fail, clean up by dropping fmode reference on the ceph_inode 65 */ 66 static int ceph_init_file(struct inode *inode, struct file *file, int fmode) 67 { 68 struct ceph_file_info *cf; 69 int ret = 0; 70 71 switch (inode->i_mode & S_IFMT) { 72 case S_IFREG: 73 case S_IFDIR: 74 dout("init_file %p %p 0%o (regular)\n", inode, file, 75 inode->i_mode); 76 cf = kmem_cache_alloc(ceph_file_cachep, GFP_NOFS | __GFP_ZERO); 77 if (cf == NULL) { 78 ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */ 79 return -ENOMEM; 80 } 81 cf->fmode = fmode; 82 cf->next_offset = 2; 83 file->private_data = cf; 84 BUG_ON(inode->i_fop->release != ceph_release); 85 break; 86 87 case S_IFLNK: 88 dout("init_file %p %p 0%o (symlink)\n", inode, file, 89 inode->i_mode); 90 ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */ 91 break; 92 93 default: 94 dout("init_file %p %p 0%o (special)\n", inode, file, 95 inode->i_mode); 96 /* 97 * we need to drop the open ref now, since we don't 98 * have .release set to ceph_release. 99 */ 100 ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */ 101 BUG_ON(inode->i_fop->release == ceph_release); 102 103 /* call the proper open fop */ 104 ret = inode->i_fop->open(inode, file); 105 } 106 return ret; 107 } 108 109 /* 110 * If the filp already has private_data, that means the file was 111 * already opened by intent during lookup, and we do nothing. 112 * 113 * If we already have the requisite capabilities, we can satisfy 114 * the open request locally (no need to request new caps from the 115 * MDS). We do, however, need to inform the MDS (asynchronously) 116 * if our wanted caps set expands. 117 */ 118 int ceph_open(struct inode *inode, struct file *file) 119 { 120 struct ceph_inode_info *ci = ceph_inode(inode); 121 struct ceph_fs_client *fsc = ceph_sb_to_client(inode->i_sb); 122 struct ceph_mds_client *mdsc = fsc->mdsc; 123 struct ceph_mds_request *req; 124 struct ceph_file_info *cf = file->private_data; 125 struct inode *parent_inode = NULL; 126 int err; 127 int flags, fmode, wanted; 128 129 if (cf) { 130 dout("open file %p is already opened\n", file); 131 return 0; 132 } 133 134 /* filter out O_CREAT|O_EXCL; vfs did that already. yuck. */ 135 flags = file->f_flags & ~(O_CREAT|O_EXCL); 136 if (S_ISDIR(inode->i_mode)) 137 flags = O_DIRECTORY; /* mds likes to know */ 138 139 dout("open inode %p ino %llx.%llx file %p flags %d (%d)\n", inode, 140 ceph_vinop(inode), file, flags, file->f_flags); 141 fmode = ceph_flags_to_mode(flags); 142 wanted = ceph_caps_for_mode(fmode); 143 144 /* snapped files are read-only */ 145 if (ceph_snap(inode) != CEPH_NOSNAP && (file->f_mode & FMODE_WRITE)) 146 return -EROFS; 147 148 /* trivially open snapdir */ 149 if (ceph_snap(inode) == CEPH_SNAPDIR) { 150 spin_lock(&inode->i_lock); 151 __ceph_get_fmode(ci, fmode); 152 spin_unlock(&inode->i_lock); 153 return ceph_init_file(inode, file, fmode); 154 } 155 156 /* 157 * No need to block if we have caps on the auth MDS (for 158 * write) or any MDS (for read). Update wanted set 159 * asynchronously. 160 */ 161 spin_lock(&inode->i_lock); 162 if (__ceph_is_any_real_caps(ci) && 163 (((fmode & CEPH_FILE_MODE_WR) == 0) || ci->i_auth_cap)) { 164 int mds_wanted = __ceph_caps_mds_wanted(ci); 165 int issued = __ceph_caps_issued(ci, NULL); 166 167 dout("open %p fmode %d want %s issued %s using existing\n", 168 inode, fmode, ceph_cap_string(wanted), 169 ceph_cap_string(issued)); 170 __ceph_get_fmode(ci, fmode); 171 spin_unlock(&inode->i_lock); 172 173 /* adjust wanted? */ 174 if ((issued & wanted) != wanted && 175 (mds_wanted & wanted) != wanted && 176 ceph_snap(inode) != CEPH_SNAPDIR) 177 ceph_check_caps(ci, 0, NULL); 178 179 return ceph_init_file(inode, file, fmode); 180 } else if (ceph_snap(inode) != CEPH_NOSNAP && 181 (ci->i_snap_caps & wanted) == wanted) { 182 __ceph_get_fmode(ci, fmode); 183 spin_unlock(&inode->i_lock); 184 return ceph_init_file(inode, file, fmode); 185 } 186 spin_unlock(&inode->i_lock); 187 188 dout("open fmode %d wants %s\n", fmode, ceph_cap_string(wanted)); 189 req = prepare_open_request(inode->i_sb, flags, 0); 190 if (IS_ERR(req)) { 191 err = PTR_ERR(req); 192 goto out; 193 } 194 req->r_inode = inode; 195 ihold(inode); 196 req->r_num_caps = 1; 197 if (flags & (O_CREAT|O_TRUNC)) 198 parent_inode = ceph_get_dentry_parent_inode(file->f_dentry); 199 err = ceph_mdsc_do_request(mdsc, parent_inode, req); 200 iput(parent_inode); 201 if (!err) 202 err = ceph_init_file(inode, file, req->r_fmode); 203 ceph_mdsc_put_request(req); 204 dout("open result=%d on %llx.%llx\n", err, ceph_vinop(inode)); 205 out: 206 return err; 207 } 208 209 210 /* 211 * Do a lookup + open with a single request. 212 * 213 * If this succeeds, but some subsequent check in the vfs 214 * may_open() fails, the struct *file gets cleaned up (i.e. 215 * ceph_release gets called). So fear not! 216 */ 217 /* 218 * flags 219 * path_lookup_open -> LOOKUP_OPEN 220 * path_lookup_create -> LOOKUP_OPEN|LOOKUP_CREATE 221 */ 222 struct dentry *ceph_lookup_open(struct inode *dir, struct dentry *dentry, 223 struct nameidata *nd, int mode, 224 int locked_dir) 225 { 226 struct ceph_fs_client *fsc = ceph_sb_to_client(dir->i_sb); 227 struct ceph_mds_client *mdsc = fsc->mdsc; 228 struct file *file; 229 struct ceph_mds_request *req; 230 struct dentry *ret; 231 int err; 232 int flags = nd->intent.open.flags; 233 234 dout("ceph_lookup_open dentry %p '%.*s' flags %d mode 0%o\n", 235 dentry, dentry->d_name.len, dentry->d_name.name, flags, mode); 236 237 /* do the open */ 238 req = prepare_open_request(dir->i_sb, flags, mode); 239 if (IS_ERR(req)) 240 return ERR_CAST(req); 241 req->r_dentry = dget(dentry); 242 req->r_num_caps = 2; 243 if (flags & O_CREAT) { 244 req->r_dentry_drop = CEPH_CAP_FILE_SHARED; 245 req->r_dentry_unless = CEPH_CAP_FILE_EXCL; 246 } 247 req->r_locked_dir = dir; /* caller holds dir->i_mutex */ 248 err = ceph_mdsc_do_request(mdsc, 249 (flags & (O_CREAT|O_TRUNC)) ? dir : NULL, 250 req); 251 err = ceph_handle_snapdir(req, dentry, err); 252 if (err) 253 goto out; 254 if ((flags & O_CREAT) && !req->r_reply_info.head->is_dentry) 255 err = ceph_handle_notrace_create(dir, dentry); 256 if (err) 257 goto out; 258 file = lookup_instantiate_filp(nd, req->r_dentry, ceph_open); 259 if (IS_ERR(file)) 260 err = PTR_ERR(file); 261 out: 262 ret = ceph_finish_lookup(req, dentry, err); 263 ceph_mdsc_put_request(req); 264 dout("ceph_lookup_open result=%p\n", ret); 265 return ret; 266 } 267 268 int ceph_release(struct inode *inode, struct file *file) 269 { 270 struct ceph_inode_info *ci = ceph_inode(inode); 271 struct ceph_file_info *cf = file->private_data; 272 273 dout("release inode %p file %p\n", inode, file); 274 ceph_put_fmode(ci, cf->fmode); 275 if (cf->last_readdir) 276 ceph_mdsc_put_request(cf->last_readdir); 277 kfree(cf->last_name); 278 kfree(cf->dir_info); 279 dput(cf->dentry); 280 kmem_cache_free(ceph_file_cachep, cf); 281 282 /* wake up anyone waiting for caps on this inode */ 283 wake_up_all(&ci->i_cap_wq); 284 return 0; 285 } 286 287 /* 288 * Read a range of bytes striped over one or more objects. Iterate over 289 * objects we stripe over. (That's not atomic, but good enough for now.) 290 * 291 * If we get a short result from the OSD, check against i_size; we need to 292 * only return a short read to the caller if we hit EOF. 293 */ 294 static int striped_read(struct inode *inode, 295 u64 off, u64 len, 296 struct page **pages, int num_pages, 297 int *checkeof, bool o_direct, 298 unsigned long buf_align) 299 { 300 struct ceph_fs_client *fsc = ceph_inode_to_client(inode); 301 struct ceph_inode_info *ci = ceph_inode(inode); 302 u64 pos, this_len; 303 int io_align, page_align; 304 int left, pages_left; 305 int read; 306 struct page **page_pos; 307 int ret; 308 bool hit_stripe, was_short; 309 310 /* 311 * we may need to do multiple reads. not atomic, unfortunately. 312 */ 313 pos = off; 314 left = len; 315 page_pos = pages; 316 pages_left = num_pages; 317 read = 0; 318 io_align = off & ~PAGE_MASK; 319 320 more: 321 if (o_direct) 322 page_align = (pos - io_align + buf_align) & ~PAGE_MASK; 323 else 324 page_align = pos & ~PAGE_MASK; 325 this_len = left; 326 ret = ceph_osdc_readpages(&fsc->client->osdc, ceph_vino(inode), 327 &ci->i_layout, pos, &this_len, 328 ci->i_truncate_seq, 329 ci->i_truncate_size, 330 page_pos, pages_left, page_align); 331 if (ret == -ENOENT) 332 ret = 0; 333 hit_stripe = this_len < left; 334 was_short = ret >= 0 && ret < this_len; 335 dout("striped_read %llu~%u (read %u) got %d%s%s\n", pos, left, read, 336 ret, hit_stripe ? " HITSTRIPE" : "", was_short ? " SHORT" : ""); 337 338 if (ret > 0) { 339 int didpages = (page_align + ret) >> PAGE_CACHE_SHIFT; 340 341 if (read < pos - off) { 342 dout(" zero gap %llu to %llu\n", off + read, pos); 343 ceph_zero_page_vector_range(page_align + read, 344 pos - off - read, pages); 345 } 346 pos += ret; 347 read = pos - off; 348 left -= ret; 349 page_pos += didpages; 350 pages_left -= didpages; 351 352 /* hit stripe? */ 353 if (left && hit_stripe) 354 goto more; 355 } 356 357 if (was_short) { 358 /* did we bounce off eof? */ 359 if (pos + left > inode->i_size) 360 *checkeof = 1; 361 362 /* zero trailing bytes (inside i_size) */ 363 if (left > 0 && pos < inode->i_size) { 364 if (pos + left > inode->i_size) 365 left = inode->i_size - pos; 366 367 dout("zero tail %d\n", left); 368 ceph_zero_page_vector_range(page_align + read, left, 369 pages); 370 read += left; 371 } 372 } 373 374 if (ret >= 0) 375 ret = read; 376 dout("striped_read returns %d\n", ret); 377 return ret; 378 } 379 380 /* 381 * Completely synchronous read and write methods. Direct from __user 382 * buffer to osd, or directly to user pages (if O_DIRECT). 383 * 384 * If the read spans object boundary, just do multiple reads. 385 */ 386 static ssize_t ceph_sync_read(struct file *file, char __user *data, 387 unsigned len, loff_t *poff, int *checkeof) 388 { 389 struct inode *inode = file->f_dentry->d_inode; 390 struct page **pages; 391 u64 off = *poff; 392 int num_pages, ret; 393 394 dout("sync_read on file %p %llu~%u %s\n", file, off, len, 395 (file->f_flags & O_DIRECT) ? "O_DIRECT" : ""); 396 397 if (file->f_flags & O_DIRECT) { 398 num_pages = calc_pages_for((unsigned long)data, len); 399 pages = ceph_get_direct_page_vector(data, num_pages, true); 400 } else { 401 num_pages = calc_pages_for(off, len); 402 pages = ceph_alloc_page_vector(num_pages, GFP_NOFS); 403 } 404 if (IS_ERR(pages)) 405 return PTR_ERR(pages); 406 407 /* 408 * flush any page cache pages in this range. this 409 * will make concurrent normal and sync io slow, 410 * but it will at least behave sensibly when they are 411 * in sequence. 412 */ 413 ret = filemap_write_and_wait(inode->i_mapping); 414 if (ret < 0) 415 goto done; 416 417 ret = striped_read(inode, off, len, pages, num_pages, checkeof, 418 file->f_flags & O_DIRECT, 419 (unsigned long)data & ~PAGE_MASK); 420 421 if (ret >= 0 && (file->f_flags & O_DIRECT) == 0) 422 ret = ceph_copy_page_vector_to_user(pages, data, off, ret); 423 if (ret >= 0) 424 *poff = off + ret; 425 426 done: 427 if (file->f_flags & O_DIRECT) 428 ceph_put_page_vector(pages, num_pages, true); 429 else 430 ceph_release_page_vector(pages, num_pages); 431 dout("sync_read result %d\n", ret); 432 return ret; 433 } 434 435 /* 436 * Write commit callback, called if we requested both an ACK and 437 * ONDISK commit reply from the OSD. 438 */ 439 static void sync_write_commit(struct ceph_osd_request *req, 440 struct ceph_msg *msg) 441 { 442 struct ceph_inode_info *ci = ceph_inode(req->r_inode); 443 444 dout("sync_write_commit %p tid %llu\n", req, req->r_tid); 445 spin_lock(&ci->i_unsafe_lock); 446 list_del_init(&req->r_unsafe_item); 447 spin_unlock(&ci->i_unsafe_lock); 448 ceph_put_cap_refs(ci, CEPH_CAP_FILE_WR); 449 } 450 451 /* 452 * Synchronous write, straight from __user pointer or user pages (if 453 * O_DIRECT). 454 * 455 * If write spans object boundary, just do multiple writes. (For a 456 * correct atomic write, we should e.g. take write locks on all 457 * objects, rollback on failure, etc.) 458 */ 459 static ssize_t ceph_sync_write(struct file *file, const char __user *data, 460 size_t left, loff_t *offset) 461 { 462 struct inode *inode = file->f_dentry->d_inode; 463 struct ceph_inode_info *ci = ceph_inode(inode); 464 struct ceph_fs_client *fsc = ceph_inode_to_client(inode); 465 struct ceph_osd_request *req; 466 struct page **pages; 467 int num_pages; 468 long long unsigned pos; 469 u64 len; 470 int written = 0; 471 int flags; 472 int do_sync = 0; 473 int check_caps = 0; 474 int page_align, io_align; 475 unsigned long buf_align; 476 int ret; 477 struct timespec mtime = CURRENT_TIME; 478 479 if (ceph_snap(file->f_dentry->d_inode) != CEPH_NOSNAP) 480 return -EROFS; 481 482 dout("sync_write on file %p %lld~%u %s\n", file, *offset, 483 (unsigned)left, (file->f_flags & O_DIRECT) ? "O_DIRECT" : ""); 484 485 if (file->f_flags & O_APPEND) 486 pos = i_size_read(inode); 487 else 488 pos = *offset; 489 490 ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + left); 491 if (ret < 0) 492 return ret; 493 494 ret = invalidate_inode_pages2_range(inode->i_mapping, 495 pos >> PAGE_CACHE_SHIFT, 496 (pos + left) >> PAGE_CACHE_SHIFT); 497 if (ret < 0) 498 dout("invalidate_inode_pages2_range returned %d\n", ret); 499 500 flags = CEPH_OSD_FLAG_ORDERSNAP | 501 CEPH_OSD_FLAG_ONDISK | 502 CEPH_OSD_FLAG_WRITE; 503 if ((file->f_flags & (O_SYNC|O_DIRECT)) == 0) 504 flags |= CEPH_OSD_FLAG_ACK; 505 else 506 do_sync = 1; 507 508 /* 509 * we may need to do multiple writes here if we span an object 510 * boundary. this isn't atomic, unfortunately. :( 511 */ 512 more: 513 io_align = pos & ~PAGE_MASK; 514 buf_align = (unsigned long)data & ~PAGE_MASK; 515 len = left; 516 if (file->f_flags & O_DIRECT) { 517 /* write from beginning of first page, regardless of 518 io alignment */ 519 page_align = (pos - io_align + buf_align) & ~PAGE_MASK; 520 num_pages = calc_pages_for((unsigned long)data, len); 521 } else { 522 page_align = pos & ~PAGE_MASK; 523 num_pages = calc_pages_for(pos, len); 524 } 525 req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout, 526 ceph_vino(inode), pos, &len, 527 CEPH_OSD_OP_WRITE, flags, 528 ci->i_snap_realm->cached_context, 529 do_sync, 530 ci->i_truncate_seq, ci->i_truncate_size, 531 &mtime, false, 2, page_align); 532 if (!req) 533 return -ENOMEM; 534 535 if (file->f_flags & O_DIRECT) { 536 pages = ceph_get_direct_page_vector(data, num_pages, false); 537 if (IS_ERR(pages)) { 538 ret = PTR_ERR(pages); 539 goto out; 540 } 541 542 /* 543 * throw out any page cache pages in this range. this 544 * may block. 545 */ 546 truncate_inode_pages_range(inode->i_mapping, pos, 547 (pos+len) | (PAGE_CACHE_SIZE-1)); 548 } else { 549 pages = ceph_alloc_page_vector(num_pages, GFP_NOFS); 550 if (IS_ERR(pages)) { 551 ret = PTR_ERR(pages); 552 goto out; 553 } 554 ret = ceph_copy_user_to_page_vector(pages, data, pos, len); 555 if (ret < 0) { 556 ceph_release_page_vector(pages, num_pages); 557 goto out; 558 } 559 560 if ((file->f_flags & O_SYNC) == 0) { 561 /* get a second commit callback */ 562 req->r_safe_callback = sync_write_commit; 563 req->r_own_pages = 1; 564 } 565 } 566 req->r_pages = pages; 567 req->r_num_pages = num_pages; 568 req->r_inode = inode; 569 570 ret = ceph_osdc_start_request(&fsc->client->osdc, req, false); 571 if (!ret) { 572 if (req->r_safe_callback) { 573 /* 574 * Add to inode unsafe list only after we 575 * start_request so that a tid has been assigned. 576 */ 577 spin_lock(&ci->i_unsafe_lock); 578 list_add_tail(&req->r_unsafe_item, 579 &ci->i_unsafe_writes); 580 spin_unlock(&ci->i_unsafe_lock); 581 ceph_get_cap_refs(ci, CEPH_CAP_FILE_WR); 582 } 583 584 ret = ceph_osdc_wait_request(&fsc->client->osdc, req); 585 if (ret < 0 && req->r_safe_callback) { 586 spin_lock(&ci->i_unsafe_lock); 587 list_del_init(&req->r_unsafe_item); 588 spin_unlock(&ci->i_unsafe_lock); 589 ceph_put_cap_refs(ci, CEPH_CAP_FILE_WR); 590 } 591 } 592 593 if (file->f_flags & O_DIRECT) 594 ceph_put_page_vector(pages, num_pages, false); 595 else if (file->f_flags & O_SYNC) 596 ceph_release_page_vector(pages, num_pages); 597 598 out: 599 ceph_osdc_put_request(req); 600 if (ret == 0) { 601 pos += len; 602 written += len; 603 left -= len; 604 data += written; 605 if (left) 606 goto more; 607 608 ret = written; 609 *offset = pos; 610 if (pos > i_size_read(inode)) 611 check_caps = ceph_inode_set_size(inode, pos); 612 if (check_caps) 613 ceph_check_caps(ceph_inode(inode), CHECK_CAPS_AUTHONLY, 614 NULL); 615 } 616 return ret; 617 } 618 619 /* 620 * Wrap generic_file_aio_read with checks for cap bits on the inode. 621 * Atomically grab references, so that those bits are not released 622 * back to the MDS mid-read. 623 * 624 * Hmm, the sync read case isn't actually async... should it be? 625 */ 626 static ssize_t ceph_aio_read(struct kiocb *iocb, const struct iovec *iov, 627 unsigned long nr_segs, loff_t pos) 628 { 629 struct file *filp = iocb->ki_filp; 630 struct ceph_file_info *fi = filp->private_data; 631 loff_t *ppos = &iocb->ki_pos; 632 size_t len = iov->iov_len; 633 struct inode *inode = filp->f_dentry->d_inode; 634 struct ceph_inode_info *ci = ceph_inode(inode); 635 void __user *base = iov->iov_base; 636 ssize_t ret; 637 int want, got = 0; 638 int checkeof = 0, read = 0; 639 640 dout("aio_read %p %llx.%llx %llu~%u trying to get caps on %p\n", 641 inode, ceph_vinop(inode), pos, (unsigned)len, inode); 642 again: 643 __ceph_do_pending_vmtruncate(inode); 644 if (fi->fmode & CEPH_FILE_MODE_LAZY) 645 want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO; 646 else 647 want = CEPH_CAP_FILE_CACHE; 648 ret = ceph_get_caps(ci, CEPH_CAP_FILE_RD, want, &got, -1); 649 if (ret < 0) 650 goto out; 651 dout("aio_read %p %llx.%llx %llu~%u got cap refs on %s\n", 652 inode, ceph_vinop(inode), pos, (unsigned)len, 653 ceph_cap_string(got)); 654 655 if ((got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0 || 656 (iocb->ki_filp->f_flags & O_DIRECT) || 657 (inode->i_sb->s_flags & MS_SYNCHRONOUS) || 658 (fi->flags & CEPH_F_SYNC)) 659 /* hmm, this isn't really async... */ 660 ret = ceph_sync_read(filp, base, len, ppos, &checkeof); 661 else 662 ret = generic_file_aio_read(iocb, iov, nr_segs, pos); 663 664 out: 665 dout("aio_read %p %llx.%llx dropping cap refs on %s = %d\n", 666 inode, ceph_vinop(inode), ceph_cap_string(got), (int)ret); 667 ceph_put_cap_refs(ci, got); 668 669 if (checkeof && ret >= 0) { 670 int statret = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE); 671 672 /* hit EOF or hole? */ 673 if (statret == 0 && *ppos < inode->i_size) { 674 dout("aio_read sync_read hit hole, ppos %lld < size %lld, reading more\n", *ppos, inode->i_size); 675 read += ret; 676 base += ret; 677 len -= ret; 678 checkeof = 0; 679 goto again; 680 } 681 } 682 if (ret >= 0) 683 ret += read; 684 685 return ret; 686 } 687 688 /* 689 * Take cap references to avoid releasing caps to MDS mid-write. 690 * 691 * If we are synchronous, and write with an old snap context, the OSD 692 * may return EOLDSNAPC. In that case, retry the write.. _after_ 693 * dropping our cap refs and allowing the pending snap to logically 694 * complete _before_ this write occurs. 695 * 696 * If we are near ENOSPC, write synchronously. 697 */ 698 static ssize_t ceph_aio_write(struct kiocb *iocb, const struct iovec *iov, 699 unsigned long nr_segs, loff_t pos) 700 { 701 struct file *file = iocb->ki_filp; 702 struct ceph_file_info *fi = file->private_data; 703 struct inode *inode = file->f_dentry->d_inode; 704 struct ceph_inode_info *ci = ceph_inode(inode); 705 struct ceph_osd_client *osdc = 706 &ceph_sb_to_client(inode->i_sb)->client->osdc; 707 loff_t endoff = pos + iov->iov_len; 708 int want, got = 0; 709 int ret, err; 710 711 if (ceph_snap(inode) != CEPH_NOSNAP) 712 return -EROFS; 713 714 retry_snap: 715 if (ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_FULL)) 716 return -ENOSPC; 717 __ceph_do_pending_vmtruncate(inode); 718 dout("aio_write %p %llx.%llx %llu~%u getting caps. i_size %llu\n", 719 inode, ceph_vinop(inode), pos, (unsigned)iov->iov_len, 720 inode->i_size); 721 if (fi->fmode & CEPH_FILE_MODE_LAZY) 722 want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO; 723 else 724 want = CEPH_CAP_FILE_BUFFER; 725 ret = ceph_get_caps(ci, CEPH_CAP_FILE_WR, want, &got, endoff); 726 if (ret < 0) 727 goto out_put; 728 729 dout("aio_write %p %llx.%llx %llu~%u got cap refs on %s\n", 730 inode, ceph_vinop(inode), pos, (unsigned)iov->iov_len, 731 ceph_cap_string(got)); 732 733 if ((got & (CEPH_CAP_FILE_BUFFER|CEPH_CAP_FILE_LAZYIO)) == 0 || 734 (iocb->ki_filp->f_flags & O_DIRECT) || 735 (inode->i_sb->s_flags & MS_SYNCHRONOUS) || 736 (fi->flags & CEPH_F_SYNC)) { 737 ret = ceph_sync_write(file, iov->iov_base, iov->iov_len, 738 &iocb->ki_pos); 739 } else { 740 /* 741 * buffered write; drop Fw early to avoid slow 742 * revocation if we get stuck on balance_dirty_pages 743 */ 744 int dirty; 745 746 spin_lock(&inode->i_lock); 747 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR); 748 spin_unlock(&inode->i_lock); 749 ceph_put_cap_refs(ci, got); 750 751 ret = generic_file_aio_write(iocb, iov, nr_segs, pos); 752 if ((ret >= 0 || ret == -EIOCBQUEUED) && 753 ((file->f_flags & O_SYNC) || IS_SYNC(file->f_mapping->host) 754 || ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_NEARFULL))) { 755 err = vfs_fsync_range(file, pos, pos + ret - 1, 1); 756 if (err < 0) 757 ret = err; 758 } 759 760 if (dirty) 761 __mark_inode_dirty(inode, dirty); 762 goto out; 763 } 764 765 if (ret >= 0) { 766 int dirty; 767 spin_lock(&inode->i_lock); 768 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR); 769 spin_unlock(&inode->i_lock); 770 if (dirty) 771 __mark_inode_dirty(inode, dirty); 772 } 773 774 out_put: 775 dout("aio_write %p %llx.%llx %llu~%u dropping cap refs on %s\n", 776 inode, ceph_vinop(inode), pos, (unsigned)iov->iov_len, 777 ceph_cap_string(got)); 778 ceph_put_cap_refs(ci, got); 779 780 out: 781 if (ret == -EOLDSNAPC) { 782 dout("aio_write %p %llx.%llx %llu~%u got EOLDSNAPC, retrying\n", 783 inode, ceph_vinop(inode), pos, (unsigned)iov->iov_len); 784 goto retry_snap; 785 } 786 787 return ret; 788 } 789 790 /* 791 * llseek. be sure to verify file size on SEEK_END. 792 */ 793 static loff_t ceph_llseek(struct file *file, loff_t offset, int origin) 794 { 795 struct inode *inode = file->f_mapping->host; 796 int ret; 797 798 mutex_lock(&inode->i_mutex); 799 __ceph_do_pending_vmtruncate(inode); 800 if (origin != SEEK_CUR || origin != SEEK_SET) { 801 ret = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE); 802 if (ret < 0) { 803 offset = ret; 804 goto out; 805 } 806 } 807 808 switch (origin) { 809 case SEEK_END: 810 offset += inode->i_size; 811 break; 812 case SEEK_CUR: 813 /* 814 * Here we special-case the lseek(fd, 0, SEEK_CUR) 815 * position-querying operation. Avoid rewriting the "same" 816 * f_pos value back to the file because a concurrent read(), 817 * write() or lseek() might have altered it 818 */ 819 if (offset == 0) { 820 offset = file->f_pos; 821 goto out; 822 } 823 offset += file->f_pos; 824 break; 825 case SEEK_DATA: 826 if (offset >= inode->i_size) { 827 ret = -ENXIO; 828 goto out; 829 } 830 break; 831 case SEEK_HOLE: 832 if (offset >= inode->i_size) { 833 ret = -ENXIO; 834 goto out; 835 } 836 offset = inode->i_size; 837 break; 838 } 839 840 if (offset < 0 || offset > inode->i_sb->s_maxbytes) { 841 offset = -EINVAL; 842 goto out; 843 } 844 845 /* Special lock needed here? */ 846 if (offset != file->f_pos) { 847 file->f_pos = offset; 848 file->f_version = 0; 849 } 850 851 out: 852 mutex_unlock(&inode->i_mutex); 853 return offset; 854 } 855 856 const struct file_operations ceph_file_fops = { 857 .open = ceph_open, 858 .release = ceph_release, 859 .llseek = ceph_llseek, 860 .read = do_sync_read, 861 .write = do_sync_write, 862 .aio_read = ceph_aio_read, 863 .aio_write = ceph_aio_write, 864 .mmap = ceph_mmap, 865 .fsync = ceph_fsync, 866 .lock = ceph_lock, 867 .flock = ceph_flock, 868 .splice_read = generic_file_splice_read, 869 .splice_write = generic_file_splice_write, 870 .unlocked_ioctl = ceph_ioctl, 871 .compat_ioctl = ceph_ioctl, 872 }; 873 874