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