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