1 #include <linux/ceph/ceph_debug.h> 2 3 #include <linux/backing-dev.h> 4 #include <linux/fs.h> 5 #include <linux/mm.h> 6 #include <linux/pagemap.h> 7 #include <linux/writeback.h> /* generic_writepages */ 8 #include <linux/slab.h> 9 #include <linux/pagevec.h> 10 #include <linux/task_io_accounting_ops.h> 11 12 #include "super.h" 13 #include "mds_client.h" 14 #include "cache.h" 15 #include <linux/ceph/osd_client.h> 16 17 /* 18 * Ceph address space ops. 19 * 20 * There are a few funny things going on here. 21 * 22 * The page->private field is used to reference a struct 23 * ceph_snap_context for _every_ dirty page. This indicates which 24 * snapshot the page was logically dirtied in, and thus which snap 25 * context needs to be associated with the osd write during writeback. 26 * 27 * Similarly, struct ceph_inode_info maintains a set of counters to 28 * count dirty pages on the inode. In the absence of snapshots, 29 * i_wrbuffer_ref == i_wrbuffer_ref_head == the dirty page count. 30 * 31 * When a snapshot is taken (that is, when the client receives 32 * notification that a snapshot was taken), each inode with caps and 33 * with dirty pages (dirty pages implies there is a cap) gets a new 34 * ceph_cap_snap in the i_cap_snaps list (which is sorted in ascending 35 * order, new snaps go to the tail). The i_wrbuffer_ref_head count is 36 * moved to capsnap->dirty. (Unless a sync write is currently in 37 * progress. In that case, the capsnap is said to be "pending", new 38 * writes cannot start, and the capsnap isn't "finalized" until the 39 * write completes (or fails) and a final size/mtime for the inode for 40 * that snap can be settled upon.) i_wrbuffer_ref_head is reset to 0. 41 * 42 * On writeback, we must submit writes to the osd IN SNAP ORDER. So, 43 * we look for the first capsnap in i_cap_snaps and write out pages in 44 * that snap context _only_. Then we move on to the next capsnap, 45 * eventually reaching the "live" or "head" context (i.e., pages that 46 * are not yet snapped) and are writing the most recently dirtied 47 * pages. 48 * 49 * Invalidate and so forth must take care to ensure the dirty page 50 * accounting is preserved. 51 */ 52 53 #define CONGESTION_ON_THRESH(congestion_kb) (congestion_kb >> (PAGE_SHIFT-10)) 54 #define CONGESTION_OFF_THRESH(congestion_kb) \ 55 (CONGESTION_ON_THRESH(congestion_kb) - \ 56 (CONGESTION_ON_THRESH(congestion_kb) >> 2)) 57 58 static inline struct ceph_snap_context *page_snap_context(struct page *page) 59 { 60 if (PagePrivate(page)) 61 return (void *)page->private; 62 return NULL; 63 } 64 65 /* 66 * Dirty a page. Optimistically adjust accounting, on the assumption 67 * that we won't race with invalidate. If we do, readjust. 68 */ 69 static int ceph_set_page_dirty(struct page *page) 70 { 71 struct address_space *mapping = page->mapping; 72 struct inode *inode; 73 struct ceph_inode_info *ci; 74 struct ceph_snap_context *snapc; 75 int ret; 76 77 if (unlikely(!mapping)) 78 return !TestSetPageDirty(page); 79 80 if (PageDirty(page)) { 81 dout("%p set_page_dirty %p idx %lu -- already dirty\n", 82 mapping->host, page, page->index); 83 BUG_ON(!PagePrivate(page)); 84 return 0; 85 } 86 87 inode = mapping->host; 88 ci = ceph_inode(inode); 89 90 /* dirty the head */ 91 spin_lock(&ci->i_ceph_lock); 92 BUG_ON(ci->i_wr_ref == 0); // caller should hold Fw reference 93 if (__ceph_have_pending_cap_snap(ci)) { 94 struct ceph_cap_snap *capsnap = 95 list_last_entry(&ci->i_cap_snaps, 96 struct ceph_cap_snap, 97 ci_item); 98 snapc = ceph_get_snap_context(capsnap->context); 99 capsnap->dirty_pages++; 100 } else { 101 BUG_ON(!ci->i_head_snapc); 102 snapc = ceph_get_snap_context(ci->i_head_snapc); 103 ++ci->i_wrbuffer_ref_head; 104 } 105 if (ci->i_wrbuffer_ref == 0) 106 ihold(inode); 107 ++ci->i_wrbuffer_ref; 108 dout("%p set_page_dirty %p idx %lu head %d/%d -> %d/%d " 109 "snapc %p seq %lld (%d snaps)\n", 110 mapping->host, page, page->index, 111 ci->i_wrbuffer_ref-1, ci->i_wrbuffer_ref_head-1, 112 ci->i_wrbuffer_ref, ci->i_wrbuffer_ref_head, 113 snapc, snapc->seq, snapc->num_snaps); 114 spin_unlock(&ci->i_ceph_lock); 115 116 /* 117 * Reference snap context in page->private. Also set 118 * PagePrivate so that we get invalidatepage callback. 119 */ 120 BUG_ON(PagePrivate(page)); 121 page->private = (unsigned long)snapc; 122 SetPagePrivate(page); 123 124 ret = __set_page_dirty_nobuffers(page); 125 WARN_ON(!PageLocked(page)); 126 WARN_ON(!page->mapping); 127 128 return ret; 129 } 130 131 /* 132 * If we are truncating the full page (i.e. offset == 0), adjust the 133 * dirty page counters appropriately. Only called if there is private 134 * data on the page. 135 */ 136 static void ceph_invalidatepage(struct page *page, unsigned int offset, 137 unsigned int length) 138 { 139 struct inode *inode; 140 struct ceph_inode_info *ci; 141 struct ceph_snap_context *snapc = page_snap_context(page); 142 143 inode = page->mapping->host; 144 ci = ceph_inode(inode); 145 146 if (offset != 0 || length != PAGE_SIZE) { 147 dout("%p invalidatepage %p idx %lu partial dirty page %u~%u\n", 148 inode, page, page->index, offset, length); 149 return; 150 } 151 152 ceph_invalidate_fscache_page(inode, page); 153 154 if (!PagePrivate(page)) 155 return; 156 157 /* 158 * We can get non-dirty pages here due to races between 159 * set_page_dirty and truncate_complete_page; just spit out a 160 * warning, in case we end up with accounting problems later. 161 */ 162 if (!PageDirty(page)) 163 pr_err("%p invalidatepage %p page not dirty\n", inode, page); 164 165 ClearPageChecked(page); 166 167 dout("%p invalidatepage %p idx %lu full dirty page\n", 168 inode, page, page->index); 169 170 ceph_put_wrbuffer_cap_refs(ci, 1, snapc); 171 ceph_put_snap_context(snapc); 172 page->private = 0; 173 ClearPagePrivate(page); 174 } 175 176 static int ceph_releasepage(struct page *page, gfp_t g) 177 { 178 dout("%p releasepage %p idx %lu (%sdirty)\n", page->mapping->host, 179 page, page->index, PageDirty(page) ? "" : "not "); 180 181 /* Can we release the page from the cache? */ 182 if (!ceph_release_fscache_page(page, g)) 183 return 0; 184 185 return !PagePrivate(page); 186 } 187 188 /* 189 * read a single page, without unlocking it. 190 */ 191 static int readpage_nounlock(struct file *filp, struct page *page) 192 { 193 struct inode *inode = file_inode(filp); 194 struct ceph_inode_info *ci = ceph_inode(inode); 195 struct ceph_osd_client *osdc = 196 &ceph_inode_to_client(inode)->client->osdc; 197 int err = 0; 198 u64 off = page_offset(page); 199 u64 len = PAGE_SIZE; 200 201 if (off >= i_size_read(inode)) { 202 zero_user_segment(page, 0, PAGE_SIZE); 203 SetPageUptodate(page); 204 return 0; 205 } 206 207 if (ci->i_inline_version != CEPH_INLINE_NONE) { 208 /* 209 * Uptodate inline data should have been added 210 * into page cache while getting Fcr caps. 211 */ 212 if (off == 0) 213 return -EINVAL; 214 zero_user_segment(page, 0, PAGE_SIZE); 215 SetPageUptodate(page); 216 return 0; 217 } 218 219 err = ceph_readpage_from_fscache(inode, page); 220 if (err == 0) 221 goto out; 222 223 dout("readpage inode %p file %p page %p index %lu\n", 224 inode, filp, page, page->index); 225 err = ceph_osdc_readpages(osdc, ceph_vino(inode), &ci->i_layout, 226 off, &len, 227 ci->i_truncate_seq, ci->i_truncate_size, 228 &page, 1, 0); 229 if (err == -ENOENT) 230 err = 0; 231 if (err < 0) { 232 SetPageError(page); 233 ceph_fscache_readpage_cancel(inode, page); 234 goto out; 235 } 236 if (err < PAGE_SIZE) 237 /* zero fill remainder of page */ 238 zero_user_segment(page, err, PAGE_SIZE); 239 else 240 flush_dcache_page(page); 241 242 SetPageUptodate(page); 243 ceph_readpage_to_fscache(inode, page); 244 245 out: 246 return err < 0 ? err : 0; 247 } 248 249 static int ceph_readpage(struct file *filp, struct page *page) 250 { 251 int r = readpage_nounlock(filp, page); 252 unlock_page(page); 253 return r; 254 } 255 256 /* 257 * Finish an async read(ahead) op. 258 */ 259 static void finish_read(struct ceph_osd_request *req) 260 { 261 struct inode *inode = req->r_inode; 262 struct ceph_osd_data *osd_data; 263 int rc = req->r_result <= 0 ? req->r_result : 0; 264 int bytes = req->r_result >= 0 ? req->r_result : 0; 265 int num_pages; 266 int i; 267 268 dout("finish_read %p req %p rc %d bytes %d\n", inode, req, rc, bytes); 269 270 /* unlock all pages, zeroing any data we didn't read */ 271 osd_data = osd_req_op_extent_osd_data(req, 0); 272 BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGES); 273 num_pages = calc_pages_for((u64)osd_data->alignment, 274 (u64)osd_data->length); 275 for (i = 0; i < num_pages; i++) { 276 struct page *page = osd_data->pages[i]; 277 278 if (rc < 0 && rc != -ENOENT) { 279 ceph_fscache_readpage_cancel(inode, page); 280 goto unlock; 281 } 282 if (bytes < (int)PAGE_SIZE) { 283 /* zero (remainder of) page */ 284 int s = bytes < 0 ? 0 : bytes; 285 zero_user_segment(page, s, PAGE_SIZE); 286 } 287 dout("finish_read %p uptodate %p idx %lu\n", inode, page, 288 page->index); 289 flush_dcache_page(page); 290 SetPageUptodate(page); 291 ceph_readpage_to_fscache(inode, page); 292 unlock: 293 unlock_page(page); 294 put_page(page); 295 bytes -= PAGE_SIZE; 296 } 297 kfree(osd_data->pages); 298 } 299 300 /* 301 * start an async read(ahead) operation. return nr_pages we submitted 302 * a read for on success, or negative error code. 303 */ 304 static int start_read(struct inode *inode, struct list_head *page_list, int max) 305 { 306 struct ceph_osd_client *osdc = 307 &ceph_inode_to_client(inode)->client->osdc; 308 struct ceph_inode_info *ci = ceph_inode(inode); 309 struct page *page = list_entry(page_list->prev, struct page, lru); 310 struct ceph_vino vino; 311 struct ceph_osd_request *req; 312 u64 off; 313 u64 len; 314 int i; 315 struct page **pages; 316 pgoff_t next_index; 317 int nr_pages = 0; 318 int got = 0; 319 int ret = 0; 320 321 if (!current->journal_info) { 322 /* caller of readpages does not hold buffer and read caps 323 * (fadvise, madvise and readahead cases) */ 324 int want = CEPH_CAP_FILE_CACHE; 325 ret = ceph_try_get_caps(ci, CEPH_CAP_FILE_RD, want, &got); 326 if (ret < 0) { 327 dout("start_read %p, error getting cap\n", inode); 328 } else if (!(got & want)) { 329 dout("start_read %p, no cache cap\n", inode); 330 ret = 0; 331 } 332 if (ret <= 0) { 333 if (got) 334 ceph_put_cap_refs(ci, got); 335 while (!list_empty(page_list)) { 336 page = list_entry(page_list->prev, 337 struct page, lru); 338 list_del(&page->lru); 339 put_page(page); 340 } 341 return ret; 342 } 343 } 344 345 off = (u64) page_offset(page); 346 347 /* count pages */ 348 next_index = page->index; 349 list_for_each_entry_reverse(page, page_list, lru) { 350 if (page->index != next_index) 351 break; 352 nr_pages++; 353 next_index++; 354 if (max && nr_pages == max) 355 break; 356 } 357 len = nr_pages << PAGE_SHIFT; 358 dout("start_read %p nr_pages %d is %lld~%lld\n", inode, nr_pages, 359 off, len); 360 vino = ceph_vino(inode); 361 req = ceph_osdc_new_request(osdc, &ci->i_layout, vino, off, &len, 362 0, 1, CEPH_OSD_OP_READ, 363 CEPH_OSD_FLAG_READ, NULL, 364 ci->i_truncate_seq, ci->i_truncate_size, 365 false); 366 if (IS_ERR(req)) { 367 ret = PTR_ERR(req); 368 goto out; 369 } 370 371 /* build page vector */ 372 nr_pages = calc_pages_for(0, len); 373 pages = kmalloc(sizeof(*pages) * nr_pages, GFP_KERNEL); 374 if (!pages) { 375 ret = -ENOMEM; 376 goto out_put; 377 } 378 for (i = 0; i < nr_pages; ++i) { 379 page = list_entry(page_list->prev, struct page, lru); 380 BUG_ON(PageLocked(page)); 381 list_del(&page->lru); 382 383 dout("start_read %p adding %p idx %lu\n", inode, page, 384 page->index); 385 if (add_to_page_cache_lru(page, &inode->i_data, page->index, 386 GFP_KERNEL)) { 387 ceph_fscache_uncache_page(inode, page); 388 put_page(page); 389 dout("start_read %p add_to_page_cache failed %p\n", 390 inode, page); 391 nr_pages = i; 392 if (nr_pages > 0) { 393 len = nr_pages << PAGE_SHIFT; 394 break; 395 } 396 goto out_pages; 397 } 398 pages[i] = page; 399 } 400 osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0, false, false); 401 req->r_callback = finish_read; 402 req->r_inode = inode; 403 404 dout("start_read %p starting %p %lld~%lld\n", inode, req, off, len); 405 ret = ceph_osdc_start_request(osdc, req, false); 406 if (ret < 0) 407 goto out_pages; 408 ceph_osdc_put_request(req); 409 410 /* After adding locked pages to page cache, the inode holds cache cap. 411 * So we can drop our cap refs. */ 412 if (got) 413 ceph_put_cap_refs(ci, got); 414 415 return nr_pages; 416 417 out_pages: 418 for (i = 0; i < nr_pages; ++i) { 419 ceph_fscache_readpage_cancel(inode, pages[i]); 420 unlock_page(pages[i]); 421 } 422 ceph_put_page_vector(pages, nr_pages, false); 423 out_put: 424 ceph_osdc_put_request(req); 425 out: 426 if (got) 427 ceph_put_cap_refs(ci, got); 428 return ret; 429 } 430 431 432 /* 433 * Read multiple pages. Leave pages we don't read + unlock in page_list; 434 * the caller (VM) cleans them up. 435 */ 436 static int ceph_readpages(struct file *file, struct address_space *mapping, 437 struct list_head *page_list, unsigned nr_pages) 438 { 439 struct inode *inode = file_inode(file); 440 struct ceph_fs_client *fsc = ceph_inode_to_client(inode); 441 int rc = 0; 442 int max = 0; 443 444 if (ceph_inode(inode)->i_inline_version != CEPH_INLINE_NONE) 445 return -EINVAL; 446 447 rc = ceph_readpages_from_fscache(mapping->host, mapping, page_list, 448 &nr_pages); 449 450 if (rc == 0) 451 goto out; 452 453 if (fsc->mount_options->rsize >= PAGE_SIZE) 454 max = (fsc->mount_options->rsize + PAGE_SIZE - 1) 455 >> PAGE_SHIFT; 456 457 dout("readpages %p file %p nr_pages %d max %d\n", inode, 458 file, nr_pages, 459 max); 460 while (!list_empty(page_list)) { 461 rc = start_read(inode, page_list, max); 462 if (rc < 0) 463 goto out; 464 } 465 out: 466 ceph_fscache_readpages_cancel(inode, page_list); 467 468 dout("readpages %p file %p ret %d\n", inode, file, rc); 469 return rc; 470 } 471 472 /* 473 * Get ref for the oldest snapc for an inode with dirty data... that is, the 474 * only snap context we are allowed to write back. 475 */ 476 static struct ceph_snap_context *get_oldest_context(struct inode *inode, 477 loff_t *snap_size) 478 { 479 struct ceph_inode_info *ci = ceph_inode(inode); 480 struct ceph_snap_context *snapc = NULL; 481 struct ceph_cap_snap *capsnap = NULL; 482 483 spin_lock(&ci->i_ceph_lock); 484 list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) { 485 dout(" cap_snap %p snapc %p has %d dirty pages\n", capsnap, 486 capsnap->context, capsnap->dirty_pages); 487 if (capsnap->dirty_pages) { 488 snapc = ceph_get_snap_context(capsnap->context); 489 if (snap_size) 490 *snap_size = capsnap->size; 491 break; 492 } 493 } 494 if (!snapc && ci->i_wrbuffer_ref_head) { 495 snapc = ceph_get_snap_context(ci->i_head_snapc); 496 dout(" head snapc %p has %d dirty pages\n", 497 snapc, ci->i_wrbuffer_ref_head); 498 } 499 spin_unlock(&ci->i_ceph_lock); 500 return snapc; 501 } 502 503 /* 504 * Write a single page, but leave the page locked. 505 * 506 * If we get a write error, set the page error bit, but still adjust the 507 * dirty page accounting (i.e., page is no longer dirty). 508 */ 509 static int writepage_nounlock(struct page *page, struct writeback_control *wbc) 510 { 511 struct inode *inode; 512 struct ceph_inode_info *ci; 513 struct ceph_fs_client *fsc; 514 struct ceph_osd_client *osdc; 515 struct ceph_snap_context *snapc, *oldest; 516 loff_t page_off = page_offset(page); 517 loff_t snap_size = -1; 518 long writeback_stat; 519 u64 truncate_size; 520 u32 truncate_seq; 521 int err = 0, len = PAGE_SIZE; 522 523 dout("writepage %p idx %lu\n", page, page->index); 524 525 if (!page->mapping || !page->mapping->host) { 526 dout("writepage %p - no mapping\n", page); 527 return -EFAULT; 528 } 529 inode = page->mapping->host; 530 ci = ceph_inode(inode); 531 fsc = ceph_inode_to_client(inode); 532 osdc = &fsc->client->osdc; 533 534 /* verify this is a writeable snap context */ 535 snapc = page_snap_context(page); 536 if (snapc == NULL) { 537 dout("writepage %p page %p not dirty?\n", inode, page); 538 goto out; 539 } 540 oldest = get_oldest_context(inode, &snap_size); 541 if (snapc->seq > oldest->seq) { 542 dout("writepage %p page %p snapc %p not writeable - noop\n", 543 inode, page, snapc); 544 /* we should only noop if called by kswapd */ 545 WARN_ON((current->flags & PF_MEMALLOC) == 0); 546 ceph_put_snap_context(oldest); 547 goto out; 548 } 549 ceph_put_snap_context(oldest); 550 551 spin_lock(&ci->i_ceph_lock); 552 truncate_seq = ci->i_truncate_seq; 553 truncate_size = ci->i_truncate_size; 554 if (snap_size == -1) 555 snap_size = i_size_read(inode); 556 spin_unlock(&ci->i_ceph_lock); 557 558 /* is this a partial page at end of file? */ 559 if (page_off >= snap_size) { 560 dout("%p page eof %llu\n", page, snap_size); 561 goto out; 562 } 563 if (snap_size < page_off + len) 564 len = snap_size - page_off; 565 566 dout("writepage %p page %p index %lu on %llu~%u snapc %p\n", 567 inode, page, page->index, page_off, len, snapc); 568 569 writeback_stat = atomic_long_inc_return(&fsc->writeback_count); 570 if (writeback_stat > 571 CONGESTION_ON_THRESH(fsc->mount_options->congestion_kb)) 572 set_bdi_congested(&fsc->backing_dev_info, BLK_RW_ASYNC); 573 574 set_page_writeback(page); 575 err = ceph_osdc_writepages(osdc, ceph_vino(inode), 576 &ci->i_layout, snapc, 577 page_off, len, 578 truncate_seq, truncate_size, 579 &inode->i_mtime, &page, 1); 580 if (err < 0) { 581 struct writeback_control tmp_wbc; 582 if (!wbc) 583 wbc = &tmp_wbc; 584 if (err == -ERESTARTSYS) { 585 /* killed by SIGKILL */ 586 dout("writepage interrupted page %p\n", page); 587 redirty_page_for_writepage(wbc, page); 588 end_page_writeback(page); 589 goto out; 590 } 591 dout("writepage setting page/mapping error %d %p\n", 592 err, page); 593 SetPageError(page); 594 mapping_set_error(&inode->i_data, err); 595 wbc->pages_skipped++; 596 } else { 597 dout("writepage cleaned page %p\n", page); 598 err = 0; /* vfs expects us to return 0 */ 599 } 600 page->private = 0; 601 ClearPagePrivate(page); 602 end_page_writeback(page); 603 ceph_put_wrbuffer_cap_refs(ci, 1, snapc); 604 ceph_put_snap_context(snapc); /* page's reference */ 605 out: 606 return err; 607 } 608 609 static int ceph_writepage(struct page *page, struct writeback_control *wbc) 610 { 611 int err; 612 struct inode *inode = page->mapping->host; 613 BUG_ON(!inode); 614 ihold(inode); 615 err = writepage_nounlock(page, wbc); 616 if (err == -ERESTARTSYS) { 617 /* direct memory reclaimer was killed by SIGKILL. return 0 618 * to prevent caller from setting mapping/page error */ 619 err = 0; 620 } 621 unlock_page(page); 622 iput(inode); 623 return err; 624 } 625 626 /* 627 * lame release_pages helper. release_pages() isn't exported to 628 * modules. 629 */ 630 static void ceph_release_pages(struct page **pages, int num) 631 { 632 struct pagevec pvec; 633 int i; 634 635 pagevec_init(&pvec, 0); 636 for (i = 0; i < num; i++) { 637 if (pagevec_add(&pvec, pages[i]) == 0) 638 pagevec_release(&pvec); 639 } 640 pagevec_release(&pvec); 641 } 642 643 /* 644 * async writeback completion handler. 645 * 646 * If we get an error, set the mapping error bit, but not the individual 647 * page error bits. 648 */ 649 static void writepages_finish(struct ceph_osd_request *req) 650 { 651 struct inode *inode = req->r_inode; 652 struct ceph_inode_info *ci = ceph_inode(inode); 653 struct ceph_osd_data *osd_data; 654 struct page *page; 655 int num_pages, total_pages = 0; 656 int i, j; 657 int rc = req->r_result; 658 struct ceph_snap_context *snapc = req->r_snapc; 659 struct address_space *mapping = inode->i_mapping; 660 struct ceph_fs_client *fsc = ceph_inode_to_client(inode); 661 bool remove_page; 662 663 dout("writepages_finish %p rc %d\n", inode, rc); 664 if (rc < 0) 665 mapping_set_error(mapping, rc); 666 667 /* 668 * We lost the cache cap, need to truncate the page before 669 * it is unlocked, otherwise we'd truncate it later in the 670 * page truncation thread, possibly losing some data that 671 * raced its way in 672 */ 673 remove_page = !(ceph_caps_issued(ci) & 674 (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)); 675 676 /* clean all pages */ 677 for (i = 0; i < req->r_num_ops; i++) { 678 if (req->r_ops[i].op != CEPH_OSD_OP_WRITE) 679 break; 680 681 osd_data = osd_req_op_extent_osd_data(req, i); 682 BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGES); 683 num_pages = calc_pages_for((u64)osd_data->alignment, 684 (u64)osd_data->length); 685 total_pages += num_pages; 686 for (j = 0; j < num_pages; j++) { 687 page = osd_data->pages[j]; 688 BUG_ON(!page); 689 WARN_ON(!PageUptodate(page)); 690 691 if (atomic_long_dec_return(&fsc->writeback_count) < 692 CONGESTION_OFF_THRESH( 693 fsc->mount_options->congestion_kb)) 694 clear_bdi_congested(&fsc->backing_dev_info, 695 BLK_RW_ASYNC); 696 697 if (rc < 0) 698 SetPageError(page); 699 700 ceph_put_snap_context(page_snap_context(page)); 701 page->private = 0; 702 ClearPagePrivate(page); 703 dout("unlocking %p\n", page); 704 end_page_writeback(page); 705 706 if (remove_page) 707 generic_error_remove_page(inode->i_mapping, 708 page); 709 710 unlock_page(page); 711 } 712 dout("writepages_finish %p wrote %llu bytes cleaned %d pages\n", 713 inode, osd_data->length, rc >= 0 ? num_pages : 0); 714 715 ceph_release_pages(osd_data->pages, num_pages); 716 } 717 718 ceph_put_wrbuffer_cap_refs(ci, total_pages, snapc); 719 720 osd_data = osd_req_op_extent_osd_data(req, 0); 721 if (osd_data->pages_from_pool) 722 mempool_free(osd_data->pages, 723 ceph_sb_to_client(inode->i_sb)->wb_pagevec_pool); 724 else 725 kfree(osd_data->pages); 726 ceph_osdc_put_request(req); 727 } 728 729 /* 730 * initiate async writeback 731 */ 732 static int ceph_writepages_start(struct address_space *mapping, 733 struct writeback_control *wbc) 734 { 735 struct inode *inode = mapping->host; 736 struct ceph_inode_info *ci = ceph_inode(inode); 737 struct ceph_fs_client *fsc = ceph_inode_to_client(inode); 738 struct ceph_vino vino = ceph_vino(inode); 739 pgoff_t index, start, end; 740 int range_whole = 0; 741 int should_loop = 1; 742 pgoff_t max_pages = 0, max_pages_ever = 0; 743 struct ceph_snap_context *snapc = NULL, *last_snapc = NULL, *pgsnapc; 744 struct pagevec pvec; 745 int done = 0; 746 int rc = 0; 747 unsigned wsize = 1 << inode->i_blkbits; 748 struct ceph_osd_request *req = NULL; 749 int do_sync = 0; 750 loff_t snap_size, i_size; 751 u64 truncate_size; 752 u32 truncate_seq; 753 754 /* 755 * Include a 'sync' in the OSD request if this is a data 756 * integrity write (e.g., O_SYNC write or fsync()), or if our 757 * cap is being revoked. 758 */ 759 if ((wbc->sync_mode == WB_SYNC_ALL) || 760 ceph_caps_revoking(ci, CEPH_CAP_FILE_BUFFER)) 761 do_sync = 1; 762 dout("writepages_start %p dosync=%d (mode=%s)\n", 763 inode, do_sync, 764 wbc->sync_mode == WB_SYNC_NONE ? "NONE" : 765 (wbc->sync_mode == WB_SYNC_ALL ? "ALL" : "HOLD")); 766 767 if (ACCESS_ONCE(fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) { 768 if (ci->i_wrbuffer_ref > 0) { 769 pr_warn_ratelimited( 770 "writepage_start %p %lld forced umount\n", 771 inode, ceph_ino(inode)); 772 } 773 mapping_set_error(mapping, -EIO); 774 return -EIO; /* we're in a forced umount, don't write! */ 775 } 776 if (fsc->mount_options->wsize && fsc->mount_options->wsize < wsize) 777 wsize = fsc->mount_options->wsize; 778 if (wsize < PAGE_SIZE) 779 wsize = PAGE_SIZE; 780 max_pages_ever = wsize >> PAGE_SHIFT; 781 782 pagevec_init(&pvec, 0); 783 784 /* where to start/end? */ 785 if (wbc->range_cyclic) { 786 start = mapping->writeback_index; /* Start from prev offset */ 787 end = -1; 788 dout(" cyclic, start at %lu\n", start); 789 } else { 790 start = wbc->range_start >> PAGE_SHIFT; 791 end = wbc->range_end >> PAGE_SHIFT; 792 if (wbc->range_start == 0 && wbc->range_end == LLONG_MAX) 793 range_whole = 1; 794 should_loop = 0; 795 dout(" not cyclic, %lu to %lu\n", start, end); 796 } 797 index = start; 798 799 retry: 800 /* find oldest snap context with dirty data */ 801 ceph_put_snap_context(snapc); 802 snap_size = -1; 803 snapc = get_oldest_context(inode, &snap_size); 804 if (!snapc) { 805 /* hmm, why does writepages get called when there 806 is no dirty data? */ 807 dout(" no snap context with dirty data?\n"); 808 goto out; 809 } 810 dout(" oldest snapc is %p seq %lld (%d snaps)\n", 811 snapc, snapc->seq, snapc->num_snaps); 812 813 spin_lock(&ci->i_ceph_lock); 814 truncate_seq = ci->i_truncate_seq; 815 truncate_size = ci->i_truncate_size; 816 i_size = i_size_read(inode); 817 spin_unlock(&ci->i_ceph_lock); 818 819 if (last_snapc && snapc != last_snapc) { 820 /* if we switched to a newer snapc, restart our scan at the 821 * start of the original file range. */ 822 dout(" snapc differs from last pass, restarting at %lu\n", 823 index); 824 index = start; 825 } 826 last_snapc = snapc; 827 828 while (!done && index <= end) { 829 unsigned i; 830 int first; 831 pgoff_t strip_unit_end = 0; 832 int num_ops = 0, op_idx; 833 int pvec_pages, locked_pages = 0; 834 struct page **pages = NULL, **data_pages; 835 mempool_t *pool = NULL; /* Becomes non-null if mempool used */ 836 struct page *page; 837 int want; 838 u64 offset = 0, len = 0; 839 840 max_pages = max_pages_ever; 841 842 get_more_pages: 843 first = -1; 844 want = min(end - index, 845 min((pgoff_t)PAGEVEC_SIZE, 846 max_pages - (pgoff_t)locked_pages) - 1) 847 + 1; 848 pvec_pages = pagevec_lookup_tag(&pvec, mapping, &index, 849 PAGECACHE_TAG_DIRTY, 850 want); 851 dout("pagevec_lookup_tag got %d\n", pvec_pages); 852 if (!pvec_pages && !locked_pages) 853 break; 854 for (i = 0; i < pvec_pages && locked_pages < max_pages; i++) { 855 page = pvec.pages[i]; 856 dout("? %p idx %lu\n", page, page->index); 857 if (locked_pages == 0) 858 lock_page(page); /* first page */ 859 else if (!trylock_page(page)) 860 break; 861 862 /* only dirty pages, or our accounting breaks */ 863 if (unlikely(!PageDirty(page)) || 864 unlikely(page->mapping != mapping)) { 865 dout("!dirty or !mapping %p\n", page); 866 unlock_page(page); 867 break; 868 } 869 if (!wbc->range_cyclic && page->index > end) { 870 dout("end of range %p\n", page); 871 done = 1; 872 unlock_page(page); 873 break; 874 } 875 if (strip_unit_end && (page->index > strip_unit_end)) { 876 dout("end of strip unit %p\n", page); 877 unlock_page(page); 878 break; 879 } 880 if (wbc->sync_mode != WB_SYNC_NONE) { 881 dout("waiting on writeback %p\n", page); 882 wait_on_page_writeback(page); 883 } 884 if (page_offset(page) >= 885 (snap_size == -1 ? i_size : snap_size)) { 886 dout("%p page eof %llu\n", page, 887 (snap_size == -1 ? i_size : snap_size)); 888 done = 1; 889 unlock_page(page); 890 break; 891 } 892 if (PageWriteback(page)) { 893 dout("%p under writeback\n", page); 894 unlock_page(page); 895 break; 896 } 897 898 /* only if matching snap context */ 899 pgsnapc = page_snap_context(page); 900 if (pgsnapc->seq > snapc->seq) { 901 dout("page snapc %p %lld > oldest %p %lld\n", 902 pgsnapc, pgsnapc->seq, snapc, snapc->seq); 903 unlock_page(page); 904 if (!locked_pages) 905 continue; /* keep looking for snap */ 906 break; 907 } 908 909 if (!clear_page_dirty_for_io(page)) { 910 dout("%p !clear_page_dirty_for_io\n", page); 911 unlock_page(page); 912 break; 913 } 914 915 /* 916 * We have something to write. If this is 917 * the first locked page this time through, 918 * calculate max possinle write size and 919 * allocate a page array 920 */ 921 if (locked_pages == 0) { 922 u64 objnum; 923 u64 objoff; 924 925 /* prepare async write request */ 926 offset = (u64)page_offset(page); 927 len = wsize; 928 929 rc = ceph_calc_file_object_mapping(&ci->i_layout, 930 offset, len, 931 &objnum, &objoff, 932 &len); 933 if (rc < 0) { 934 unlock_page(page); 935 break; 936 } 937 938 num_ops = 1 + do_sync; 939 strip_unit_end = page->index + 940 ((len - 1) >> PAGE_SHIFT); 941 942 BUG_ON(pages); 943 max_pages = calc_pages_for(0, (u64)len); 944 pages = kmalloc(max_pages * sizeof (*pages), 945 GFP_NOFS); 946 if (!pages) { 947 pool = fsc->wb_pagevec_pool; 948 pages = mempool_alloc(pool, GFP_NOFS); 949 BUG_ON(!pages); 950 } 951 952 len = 0; 953 } else if (page->index != 954 (offset + len) >> PAGE_SHIFT) { 955 if (num_ops >= (pool ? CEPH_OSD_SLAB_OPS : 956 CEPH_OSD_MAX_OPS)) { 957 redirty_page_for_writepage(wbc, page); 958 unlock_page(page); 959 break; 960 } 961 962 num_ops++; 963 offset = (u64)page_offset(page); 964 len = 0; 965 } 966 967 /* note position of first page in pvec */ 968 if (first < 0) 969 first = i; 970 dout("%p will write page %p idx %lu\n", 971 inode, page, page->index); 972 973 if (atomic_long_inc_return(&fsc->writeback_count) > 974 CONGESTION_ON_THRESH( 975 fsc->mount_options->congestion_kb)) { 976 set_bdi_congested(&fsc->backing_dev_info, 977 BLK_RW_ASYNC); 978 } 979 980 pages[locked_pages] = page; 981 locked_pages++; 982 len += PAGE_SIZE; 983 } 984 985 /* did we get anything? */ 986 if (!locked_pages) 987 goto release_pvec_pages; 988 if (i) { 989 int j; 990 BUG_ON(!locked_pages || first < 0); 991 992 if (pvec_pages && i == pvec_pages && 993 locked_pages < max_pages) { 994 dout("reached end pvec, trying for more\n"); 995 pagevec_reinit(&pvec); 996 goto get_more_pages; 997 } 998 999 /* shift unused pages over in the pvec... we 1000 * will need to release them below. */ 1001 for (j = i; j < pvec_pages; j++) { 1002 dout(" pvec leftover page %p\n", pvec.pages[j]); 1003 pvec.pages[j-i+first] = pvec.pages[j]; 1004 } 1005 pvec.nr -= i-first; 1006 } 1007 1008 new_request: 1009 offset = page_offset(pages[0]); 1010 len = wsize; 1011 1012 req = ceph_osdc_new_request(&fsc->client->osdc, 1013 &ci->i_layout, vino, 1014 offset, &len, 0, num_ops, 1015 CEPH_OSD_OP_WRITE, 1016 CEPH_OSD_FLAG_WRITE | 1017 CEPH_OSD_FLAG_ONDISK, 1018 snapc, truncate_seq, 1019 truncate_size, false); 1020 if (IS_ERR(req)) { 1021 req = ceph_osdc_new_request(&fsc->client->osdc, 1022 &ci->i_layout, vino, 1023 offset, &len, 0, 1024 min(num_ops, 1025 CEPH_OSD_SLAB_OPS), 1026 CEPH_OSD_OP_WRITE, 1027 CEPH_OSD_FLAG_WRITE | 1028 CEPH_OSD_FLAG_ONDISK, 1029 snapc, truncate_seq, 1030 truncate_size, true); 1031 BUG_ON(IS_ERR(req)); 1032 } 1033 BUG_ON(len < page_offset(pages[locked_pages - 1]) + 1034 PAGE_SIZE - offset); 1035 1036 req->r_callback = writepages_finish; 1037 req->r_inode = inode; 1038 1039 /* Format the osd request message and submit the write */ 1040 len = 0; 1041 data_pages = pages; 1042 op_idx = 0; 1043 for (i = 0; i < locked_pages; i++) { 1044 u64 cur_offset = page_offset(pages[i]); 1045 if (offset + len != cur_offset) { 1046 if (op_idx + do_sync + 1 == req->r_num_ops) 1047 break; 1048 osd_req_op_extent_dup_last(req, op_idx, 1049 cur_offset - offset); 1050 dout("writepages got pages at %llu~%llu\n", 1051 offset, len); 1052 osd_req_op_extent_osd_data_pages(req, op_idx, 1053 data_pages, len, 0, 1054 !!pool, false); 1055 osd_req_op_extent_update(req, op_idx, len); 1056 1057 len = 0; 1058 offset = cur_offset; 1059 data_pages = pages + i; 1060 op_idx++; 1061 } 1062 1063 set_page_writeback(pages[i]); 1064 len += PAGE_SIZE; 1065 } 1066 1067 if (snap_size != -1) { 1068 len = min(len, snap_size - offset); 1069 } else if (i == locked_pages) { 1070 /* writepages_finish() clears writeback pages 1071 * according to the data length, so make sure 1072 * data length covers all locked pages */ 1073 u64 min_len = len + 1 - PAGE_SIZE; 1074 len = min(len, (u64)i_size_read(inode) - offset); 1075 len = max(len, min_len); 1076 } 1077 dout("writepages got pages at %llu~%llu\n", offset, len); 1078 1079 osd_req_op_extent_osd_data_pages(req, op_idx, data_pages, len, 1080 0, !!pool, false); 1081 osd_req_op_extent_update(req, op_idx, len); 1082 1083 if (do_sync) { 1084 op_idx++; 1085 osd_req_op_init(req, op_idx, CEPH_OSD_OP_STARTSYNC, 0); 1086 } 1087 BUG_ON(op_idx + 1 != req->r_num_ops); 1088 1089 pool = NULL; 1090 if (i < locked_pages) { 1091 BUG_ON(num_ops <= req->r_num_ops); 1092 num_ops -= req->r_num_ops; 1093 num_ops += do_sync; 1094 locked_pages -= i; 1095 1096 /* allocate new pages array for next request */ 1097 data_pages = pages; 1098 pages = kmalloc(locked_pages * sizeof (*pages), 1099 GFP_NOFS); 1100 if (!pages) { 1101 pool = fsc->wb_pagevec_pool; 1102 pages = mempool_alloc(pool, GFP_NOFS); 1103 BUG_ON(!pages); 1104 } 1105 memcpy(pages, data_pages + i, 1106 locked_pages * sizeof(*pages)); 1107 memset(data_pages + i, 0, 1108 locked_pages * sizeof(*pages)); 1109 } else { 1110 BUG_ON(num_ops != req->r_num_ops); 1111 index = pages[i - 1]->index + 1; 1112 /* request message now owns the pages array */ 1113 pages = NULL; 1114 } 1115 1116 req->r_mtime = inode->i_mtime; 1117 rc = ceph_osdc_start_request(&fsc->client->osdc, req, true); 1118 BUG_ON(rc); 1119 req = NULL; 1120 1121 wbc->nr_to_write -= i; 1122 if (pages) 1123 goto new_request; 1124 1125 if (wbc->nr_to_write <= 0) 1126 done = 1; 1127 1128 release_pvec_pages: 1129 dout("pagevec_release on %d pages (%p)\n", (int)pvec.nr, 1130 pvec.nr ? pvec.pages[0] : NULL); 1131 pagevec_release(&pvec); 1132 1133 if (locked_pages && !done) 1134 goto retry; 1135 } 1136 1137 if (should_loop && !done) { 1138 /* more to do; loop back to beginning of file */ 1139 dout("writepages looping back to beginning of file\n"); 1140 should_loop = 0; 1141 index = 0; 1142 goto retry; 1143 } 1144 1145 if (wbc->range_cyclic || (range_whole && wbc->nr_to_write > 0)) 1146 mapping->writeback_index = index; 1147 1148 out: 1149 ceph_osdc_put_request(req); 1150 ceph_put_snap_context(snapc); 1151 dout("writepages done, rc = %d\n", rc); 1152 return rc; 1153 } 1154 1155 1156 1157 /* 1158 * See if a given @snapc is either writeable, or already written. 1159 */ 1160 static int context_is_writeable_or_written(struct inode *inode, 1161 struct ceph_snap_context *snapc) 1162 { 1163 struct ceph_snap_context *oldest = get_oldest_context(inode, NULL); 1164 int ret = !oldest || snapc->seq <= oldest->seq; 1165 1166 ceph_put_snap_context(oldest); 1167 return ret; 1168 } 1169 1170 /* 1171 * We are only allowed to write into/dirty the page if the page is 1172 * clean, or already dirty within the same snap context. 1173 * 1174 * called with page locked. 1175 * return success with page locked, 1176 * or any failure (incl -EAGAIN) with page unlocked. 1177 */ 1178 static int ceph_update_writeable_page(struct file *file, 1179 loff_t pos, unsigned len, 1180 struct page *page) 1181 { 1182 struct inode *inode = file_inode(file); 1183 struct ceph_fs_client *fsc = ceph_inode_to_client(inode); 1184 struct ceph_inode_info *ci = ceph_inode(inode); 1185 loff_t page_off = pos & PAGE_MASK; 1186 int pos_in_page = pos & ~PAGE_MASK; 1187 int end_in_page = pos_in_page + len; 1188 loff_t i_size; 1189 int r; 1190 struct ceph_snap_context *snapc, *oldest; 1191 1192 if (ACCESS_ONCE(fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) { 1193 dout(" page %p forced umount\n", page); 1194 unlock_page(page); 1195 return -EIO; 1196 } 1197 1198 retry_locked: 1199 /* writepages currently holds page lock, but if we change that later, */ 1200 wait_on_page_writeback(page); 1201 1202 snapc = page_snap_context(page); 1203 if (snapc && snapc != ci->i_head_snapc) { 1204 /* 1205 * this page is already dirty in another (older) snap 1206 * context! is it writeable now? 1207 */ 1208 oldest = get_oldest_context(inode, NULL); 1209 1210 if (snapc->seq > oldest->seq) { 1211 ceph_put_snap_context(oldest); 1212 dout(" page %p snapc %p not current or oldest\n", 1213 page, snapc); 1214 /* 1215 * queue for writeback, and wait for snapc to 1216 * be writeable or written 1217 */ 1218 snapc = ceph_get_snap_context(snapc); 1219 unlock_page(page); 1220 ceph_queue_writeback(inode); 1221 r = wait_event_killable(ci->i_cap_wq, 1222 context_is_writeable_or_written(inode, snapc)); 1223 ceph_put_snap_context(snapc); 1224 if (r == -ERESTARTSYS) 1225 return r; 1226 return -EAGAIN; 1227 } 1228 ceph_put_snap_context(oldest); 1229 1230 /* yay, writeable, do it now (without dropping page lock) */ 1231 dout(" page %p snapc %p not current, but oldest\n", 1232 page, snapc); 1233 if (!clear_page_dirty_for_io(page)) 1234 goto retry_locked; 1235 r = writepage_nounlock(page, NULL); 1236 if (r < 0) 1237 goto fail_nosnap; 1238 goto retry_locked; 1239 } 1240 1241 if (PageUptodate(page)) { 1242 dout(" page %p already uptodate\n", page); 1243 return 0; 1244 } 1245 1246 /* full page? */ 1247 if (pos_in_page == 0 && len == PAGE_SIZE) 1248 return 0; 1249 1250 /* past end of file? */ 1251 i_size = i_size_read(inode); 1252 1253 if (page_off >= i_size || 1254 (pos_in_page == 0 && (pos+len) >= i_size && 1255 end_in_page - pos_in_page != PAGE_SIZE)) { 1256 dout(" zeroing %p 0 - %d and %d - %d\n", 1257 page, pos_in_page, end_in_page, (int)PAGE_SIZE); 1258 zero_user_segments(page, 1259 0, pos_in_page, 1260 end_in_page, PAGE_SIZE); 1261 return 0; 1262 } 1263 1264 /* we need to read it. */ 1265 r = readpage_nounlock(file, page); 1266 if (r < 0) 1267 goto fail_nosnap; 1268 goto retry_locked; 1269 fail_nosnap: 1270 unlock_page(page); 1271 return r; 1272 } 1273 1274 /* 1275 * We are only allowed to write into/dirty the page if the page is 1276 * clean, or already dirty within the same snap context. 1277 */ 1278 static int ceph_write_begin(struct file *file, struct address_space *mapping, 1279 loff_t pos, unsigned len, unsigned flags, 1280 struct page **pagep, void **fsdata) 1281 { 1282 struct inode *inode = file_inode(file); 1283 struct page *page; 1284 pgoff_t index = pos >> PAGE_SHIFT; 1285 int r; 1286 1287 do { 1288 /* get a page */ 1289 page = grab_cache_page_write_begin(mapping, index, 0); 1290 if (!page) 1291 return -ENOMEM; 1292 1293 dout("write_begin file %p inode %p page %p %d~%d\n", file, 1294 inode, page, (int)pos, (int)len); 1295 1296 r = ceph_update_writeable_page(file, pos, len, page); 1297 if (r < 0) 1298 put_page(page); 1299 else 1300 *pagep = page; 1301 } while (r == -EAGAIN); 1302 1303 return r; 1304 } 1305 1306 /* 1307 * we don't do anything in here that simple_write_end doesn't do 1308 * except adjust dirty page accounting 1309 */ 1310 static int ceph_write_end(struct file *file, struct address_space *mapping, 1311 loff_t pos, unsigned len, unsigned copied, 1312 struct page *page, void *fsdata) 1313 { 1314 struct inode *inode = file_inode(file); 1315 unsigned from = pos & (PAGE_SIZE - 1); 1316 int check_cap = 0; 1317 1318 dout("write_end file %p inode %p page %p %d~%d (%d)\n", file, 1319 inode, page, (int)pos, (int)copied, (int)len); 1320 1321 /* zero the stale part of the page if we did a short copy */ 1322 if (copied < len) 1323 zero_user_segment(page, from+copied, len); 1324 1325 /* did file size increase? */ 1326 if (pos+copied > i_size_read(inode)) 1327 check_cap = ceph_inode_set_size(inode, pos+copied); 1328 1329 if (!PageUptodate(page)) 1330 SetPageUptodate(page); 1331 1332 set_page_dirty(page); 1333 1334 unlock_page(page); 1335 put_page(page); 1336 1337 if (check_cap) 1338 ceph_check_caps(ceph_inode(inode), CHECK_CAPS_AUTHONLY, NULL); 1339 1340 return copied; 1341 } 1342 1343 /* 1344 * we set .direct_IO to indicate direct io is supported, but since we 1345 * intercept O_DIRECT reads and writes early, this function should 1346 * never get called. 1347 */ 1348 static ssize_t ceph_direct_io(struct kiocb *iocb, struct iov_iter *iter) 1349 { 1350 WARN_ON(1); 1351 return -EINVAL; 1352 } 1353 1354 const struct address_space_operations ceph_aops = { 1355 .readpage = ceph_readpage, 1356 .readpages = ceph_readpages, 1357 .writepage = ceph_writepage, 1358 .writepages = ceph_writepages_start, 1359 .write_begin = ceph_write_begin, 1360 .write_end = ceph_write_end, 1361 .set_page_dirty = ceph_set_page_dirty, 1362 .invalidatepage = ceph_invalidatepage, 1363 .releasepage = ceph_releasepage, 1364 .direct_IO = ceph_direct_io, 1365 }; 1366 1367 static void ceph_block_sigs(sigset_t *oldset) 1368 { 1369 sigset_t mask; 1370 siginitsetinv(&mask, sigmask(SIGKILL)); 1371 sigprocmask(SIG_BLOCK, &mask, oldset); 1372 } 1373 1374 static void ceph_restore_sigs(sigset_t *oldset) 1375 { 1376 sigprocmask(SIG_SETMASK, oldset, NULL); 1377 } 1378 1379 /* 1380 * vm ops 1381 */ 1382 static int ceph_filemap_fault(struct vm_area_struct *vma, struct vm_fault *vmf) 1383 { 1384 struct inode *inode = file_inode(vma->vm_file); 1385 struct ceph_inode_info *ci = ceph_inode(inode); 1386 struct ceph_file_info *fi = vma->vm_file->private_data; 1387 struct page *pinned_page = NULL; 1388 loff_t off = vmf->pgoff << PAGE_SHIFT; 1389 int want, got, ret; 1390 sigset_t oldset; 1391 1392 ceph_block_sigs(&oldset); 1393 1394 dout("filemap_fault %p %llx.%llx %llu~%zd trying to get caps\n", 1395 inode, ceph_vinop(inode), off, (size_t)PAGE_SIZE); 1396 if (fi->fmode & CEPH_FILE_MODE_LAZY) 1397 want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO; 1398 else 1399 want = CEPH_CAP_FILE_CACHE; 1400 1401 got = 0; 1402 ret = ceph_get_caps(ci, CEPH_CAP_FILE_RD, want, -1, &got, &pinned_page); 1403 if (ret < 0) 1404 goto out_restore; 1405 1406 dout("filemap_fault %p %llu~%zd got cap refs on %s\n", 1407 inode, off, (size_t)PAGE_SIZE, ceph_cap_string(got)); 1408 1409 if ((got & (CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO)) || 1410 ci->i_inline_version == CEPH_INLINE_NONE) { 1411 current->journal_info = vma->vm_file; 1412 ret = filemap_fault(vma, vmf); 1413 current->journal_info = NULL; 1414 } else 1415 ret = -EAGAIN; 1416 1417 dout("filemap_fault %p %llu~%zd dropping cap refs on %s ret %d\n", 1418 inode, off, (size_t)PAGE_SIZE, ceph_cap_string(got), ret); 1419 if (pinned_page) 1420 put_page(pinned_page); 1421 ceph_put_cap_refs(ci, got); 1422 1423 if (ret != -EAGAIN) 1424 goto out_restore; 1425 1426 /* read inline data */ 1427 if (off >= PAGE_SIZE) { 1428 /* does not support inline data > PAGE_SIZE */ 1429 ret = VM_FAULT_SIGBUS; 1430 } else { 1431 int ret1; 1432 struct address_space *mapping = inode->i_mapping; 1433 struct page *page = find_or_create_page(mapping, 0, 1434 mapping_gfp_constraint(mapping, 1435 ~__GFP_FS)); 1436 if (!page) { 1437 ret = VM_FAULT_OOM; 1438 goto out_inline; 1439 } 1440 ret1 = __ceph_do_getattr(inode, page, 1441 CEPH_STAT_CAP_INLINE_DATA, true); 1442 if (ret1 < 0 || off >= i_size_read(inode)) { 1443 unlock_page(page); 1444 put_page(page); 1445 if (ret1 < 0) 1446 ret = ret1; 1447 else 1448 ret = VM_FAULT_SIGBUS; 1449 goto out_inline; 1450 } 1451 if (ret1 < PAGE_SIZE) 1452 zero_user_segment(page, ret1, PAGE_SIZE); 1453 else 1454 flush_dcache_page(page); 1455 SetPageUptodate(page); 1456 vmf->page = page; 1457 ret = VM_FAULT_MAJOR | VM_FAULT_LOCKED; 1458 out_inline: 1459 dout("filemap_fault %p %llu~%zd read inline data ret %d\n", 1460 inode, off, (size_t)PAGE_SIZE, ret); 1461 } 1462 out_restore: 1463 ceph_restore_sigs(&oldset); 1464 if (ret < 0) 1465 ret = (ret == -ENOMEM) ? VM_FAULT_OOM : VM_FAULT_SIGBUS; 1466 1467 return ret; 1468 } 1469 1470 /* 1471 * Reuse write_begin here for simplicity. 1472 */ 1473 static int ceph_page_mkwrite(struct vm_area_struct *vma, struct vm_fault *vmf) 1474 { 1475 struct inode *inode = file_inode(vma->vm_file); 1476 struct ceph_inode_info *ci = ceph_inode(inode); 1477 struct ceph_file_info *fi = vma->vm_file->private_data; 1478 struct ceph_cap_flush *prealloc_cf; 1479 struct page *page = vmf->page; 1480 loff_t off = page_offset(page); 1481 loff_t size = i_size_read(inode); 1482 size_t len; 1483 int want, got, ret; 1484 sigset_t oldset; 1485 1486 prealloc_cf = ceph_alloc_cap_flush(); 1487 if (!prealloc_cf) 1488 return VM_FAULT_OOM; 1489 1490 ceph_block_sigs(&oldset); 1491 1492 if (ci->i_inline_version != CEPH_INLINE_NONE) { 1493 struct page *locked_page = NULL; 1494 if (off == 0) { 1495 lock_page(page); 1496 locked_page = page; 1497 } 1498 ret = ceph_uninline_data(vma->vm_file, locked_page); 1499 if (locked_page) 1500 unlock_page(locked_page); 1501 if (ret < 0) 1502 goto out_free; 1503 } 1504 1505 if (off + PAGE_SIZE <= size) 1506 len = PAGE_SIZE; 1507 else 1508 len = size & ~PAGE_MASK; 1509 1510 dout("page_mkwrite %p %llx.%llx %llu~%zd getting caps i_size %llu\n", 1511 inode, ceph_vinop(inode), off, len, size); 1512 if (fi->fmode & CEPH_FILE_MODE_LAZY) 1513 want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO; 1514 else 1515 want = CEPH_CAP_FILE_BUFFER; 1516 1517 got = 0; 1518 ret = ceph_get_caps(ci, CEPH_CAP_FILE_WR, want, off + len, 1519 &got, NULL); 1520 if (ret < 0) 1521 goto out_free; 1522 1523 dout("page_mkwrite %p %llu~%zd got cap refs on %s\n", 1524 inode, off, len, ceph_cap_string(got)); 1525 1526 /* Update time before taking page lock */ 1527 file_update_time(vma->vm_file); 1528 1529 do { 1530 lock_page(page); 1531 1532 if ((off > size) || (page->mapping != inode->i_mapping)) { 1533 unlock_page(page); 1534 ret = VM_FAULT_NOPAGE; 1535 break; 1536 } 1537 1538 ret = ceph_update_writeable_page(vma->vm_file, off, len, page); 1539 if (ret >= 0) { 1540 /* success. we'll keep the page locked. */ 1541 set_page_dirty(page); 1542 ret = VM_FAULT_LOCKED; 1543 } 1544 } while (ret == -EAGAIN); 1545 1546 if (ret == VM_FAULT_LOCKED || 1547 ci->i_inline_version != CEPH_INLINE_NONE) { 1548 int dirty; 1549 spin_lock(&ci->i_ceph_lock); 1550 ci->i_inline_version = CEPH_INLINE_NONE; 1551 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR, 1552 &prealloc_cf); 1553 spin_unlock(&ci->i_ceph_lock); 1554 if (dirty) 1555 __mark_inode_dirty(inode, dirty); 1556 } 1557 1558 dout("page_mkwrite %p %llu~%zd dropping cap refs on %s ret %d\n", 1559 inode, off, len, ceph_cap_string(got), ret); 1560 ceph_put_cap_refs(ci, got); 1561 out_free: 1562 ceph_restore_sigs(&oldset); 1563 ceph_free_cap_flush(prealloc_cf); 1564 if (ret < 0) 1565 ret = (ret == -ENOMEM) ? VM_FAULT_OOM : VM_FAULT_SIGBUS; 1566 return ret; 1567 } 1568 1569 void ceph_fill_inline_data(struct inode *inode, struct page *locked_page, 1570 char *data, size_t len) 1571 { 1572 struct address_space *mapping = inode->i_mapping; 1573 struct page *page; 1574 1575 if (locked_page) { 1576 page = locked_page; 1577 } else { 1578 if (i_size_read(inode) == 0) 1579 return; 1580 page = find_or_create_page(mapping, 0, 1581 mapping_gfp_constraint(mapping, 1582 ~__GFP_FS)); 1583 if (!page) 1584 return; 1585 if (PageUptodate(page)) { 1586 unlock_page(page); 1587 put_page(page); 1588 return; 1589 } 1590 } 1591 1592 dout("fill_inline_data %p %llx.%llx len %zu locked_page %p\n", 1593 inode, ceph_vinop(inode), len, locked_page); 1594 1595 if (len > 0) { 1596 void *kaddr = kmap_atomic(page); 1597 memcpy(kaddr, data, len); 1598 kunmap_atomic(kaddr); 1599 } 1600 1601 if (page != locked_page) { 1602 if (len < PAGE_SIZE) 1603 zero_user_segment(page, len, PAGE_SIZE); 1604 else 1605 flush_dcache_page(page); 1606 1607 SetPageUptodate(page); 1608 unlock_page(page); 1609 put_page(page); 1610 } 1611 } 1612 1613 int ceph_uninline_data(struct file *filp, struct page *locked_page) 1614 { 1615 struct inode *inode = file_inode(filp); 1616 struct ceph_inode_info *ci = ceph_inode(inode); 1617 struct ceph_fs_client *fsc = ceph_inode_to_client(inode); 1618 struct ceph_osd_request *req; 1619 struct page *page = NULL; 1620 u64 len, inline_version; 1621 int err = 0; 1622 bool from_pagecache = false; 1623 1624 spin_lock(&ci->i_ceph_lock); 1625 inline_version = ci->i_inline_version; 1626 spin_unlock(&ci->i_ceph_lock); 1627 1628 dout("uninline_data %p %llx.%llx inline_version %llu\n", 1629 inode, ceph_vinop(inode), inline_version); 1630 1631 if (inline_version == 1 || /* initial version, no data */ 1632 inline_version == CEPH_INLINE_NONE) 1633 goto out; 1634 1635 if (locked_page) { 1636 page = locked_page; 1637 WARN_ON(!PageUptodate(page)); 1638 } else if (ceph_caps_issued(ci) & 1639 (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) { 1640 page = find_get_page(inode->i_mapping, 0); 1641 if (page) { 1642 if (PageUptodate(page)) { 1643 from_pagecache = true; 1644 lock_page(page); 1645 } else { 1646 put_page(page); 1647 page = NULL; 1648 } 1649 } 1650 } 1651 1652 if (page) { 1653 len = i_size_read(inode); 1654 if (len > PAGE_SIZE) 1655 len = PAGE_SIZE; 1656 } else { 1657 page = __page_cache_alloc(GFP_NOFS); 1658 if (!page) { 1659 err = -ENOMEM; 1660 goto out; 1661 } 1662 err = __ceph_do_getattr(inode, page, 1663 CEPH_STAT_CAP_INLINE_DATA, true); 1664 if (err < 0) { 1665 /* no inline data */ 1666 if (err == -ENODATA) 1667 err = 0; 1668 goto out; 1669 } 1670 len = err; 1671 } 1672 1673 req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout, 1674 ceph_vino(inode), 0, &len, 0, 1, 1675 CEPH_OSD_OP_CREATE, 1676 CEPH_OSD_FLAG_ONDISK | CEPH_OSD_FLAG_WRITE, 1677 NULL, 0, 0, false); 1678 if (IS_ERR(req)) { 1679 err = PTR_ERR(req); 1680 goto out; 1681 } 1682 1683 req->r_mtime = inode->i_mtime; 1684 err = ceph_osdc_start_request(&fsc->client->osdc, req, false); 1685 if (!err) 1686 err = ceph_osdc_wait_request(&fsc->client->osdc, req); 1687 ceph_osdc_put_request(req); 1688 if (err < 0) 1689 goto out; 1690 1691 req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout, 1692 ceph_vino(inode), 0, &len, 1, 3, 1693 CEPH_OSD_OP_WRITE, 1694 CEPH_OSD_FLAG_ONDISK | CEPH_OSD_FLAG_WRITE, 1695 NULL, ci->i_truncate_seq, 1696 ci->i_truncate_size, false); 1697 if (IS_ERR(req)) { 1698 err = PTR_ERR(req); 1699 goto out; 1700 } 1701 1702 osd_req_op_extent_osd_data_pages(req, 1, &page, len, 0, false, false); 1703 1704 { 1705 __le64 xattr_buf = cpu_to_le64(inline_version); 1706 err = osd_req_op_xattr_init(req, 0, CEPH_OSD_OP_CMPXATTR, 1707 "inline_version", &xattr_buf, 1708 sizeof(xattr_buf), 1709 CEPH_OSD_CMPXATTR_OP_GT, 1710 CEPH_OSD_CMPXATTR_MODE_U64); 1711 if (err) 1712 goto out_put; 1713 } 1714 1715 { 1716 char xattr_buf[32]; 1717 int xattr_len = snprintf(xattr_buf, sizeof(xattr_buf), 1718 "%llu", inline_version); 1719 err = osd_req_op_xattr_init(req, 2, CEPH_OSD_OP_SETXATTR, 1720 "inline_version", 1721 xattr_buf, xattr_len, 0, 0); 1722 if (err) 1723 goto out_put; 1724 } 1725 1726 req->r_mtime = inode->i_mtime; 1727 err = ceph_osdc_start_request(&fsc->client->osdc, req, false); 1728 if (!err) 1729 err = ceph_osdc_wait_request(&fsc->client->osdc, req); 1730 out_put: 1731 ceph_osdc_put_request(req); 1732 if (err == -ECANCELED) 1733 err = 0; 1734 out: 1735 if (page && page != locked_page) { 1736 if (from_pagecache) { 1737 unlock_page(page); 1738 put_page(page); 1739 } else 1740 __free_pages(page, 0); 1741 } 1742 1743 dout("uninline_data %p %llx.%llx inline_version %llu = %d\n", 1744 inode, ceph_vinop(inode), inline_version, err); 1745 return err; 1746 } 1747 1748 static const struct vm_operations_struct ceph_vmops = { 1749 .fault = ceph_filemap_fault, 1750 .page_mkwrite = ceph_page_mkwrite, 1751 }; 1752 1753 int ceph_mmap(struct file *file, struct vm_area_struct *vma) 1754 { 1755 struct address_space *mapping = file->f_mapping; 1756 1757 if (!mapping->a_ops->readpage) 1758 return -ENOEXEC; 1759 file_accessed(file); 1760 vma->vm_ops = &ceph_vmops; 1761 return 0; 1762 } 1763 1764 enum { 1765 POOL_READ = 1, 1766 POOL_WRITE = 2, 1767 }; 1768 1769 static int __ceph_pool_perm_get(struct ceph_inode_info *ci, 1770 s64 pool, struct ceph_string *pool_ns) 1771 { 1772 struct ceph_fs_client *fsc = ceph_inode_to_client(&ci->vfs_inode); 1773 struct ceph_mds_client *mdsc = fsc->mdsc; 1774 struct ceph_osd_request *rd_req = NULL, *wr_req = NULL; 1775 struct rb_node **p, *parent; 1776 struct ceph_pool_perm *perm; 1777 struct page **pages; 1778 size_t pool_ns_len; 1779 int err = 0, err2 = 0, have = 0; 1780 1781 down_read(&mdsc->pool_perm_rwsem); 1782 p = &mdsc->pool_perm_tree.rb_node; 1783 while (*p) { 1784 perm = rb_entry(*p, struct ceph_pool_perm, node); 1785 if (pool < perm->pool) 1786 p = &(*p)->rb_left; 1787 else if (pool > perm->pool) 1788 p = &(*p)->rb_right; 1789 else { 1790 int ret = ceph_compare_string(pool_ns, 1791 perm->pool_ns, 1792 perm->pool_ns_len); 1793 if (ret < 0) 1794 p = &(*p)->rb_left; 1795 else if (ret > 0) 1796 p = &(*p)->rb_right; 1797 else { 1798 have = perm->perm; 1799 break; 1800 } 1801 } 1802 } 1803 up_read(&mdsc->pool_perm_rwsem); 1804 if (*p) 1805 goto out; 1806 1807 if (pool_ns) 1808 dout("__ceph_pool_perm_get pool %lld ns %.*s no perm cached\n", 1809 pool, (int)pool_ns->len, pool_ns->str); 1810 else 1811 dout("__ceph_pool_perm_get pool %lld no perm cached\n", pool); 1812 1813 down_write(&mdsc->pool_perm_rwsem); 1814 p = &mdsc->pool_perm_tree.rb_node; 1815 parent = NULL; 1816 while (*p) { 1817 parent = *p; 1818 perm = rb_entry(parent, struct ceph_pool_perm, node); 1819 if (pool < perm->pool) 1820 p = &(*p)->rb_left; 1821 else if (pool > perm->pool) 1822 p = &(*p)->rb_right; 1823 else { 1824 int ret = ceph_compare_string(pool_ns, 1825 perm->pool_ns, 1826 perm->pool_ns_len); 1827 if (ret < 0) 1828 p = &(*p)->rb_left; 1829 else if (ret > 0) 1830 p = &(*p)->rb_right; 1831 else { 1832 have = perm->perm; 1833 break; 1834 } 1835 } 1836 } 1837 if (*p) { 1838 up_write(&mdsc->pool_perm_rwsem); 1839 goto out; 1840 } 1841 1842 rd_req = ceph_osdc_alloc_request(&fsc->client->osdc, NULL, 1843 1, false, GFP_NOFS); 1844 if (!rd_req) { 1845 err = -ENOMEM; 1846 goto out_unlock; 1847 } 1848 1849 rd_req->r_flags = CEPH_OSD_FLAG_READ; 1850 osd_req_op_init(rd_req, 0, CEPH_OSD_OP_STAT, 0); 1851 rd_req->r_base_oloc.pool = pool; 1852 if (pool_ns) 1853 rd_req->r_base_oloc.pool_ns = ceph_get_string(pool_ns); 1854 ceph_oid_printf(&rd_req->r_base_oid, "%llx.00000000", ci->i_vino.ino); 1855 1856 err = ceph_osdc_alloc_messages(rd_req, GFP_NOFS); 1857 if (err) 1858 goto out_unlock; 1859 1860 wr_req = ceph_osdc_alloc_request(&fsc->client->osdc, NULL, 1861 1, false, GFP_NOFS); 1862 if (!wr_req) { 1863 err = -ENOMEM; 1864 goto out_unlock; 1865 } 1866 1867 wr_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ACK; 1868 osd_req_op_init(wr_req, 0, CEPH_OSD_OP_CREATE, CEPH_OSD_OP_FLAG_EXCL); 1869 ceph_oloc_copy(&wr_req->r_base_oloc, &rd_req->r_base_oloc); 1870 ceph_oid_copy(&wr_req->r_base_oid, &rd_req->r_base_oid); 1871 1872 err = ceph_osdc_alloc_messages(wr_req, GFP_NOFS); 1873 if (err) 1874 goto out_unlock; 1875 1876 /* one page should be large enough for STAT data */ 1877 pages = ceph_alloc_page_vector(1, GFP_KERNEL); 1878 if (IS_ERR(pages)) { 1879 err = PTR_ERR(pages); 1880 goto out_unlock; 1881 } 1882 1883 osd_req_op_raw_data_in_pages(rd_req, 0, pages, PAGE_SIZE, 1884 0, false, true); 1885 err = ceph_osdc_start_request(&fsc->client->osdc, rd_req, false); 1886 1887 wr_req->r_mtime = ci->vfs_inode.i_mtime; 1888 err2 = ceph_osdc_start_request(&fsc->client->osdc, wr_req, false); 1889 1890 if (!err) 1891 err = ceph_osdc_wait_request(&fsc->client->osdc, rd_req); 1892 if (!err2) 1893 err2 = ceph_osdc_wait_request(&fsc->client->osdc, wr_req); 1894 1895 if (err >= 0 || err == -ENOENT) 1896 have |= POOL_READ; 1897 else if (err != -EPERM) 1898 goto out_unlock; 1899 1900 if (err2 == 0 || err2 == -EEXIST) 1901 have |= POOL_WRITE; 1902 else if (err2 != -EPERM) { 1903 err = err2; 1904 goto out_unlock; 1905 } 1906 1907 pool_ns_len = pool_ns ? pool_ns->len : 0; 1908 perm = kmalloc(sizeof(*perm) + pool_ns_len + 1, GFP_NOFS); 1909 if (!perm) { 1910 err = -ENOMEM; 1911 goto out_unlock; 1912 } 1913 1914 perm->pool = pool; 1915 perm->perm = have; 1916 perm->pool_ns_len = pool_ns_len; 1917 if (pool_ns_len > 0) 1918 memcpy(perm->pool_ns, pool_ns->str, pool_ns_len); 1919 perm->pool_ns[pool_ns_len] = 0; 1920 1921 rb_link_node(&perm->node, parent, p); 1922 rb_insert_color(&perm->node, &mdsc->pool_perm_tree); 1923 err = 0; 1924 out_unlock: 1925 up_write(&mdsc->pool_perm_rwsem); 1926 1927 ceph_osdc_put_request(rd_req); 1928 ceph_osdc_put_request(wr_req); 1929 out: 1930 if (!err) 1931 err = have; 1932 if (pool_ns) 1933 dout("__ceph_pool_perm_get pool %lld ns %.*s result = %d\n", 1934 pool, (int)pool_ns->len, pool_ns->str, err); 1935 else 1936 dout("__ceph_pool_perm_get pool %lld result = %d\n", pool, err); 1937 return err; 1938 } 1939 1940 int ceph_pool_perm_check(struct ceph_inode_info *ci, int need) 1941 { 1942 s64 pool; 1943 struct ceph_string *pool_ns; 1944 int ret, flags; 1945 1946 if (ceph_test_mount_opt(ceph_inode_to_client(&ci->vfs_inode), 1947 NOPOOLPERM)) 1948 return 0; 1949 1950 spin_lock(&ci->i_ceph_lock); 1951 flags = ci->i_ceph_flags; 1952 pool = ci->i_layout.pool_id; 1953 spin_unlock(&ci->i_ceph_lock); 1954 check: 1955 if (flags & CEPH_I_POOL_PERM) { 1956 if ((need & CEPH_CAP_FILE_RD) && !(flags & CEPH_I_POOL_RD)) { 1957 dout("ceph_pool_perm_check pool %lld no read perm\n", 1958 pool); 1959 return -EPERM; 1960 } 1961 if ((need & CEPH_CAP_FILE_WR) && !(flags & CEPH_I_POOL_WR)) { 1962 dout("ceph_pool_perm_check pool %lld no write perm\n", 1963 pool); 1964 return -EPERM; 1965 } 1966 return 0; 1967 } 1968 1969 pool_ns = ceph_try_get_string(ci->i_layout.pool_ns); 1970 ret = __ceph_pool_perm_get(ci, pool, pool_ns); 1971 ceph_put_string(pool_ns); 1972 if (ret < 0) 1973 return ret; 1974 1975 flags = CEPH_I_POOL_PERM; 1976 if (ret & POOL_READ) 1977 flags |= CEPH_I_POOL_RD; 1978 if (ret & POOL_WRITE) 1979 flags |= CEPH_I_POOL_WR; 1980 1981 spin_lock(&ci->i_ceph_lock); 1982 if (pool == ci->i_layout.pool_id && 1983 pool_ns == rcu_dereference_raw(ci->i_layout.pool_ns)) { 1984 ci->i_ceph_flags |= flags; 1985 } else { 1986 pool = ci->i_layout.pool_id; 1987 flags = ci->i_ceph_flags; 1988 } 1989 spin_unlock(&ci->i_ceph_lock); 1990 goto check; 1991 } 1992 1993 void ceph_pool_perm_destroy(struct ceph_mds_client *mdsc) 1994 { 1995 struct ceph_pool_perm *perm; 1996 struct rb_node *n; 1997 1998 while (!RB_EMPTY_ROOT(&mdsc->pool_perm_tree)) { 1999 n = rb_first(&mdsc->pool_perm_tree); 2000 perm = rb_entry(n, struct ceph_pool_perm, node); 2001 rb_erase(n, &mdsc->pool_perm_tree); 2002 kfree(perm); 2003 } 2004 } 2005