fs/ceph/addr.c

   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 *folio, 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                 BUG_ON(!folio_get_private(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         BUG_ON(folio_get_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_get_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 int ceph_releasepage(struct page *page, gfp_t gfp)
 166 {
 167         struct inode *inode = page->mapping->host;
 168
 169         dout("%llx:%llx releasepage %p idx %lu (%sdirty)\n",
 170              ceph_vinop(inode), page,
 171              page->index, PageDirty(page) ? "" : "not ");
 172
 173         if (PagePrivate(page))
 174                 return 0;
 175
 176         if (PageFsCache(page)) {
 177                 if (current_is_kswapd() || !(gfp & __GFP_FS))
 178                         return 0;
 179                 wait_on_page_fscache(page);
 180         }
 181         ceph_fscache_note_page_release(inode);
 182         return 1;
 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, true);
 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
 260         __set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags);
 261         __clear_bit(NETFS_SREQ_COPY_TO_CACHE, &subreq->flags);
 262
 263         if (subreq->start >= inode->i_size)
 264                 goto out;
 265
 266         /* We need to fetch the inline data. */
 267         req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_GETATTR, USE_ANY_MDS);
 268         if (IS_ERR(req)) {
 269                 err = PTR_ERR(req);
 270                 goto out;
 271         }
 272         req->r_ino1 = ci->i_vino;
 273         req->r_args.getattr.mask = cpu_to_le32(CEPH_STAT_CAP_INLINE_DATA);
 274         req->r_num_caps = 2;
 275
 276         err = ceph_mdsc_do_request(mdsc, NULL, req);
 277         if (err < 0)
 278                 goto out;
 279
 280         rinfo = &req->r_reply_info;
 281         iinfo = &rinfo->targeti;
 282         if (iinfo->inline_version == CEPH_INLINE_NONE) {
 283                 /* The data got uninlined */
 284                 ceph_mdsc_put_request(req);
 285                 return false;
 286         }
 287
 288         len = min_t(size_t, iinfo->inline_len - subreq->start, subreq->len);
 289         iov_iter_xarray(&iter, READ, &rreq->mapping->i_pages, subreq->start, len);
 290         err = copy_to_iter(iinfo->inline_data + subreq->start, len, &iter);
 291         if (err == 0)
 292                 err = -EFAULT;
 293
 294         ceph_mdsc_put_request(req);
 295 out:
 296         netfs_subreq_terminated(subreq, err, false);
 297         return true;
 298 }
 299
 300 static void ceph_netfs_issue_read(struct netfs_io_subrequest *subreq)
 301 {
 302         struct netfs_io_request *rreq = subreq->rreq;
 303         struct inode *inode = rreq->inode;
 304         struct ceph_inode_info *ci = ceph_inode(inode);
 305         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
 306         struct ceph_osd_request *req;
 307         struct ceph_vino vino = ceph_vino(inode);
 308         struct iov_iter iter;
 309         struct page **pages;
 310         size_t page_off;
 311         int err = 0;
 312         u64 len = subreq->len;
 313
 314         if (ci->i_inline_version != CEPH_INLINE_NONE &&
 315             ceph_netfs_issue_op_inline(subreq))
 316                 return;
 317
 318         req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout, vino, subreq->start, &len,
 319                         0, 1, CEPH_OSD_OP_READ,
 320                         CEPH_OSD_FLAG_READ | fsc->client->osdc.client->options->read_from_replica,
 321                         NULL, ci->i_truncate_seq, ci->i_truncate_size, false);
 322         if (IS_ERR(req)) {
 323                 err = PTR_ERR(req);
 324                 req = NULL;
 325                 goto out;
 326         }
 327
 328         dout("%s: pos=%llu orig_len=%zu len=%llu\n", __func__, subreq->start, subreq->len, len);
 329         iov_iter_xarray(&iter, READ, &rreq->mapping->i_pages, subreq->start, len);
 330         err = iov_iter_get_pages_alloc(&iter, &pages, len, &page_off);
 331         if (err < 0) {
 332                 dout("%s: iov_ter_get_pages_alloc returned %d\n", __func__, err);
 333                 goto out;
 334         }
 335
 336         /* should always give us a page-aligned read */
 337         WARN_ON_ONCE(page_off);
 338         len = err;
 339
 340         osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0, false, false);
 341         req->r_callback = finish_netfs_read;
 342         req->r_priv = subreq;
 343         req->r_inode = inode;
 344         ihold(inode);
 345
 346         err = ceph_osdc_start_request(req->r_osdc, req, false);
 347         if (err)
 348                 iput(inode);
 349 out:
 350         ceph_osdc_put_request(req);
 351         if (err)
 352                 netfs_subreq_terminated(subreq, err, false);
 353         dout("%s: result %d\n", __func__, err);
 354 }
 355
 356 static int ceph_init_request(struct netfs_io_request *rreq, struct file *file)
 357 {
 358         struct inode *inode = rreq->inode;
 359         int got = 0, want = CEPH_CAP_FILE_CACHE;
 360         int ret = 0;
 361
 362         if (rreq->origin != NETFS_READAHEAD)
 363                 return 0;
 364
 365         if (file) {
 366                 struct ceph_rw_context *rw_ctx;
 367                 struct ceph_file_info *fi = file->private_data;
 368
 369                 rw_ctx = ceph_find_rw_context(fi);
 370                 if (rw_ctx)
 371                         return 0;
 372         }
 373
 374         /*
 375          * readahead callers do not necessarily hold Fcb caps
 376          * (e.g. fadvise, madvise).
 377          */
 378         ret = ceph_try_get_caps(inode, CEPH_CAP_FILE_RD, want, true, &got);
 379         if (ret < 0) {
 380                 dout("start_read %p, error getting cap\n", inode);
 381                 return ret;
 382         }
 383
 384         if (!(got & want)) {
 385                 dout("start_read %p, no cache cap\n", inode);
 386                 return -EACCES;
 387         }
 388         if (ret == 0)
 389                 return -EACCES;
 390
 391         rreq->netfs_priv = (void *)(uintptr_t)got;
 392         return 0;
 393 }
 394
 395 static void ceph_readahead_cleanup(struct address_space *mapping, void *priv)
 396 {
 397         struct inode *inode = mapping->host;
 398         struct ceph_inode_info *ci = ceph_inode(inode);
 399         int got = (uintptr_t)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         .begin_cache_operation  = ceph_begin_cache_operation,
 408         .issue_read             = ceph_netfs_issue_read,
 409         .expand_readahead       = ceph_netfs_expand_readahead,
 410         .clamp_length           = ceph_netfs_clamp_length,
 411         .check_write_begin      = ceph_netfs_check_write_begin,
 412         .cleanup                = ceph_readahead_cleanup,
 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                 return PTR_ERR(req);
 609
 610         set_page_writeback(page);
 611         if (caching)
 612                 ceph_set_page_fscache(page);
 613         ceph_fscache_write_to_cache(inode, page_off, len, caching);
 614
 615         /* it may be a short write due to an object boundary */
 616         WARN_ON_ONCE(len > thp_size(page));
 617         osd_req_op_extent_osd_data_pages(req, 0, &page, len, 0, false, false);
 618         dout("writepage %llu~%llu (%llu bytes)\n", page_off, len, len);
 619
 620         req->r_mtime = inode->i_mtime;
 621         err = ceph_osdc_start_request(osdc, req, true);
 622         if (!err)
 623                 err = ceph_osdc_wait_request(osdc, req);
 624
 625         ceph_update_write_metrics(&fsc->mdsc->metric, req->r_start_latency,
 626                                   req->r_end_latency, len, err);
 627
 628         ceph_osdc_put_request(req);
 629         if (err == 0)
 630                 err = len;
 631
 632         if (err < 0) {
 633                 struct writeback_control tmp_wbc;
 634                 if (!wbc)
 635                         wbc = &tmp_wbc;
 636                 if (err == -ERESTARTSYS) {
 637                         /* killed by SIGKILL */
 638                         dout("writepage interrupted page %p\n", page);
 639                         redirty_page_for_writepage(wbc, page);
 640                         end_page_writeback(page);
 641                         return err;
 642                 }
 643                 if (err == -EBLOCKLISTED)
 644                         fsc->blocklisted = true;
 645                 dout("writepage setting page/mapping error %d %p\n",
 646                      err, page);
 647                 mapping_set_error(&inode->i_data, err);
 648                 wbc->pages_skipped++;
 649         } else {
 650                 dout("writepage cleaned page %p\n", page);
 651                 err = 0;  /* vfs expects us to return 0 */
 652         }
 653         oldest = detach_page_private(page);
 654         WARN_ON_ONCE(oldest != snapc);
 655         end_page_writeback(page);
 656         ceph_put_wrbuffer_cap_refs(ci, 1, snapc);
 657         ceph_put_snap_context(snapc);  /* page's reference */
 658
 659         if (atomic_long_dec_return(&fsc->writeback_count) <
 660             CONGESTION_OFF_THRESH(fsc->mount_options->congestion_kb))
 661                 fsc->write_congested = false;
 662
 663         return err;
 664 }
 665
 666 static int ceph_writepage(struct page *page, struct writeback_control *wbc)
 667 {
 668         int err;
 669         struct inode *inode = page->mapping->host;
 670         BUG_ON(!inode);
 671         ihold(inode);
 672
 673         if (wbc->sync_mode == WB_SYNC_NONE &&
 674             ceph_inode_to_client(inode)->write_congested)
 675                 return AOP_WRITEPAGE_ACTIVATE;
 676
 677         wait_on_page_fscache(page);
 678
 679         err = writepage_nounlock(page, wbc);
 680         if (err == -ERESTARTSYS) {
 681                 /* direct memory reclaimer was killed by SIGKILL. return 0
 682                  * to prevent caller from setting mapping/page error */
 683                 err = 0;
 684         }
 685         unlock_page(page);
 686         iput(inode);
 687         return err;
 688 }
 689
 690 /*
 691  * async writeback completion handler.
 692  *
 693  * If we get an error, set the mapping error bit, but not the individual
 694  * page error bits.
 695  */
 696 static void writepages_finish(struct ceph_osd_request *req)
 697 {
 698         struct inode *inode = req->r_inode;
 699         struct ceph_inode_info *ci = ceph_inode(inode);
 700         struct ceph_osd_data *osd_data;
 701         struct page *page;
 702         int num_pages, total_pages = 0;
 703         int i, j;
 704         int rc = req->r_result;
 705         struct ceph_snap_context *snapc = req->r_snapc;
 706         struct address_space *mapping = inode->i_mapping;
 707         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
 708         unsigned int len = 0;
 709         bool remove_page;
 710
 711         dout("writepages_finish %p rc %d\n", inode, rc);
 712         if (rc < 0) {
 713                 mapping_set_error(mapping, rc);
 714                 ceph_set_error_write(ci);
 715                 if (rc == -EBLOCKLISTED)
 716                         fsc->blocklisted = true;
 717         } else {
 718                 ceph_clear_error_write(ci);
 719         }
 720
 721         /*
 722          * We lost the cache cap, need to truncate the page before
 723          * it is unlocked, otherwise we'd truncate it later in the
 724          * page truncation thread, possibly losing some data that
 725          * raced its way in
 726          */
 727         remove_page = !(ceph_caps_issued(ci) &
 728                         (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO));
 729
 730         /* clean all pages */
 731         for (i = 0; i < req->r_num_ops; i++) {
 732                 if (req->r_ops[i].op != CEPH_OSD_OP_WRITE)
 733                         break;
 734
 735                 osd_data = osd_req_op_extent_osd_data(req, i);
 736                 BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGES);
 737                 len += osd_data->length;
 738                 num_pages = calc_pages_for((u64)osd_data->alignment,
 739                                            (u64)osd_data->length);
 740                 total_pages += num_pages;
 741                 for (j = 0; j < num_pages; j++) {
 742                         page = osd_data->pages[j];
 743                         BUG_ON(!page);
 744                         WARN_ON(!PageUptodate(page));
 745
 746                         if (atomic_long_dec_return(&fsc->writeback_count) <
 747                              CONGESTION_OFF_THRESH(
 748                                         fsc->mount_options->congestion_kb))
 749                                 fsc->write_congested = false;
 750
 751                         ceph_put_snap_context(detach_page_private(page));
 752                         end_page_writeback(page);
 753                         dout("unlocking %p\n", page);
 754
 755                         if (remove_page)
 756                                 generic_error_remove_page(inode->i_mapping,
 757                                                           page);
 758
 759                         unlock_page(page);
 760                 }
 761                 dout("writepages_finish %p wrote %llu bytes cleaned %d pages\n",
 762                      inode, osd_data->length, rc >= 0 ? num_pages : 0);
 763
 764                 release_pages(osd_data->pages, num_pages);
 765         }
 766
 767         ceph_update_write_metrics(&fsc->mdsc->metric, req->r_start_latency,
 768                                   req->r_end_latency, len, rc);
 769
 770         ceph_put_wrbuffer_cap_refs(ci, total_pages, snapc);
 771
 772         osd_data = osd_req_op_extent_osd_data(req, 0);
 773         if (osd_data->pages_from_pool)
 774                 mempool_free(osd_data->pages, ceph_wb_pagevec_pool);
 775         else
 776                 kfree(osd_data->pages);
 777         ceph_osdc_put_request(req);
 778 }
 779
 780 /*
 781  * initiate async writeback
 782  */
 783 static int ceph_writepages_start(struct address_space *mapping,
 784                                  struct writeback_control *wbc)
 785 {
 786         struct inode *inode = mapping->host;
 787         struct ceph_inode_info *ci = ceph_inode(inode);
 788         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
 789         struct ceph_vino vino = ceph_vino(inode);
 790         pgoff_t index, start_index, end = -1;
 791         struct ceph_snap_context *snapc = NULL, *last_snapc = NULL, *pgsnapc;
 792         struct pagevec pvec;
 793         int rc = 0;
 794         unsigned int wsize = i_blocksize(inode);
 795         struct ceph_osd_request *req = NULL;
 796         struct ceph_writeback_ctl ceph_wbc;
 797         bool should_loop, range_whole = false;
 798         bool done = false;
 799         bool caching = ceph_is_cache_enabled(inode);
 800
 801         if (wbc->sync_mode == WB_SYNC_NONE &&
 802             fsc->write_congested)
 803                 return 0;
 804
 805         dout("writepages_start %p (mode=%s)\n", inode,
 806              wbc->sync_mode == WB_SYNC_NONE ? "NONE" :
 807              (wbc->sync_mode == WB_SYNC_ALL ? "ALL" : "HOLD"));
 808
 809         if (ceph_inode_is_shutdown(inode)) {
 810                 if (ci->i_wrbuffer_ref > 0) {
 811                         pr_warn_ratelimited(
 812                                 "writepage_start %p %lld forced umount\n",
 813                                 inode, ceph_ino(inode));
 814                 }
 815                 mapping_set_error(mapping, -EIO);
 816                 return -EIO; /* we're in a forced umount, don't write! */
 817         }
 818         if (fsc->mount_options->wsize < wsize)
 819                 wsize = fsc->mount_options->wsize;
 820
 821         pagevec_init(&pvec);
 822
 823         start_index = wbc->range_cyclic ? mapping->writeback_index : 0;
 824         index = start_index;
 825
 826 retry:
 827         /* find oldest snap context with dirty data */
 828         snapc = get_oldest_context(inode, &ceph_wbc, NULL);
 829         if (!snapc) {
 830                 /* hmm, why does writepages get called when there
 831                    is no dirty data? */
 832                 dout(" no snap context with dirty data?\n");
 833                 goto out;
 834         }
 835         dout(" oldest snapc is %p seq %lld (%d snaps)\n",
 836              snapc, snapc->seq, snapc->num_snaps);
 837
 838         should_loop = false;
 839         if (ceph_wbc.head_snapc && snapc != last_snapc) {
 840                 /* where to start/end? */
 841                 if (wbc->range_cyclic) {
 842                         index = start_index;
 843                         end = -1;
 844                         if (index > 0)
 845                                 should_loop = true;
 846                         dout(" cyclic, start at %lu\n", index);
 847                 } else {
 848                         index = wbc->range_start >> PAGE_SHIFT;
 849                         end = wbc->range_end >> PAGE_SHIFT;
 850                         if (wbc->range_start == 0 && wbc->range_end == LLONG_MAX)
 851                                 range_whole = true;
 852                         dout(" not cyclic, %lu to %lu\n", index, end);
 853                 }
 854         } else if (!ceph_wbc.head_snapc) {
 855                 /* Do not respect wbc->range_{start,end}. Dirty pages
 856                  * in that range can be associated with newer snapc.
 857                  * They are not writeable until we write all dirty pages
 858                  * associated with 'snapc' get written */
 859                 if (index > 0)
 860                         should_loop = true;
 861                 dout(" non-head snapc, range whole\n");
 862         }
 863
 864         ceph_put_snap_context(last_snapc);
 865         last_snapc = snapc;
 866
 867         while (!done && index <= end) {
 868                 int num_ops = 0, op_idx;
 869                 unsigned i, pvec_pages, max_pages, locked_pages = 0;
 870                 struct page **pages = NULL, **data_pages;
 871                 struct page *page;
 872                 pgoff_t strip_unit_end = 0;
 873                 u64 offset = 0, len = 0;
 874                 bool from_pool = false;
 875
 876                 max_pages = wsize >> PAGE_SHIFT;
 877
 878 get_more_pages:
 879                 pvec_pages = pagevec_lookup_range_tag(&pvec, mapping, &index,
 880                                                 end, PAGECACHE_TAG_DIRTY);
 881                 dout("pagevec_lookup_range_tag got %d\n", pvec_pages);
 882                 if (!pvec_pages && !locked_pages)
 883                         break;
 884                 for (i = 0; i < pvec_pages && locked_pages < max_pages; i++) {
 885                         page = pvec.pages[i];
 886                         dout("? %p idx %lu\n", page, page->index);
 887                         if (locked_pages == 0)
 888                                 lock_page(page);  /* first page */
 889                         else if (!trylock_page(page))
 890                                 break;
 891
 892                         /* only dirty pages, or our accounting breaks */
 893                         if (unlikely(!PageDirty(page)) ||
 894                             unlikely(page->mapping != mapping)) {
 895                                 dout("!dirty or !mapping %p\n", page);
 896                                 unlock_page(page);
 897                                 continue;
 898                         }
 899                         /* only if matching snap context */
 900                         pgsnapc = page_snap_context(page);
 901                         if (pgsnapc != snapc) {
 902                                 dout("page snapc %p %lld != oldest %p %lld\n",
 903                                      pgsnapc, pgsnapc->seq, snapc, snapc->seq);
 904                                 if (!should_loop &&
 905                                     !ceph_wbc.head_snapc &&
 906                                     wbc->sync_mode != WB_SYNC_NONE)
 907                                         should_loop = true;
 908                                 unlock_page(page);
 909                                 continue;
 910                         }
 911                         if (page_offset(page) >= ceph_wbc.i_size) {
 912                                 struct folio *folio = page_folio(page);
 913
 914                                 dout("folio at %lu beyond eof %llu\n",
 915                                      folio->index, ceph_wbc.i_size);
 916                                 if ((ceph_wbc.size_stable ||
 917                                     folio_pos(folio) >= i_size_read(inode)) &&
 918                                     folio_clear_dirty_for_io(folio))
 919                                         folio_invalidate(folio, 0,
 920                                                         folio_size(folio));
 921                                 folio_unlock(folio);
 922                                 continue;
 923                         }
 924                         if (strip_unit_end && (page->index > strip_unit_end)) {
 925                                 dout("end of strip unit %p\n", page);
 926                                 unlock_page(page);
 927                                 break;
 928                         }
 929                         if (PageWriteback(page) || PageFsCache(page)) {
 930                                 if (wbc->sync_mode == WB_SYNC_NONE) {
 931                                         dout("%p under writeback\n", page);
 932                                         unlock_page(page);
 933                                         continue;
 934                                 }
 935                                 dout("waiting on writeback %p\n", page);
 936                                 wait_on_page_writeback(page);
 937                                 wait_on_page_fscache(page);
 938                         }
 939
 940                         if (!clear_page_dirty_for_io(page)) {
 941                                 dout("%p !clear_page_dirty_for_io\n", page);
 942                                 unlock_page(page);
 943                                 continue;
 944                         }
 945
 946                         /*
 947                          * We have something to write.  If this is
 948                          * the first locked page this time through,
 949                          * calculate max possinle write size and
 950                          * allocate a page array
 951                          */
 952                         if (locked_pages == 0) {
 953                                 u64 objnum;
 954                                 u64 objoff;
 955                                 u32 xlen;
 956
 957                                 /* prepare async write request */
 958                                 offset = (u64)page_offset(page);
 959                                 ceph_calc_file_object_mapping(&ci->i_layout,
 960                                                               offset, wsize,
 961                                                               &objnum, &objoff,
 962                                                               &xlen);
 963                                 len = xlen;
 964
 965                                 num_ops = 1;
 966                                 strip_unit_end = page->index +
 967                                         ((len - 1) >> PAGE_SHIFT);
 968
 969                                 BUG_ON(pages);
 970                                 max_pages = calc_pages_for(0, (u64)len);
 971                                 pages = kmalloc_array(max_pages,
 972                                                       sizeof(*pages),
 973                                                       GFP_NOFS);
 974                                 if (!pages) {
 975                                         from_pool = true;
 976                                         pages = mempool_alloc(ceph_wb_pagevec_pool, GFP_NOFS);
 977                                         BUG_ON(!pages);
 978                                 }
 979
 980                                 len = 0;
 981                         } else if (page->index !=
 982                                    (offset + len) >> PAGE_SHIFT) {
 983                                 if (num_ops >= (from_pool ?  CEPH_OSD_SLAB_OPS :
 984                                                              CEPH_OSD_MAX_OPS)) {
 985                                         redirty_page_for_writepage(wbc, page);
 986                                         unlock_page(page);
 987                                         break;
 988                                 }
 989
 990                                 num_ops++;
 991                                 offset = (u64)page_offset(page);
 992                                 len = 0;
 993                         }
 994
 995                         /* note position of first page in pvec */
 996                         dout("%p will write page %p idx %lu\n",
 997                              inode, page, page->index);
 998
 999                         if (atomic_long_inc_return(&fsc->writeback_count) >
1000                             CONGESTION_ON_THRESH(
1001                                     fsc->mount_options->congestion_kb))
1002                                 fsc->write_congested = true;
1003
1004                         pages[locked_pages++] = page;
1005                         pvec.pages[i] = NULL;
1006
1007                         len += thp_size(page);
1008                 }
1009
1010                 /* did we get anything? */
1011                 if (!locked_pages)
1012                         goto release_pvec_pages;
1013                 if (i) {
1014                         unsigned j, n = 0;
1015                         /* shift unused page to beginning of pvec */
1016                         for (j = 0; j < pvec_pages; j++) {
1017                                 if (!pvec.pages[j])
1018                                         continue;
1019                                 if (n < j)
1020                                         pvec.pages[n] = pvec.pages[j];
1021                                 n++;
1022                         }
1023                         pvec.nr = n;
1024
1025                         if (pvec_pages && i == pvec_pages &&
1026                             locked_pages < max_pages) {
1027                                 dout("reached end pvec, trying for more\n");
1028                                 pagevec_release(&pvec);
1029                                 goto get_more_pages;
1030                         }
1031                 }
1032
1033 new_request:
1034                 offset = page_offset(pages[0]);
1035                 len = wsize;
1036
1037                 req = ceph_osdc_new_request(&fsc->client->osdc,
1038                                         &ci->i_layout, vino,
1039                                         offset, &len, 0, num_ops,
1040                                         CEPH_OSD_OP_WRITE, CEPH_OSD_FLAG_WRITE,
1041                                         snapc, ceph_wbc.truncate_seq,
1042                                         ceph_wbc.truncate_size, false);
1043                 if (IS_ERR(req)) {
1044                         req = ceph_osdc_new_request(&fsc->client->osdc,
1045                                                 &ci->i_layout, vino,
1046                                                 offset, &len, 0,
1047                                                 min(num_ops,
1048                                                     CEPH_OSD_SLAB_OPS),
1049                                                 CEPH_OSD_OP_WRITE,
1050                                                 CEPH_OSD_FLAG_WRITE,
1051                                                 snapc, ceph_wbc.truncate_seq,
1052                                                 ceph_wbc.truncate_size, true);
1053                         BUG_ON(IS_ERR(req));
1054                 }
1055                 BUG_ON(len < page_offset(pages[locked_pages - 1]) +
1056                              thp_size(page) - offset);
1057
1058                 req->r_callback = writepages_finish;
1059                 req->r_inode = inode;
1060
1061                 /* Format the osd request message and submit the write */
1062                 len = 0;
1063                 data_pages = pages;
1064                 op_idx = 0;
1065                 for (i = 0; i < locked_pages; i++) {
1066                         u64 cur_offset = page_offset(pages[i]);
1067                         /*
1068                          * Discontinuity in page range? Ceph can handle that by just passing
1069                          * multiple extents in the write op.
1070                          */
1071                         if (offset + len != cur_offset) {
1072                                 /* If it's full, stop here */
1073                                 if (op_idx + 1 == req->r_num_ops)
1074                                         break;
1075
1076                                 /* Kick off an fscache write with what we have so far. */
1077                                 ceph_fscache_write_to_cache(inode, offset, len, caching);
1078
1079                                 /* Start a new extent */
1080                                 osd_req_op_extent_dup_last(req, op_idx,
1081                                                            cur_offset - offset);
1082                                 dout("writepages got pages at %llu~%llu\n",
1083                                      offset, len);
1084                                 osd_req_op_extent_osd_data_pages(req, op_idx,
1085                                                         data_pages, len, 0,
1086                                                         from_pool, false);
1087                                 osd_req_op_extent_update(req, op_idx, len);
1088
1089                                 len = 0;
1090                                 offset = cur_offset;
1091                                 data_pages = pages + i;
1092                                 op_idx++;
1093                         }
1094
1095                         set_page_writeback(pages[i]);
1096                         if (caching)
1097                                 ceph_set_page_fscache(pages[i]);
1098                         len += thp_size(page);
1099                 }
1100                 ceph_fscache_write_to_cache(inode, offset, len, caching);
1101
1102                 if (ceph_wbc.size_stable) {
1103                         len = min(len, ceph_wbc.i_size - offset);
1104                 } else if (i == locked_pages) {
1105                         /* writepages_finish() clears writeback pages
1106                          * according to the data length, so make sure
1107                          * data length covers all locked pages */
1108                         u64 min_len = len + 1 - thp_size(page);
1109                         len = get_writepages_data_length(inode, pages[i - 1],
1110                                                          offset);
1111                         len = max(len, min_len);
1112                 }
1113                 dout("writepages got pages at %llu~%llu\n", offset, len);
1114
1115                 osd_req_op_extent_osd_data_pages(req, op_idx, data_pages, len,
1116                                                  0, from_pool, false);
1117                 osd_req_op_extent_update(req, op_idx, len);
1118
1119                 BUG_ON(op_idx + 1 != req->r_num_ops);
1120
1121                 from_pool = false;
1122                 if (i < locked_pages) {
1123                         BUG_ON(num_ops <= req->r_num_ops);
1124                         num_ops -= req->r_num_ops;
1125                         locked_pages -= i;
1126
1127                         /* allocate new pages array for next request */
1128                         data_pages = pages;
1129                         pages = kmalloc_array(locked_pages, sizeof(*pages),
1130                                               GFP_NOFS);
1131                         if (!pages) {
1132                                 from_pool = true;
1133                                 pages = mempool_alloc(ceph_wb_pagevec_pool, GFP_NOFS);
1134                                 BUG_ON(!pages);
1135                         }
1136                         memcpy(pages, data_pages + i,
1137                                locked_pages * sizeof(*pages));
1138                         memset(data_pages + i, 0,
1139                                locked_pages * sizeof(*pages));
1140                 } else {
1141                         BUG_ON(num_ops != req->r_num_ops);
1142                         index = pages[i - 1]->index + 1;
1143                         /* request message now owns the pages array */
1144                         pages = NULL;
1145                 }
1146
1147                 req->r_mtime = inode->i_mtime;
1148                 rc = ceph_osdc_start_request(&fsc->client->osdc, req, true);
1149                 BUG_ON(rc);
1150                 req = NULL;
1151
1152                 wbc->nr_to_write -= i;
1153                 if (pages)
1154                         goto new_request;
1155
1156                 /*
1157                  * We stop writing back only if we are not doing
1158                  * integrity sync. In case of integrity sync we have to
1159                  * keep going until we have written all the pages
1160                  * we tagged for writeback prior to entering this loop.
1161                  */
1162                 if (wbc->nr_to_write <= 0 && wbc->sync_mode == WB_SYNC_NONE)
1163                         done = true;
1164
1165 release_pvec_pages:
1166                 dout("pagevec_release on %d pages (%p)\n", (int)pvec.nr,
1167                      pvec.nr ? pvec.pages[0] : NULL);
1168                 pagevec_release(&pvec);
1169         }
1170
1171         if (should_loop && !done) {
1172                 /* more to do; loop back to beginning of file */
1173                 dout("writepages looping back to beginning of file\n");
1174                 end = start_index - 1; /* OK even when start_index == 0 */
1175
1176                 /* to write dirty pages associated with next snapc,
1177                  * we need to wait until current writes complete */
1178                 if (wbc->sync_mode != WB_SYNC_NONE &&
1179                     start_index == 0 && /* all dirty pages were checked */
1180                     !ceph_wbc.head_snapc) {
1181                         struct page *page;
1182                         unsigned i, nr;
1183                         index = 0;
1184                         while ((index <= end) &&
1185                                (nr = pagevec_lookup_tag(&pvec, mapping, &index,
1186                                                 PAGECACHE_TAG_WRITEBACK))) {
1187                                 for (i = 0; i < nr; i++) {
1188                                         page = pvec.pages[i];
1189                                         if (page_snap_context(page) != snapc)
1190                                                 continue;
1191                                         wait_on_page_writeback(page);
1192                                 }
1193                                 pagevec_release(&pvec);
1194                                 cond_resched();
1195                         }
1196                 }
1197
1198                 start_index = 0;
1199                 index = 0;
1200                 goto retry;
1201         }
1202
1203         if (wbc->range_cyclic || (range_whole && wbc->nr_to_write > 0))
1204                 mapping->writeback_index = index;
1205
1206 out:
1207         ceph_osdc_put_request(req);
1208         ceph_put_snap_context(last_snapc);
1209         dout("writepages dend - startone, rc = %d\n", rc);
1210         return rc;
1211 }
1212
1213
1214
1215 /*
1216  * See if a given @snapc is either writeable, or already written.
1217  */
1218 static int context_is_writeable_or_written(struct inode *inode,
1219                                            struct ceph_snap_context *snapc)
1220 {
1221         struct ceph_snap_context *oldest = get_oldest_context(inode, NULL, NULL);
1222         int ret = !oldest || snapc->seq <= oldest->seq;
1223
1224         ceph_put_snap_context(oldest);
1225         return ret;
1226 }
1227
1228 /**
1229  * ceph_find_incompatible - find an incompatible context and return it
1230  * @page: page being dirtied
1231  *
1232  * We are only allowed to write into/dirty a page if the page is
1233  * clean, or already dirty within the same snap context. Returns a
1234  * conflicting context if there is one, NULL if there isn't, or a
1235  * negative error code on other errors.
1236  *
1237  * Must be called with page lock held.
1238  */
1239 static struct ceph_snap_context *
1240 ceph_find_incompatible(struct page *page)
1241 {
1242         struct inode *inode = page->mapping->host;
1243         struct ceph_inode_info *ci = ceph_inode(inode);
1244
1245         if (ceph_inode_is_shutdown(inode)) {
1246                 dout(" page %p %llx:%llx is shutdown\n", page,
1247                      ceph_vinop(inode));
1248                 return ERR_PTR(-ESTALE);
1249         }
1250
1251         for (;;) {
1252                 struct ceph_snap_context *snapc, *oldest;
1253
1254                 wait_on_page_writeback(page);
1255
1256                 snapc = page_snap_context(page);
1257                 if (!snapc || snapc == ci->i_head_snapc)
1258                         break;
1259
1260                 /*
1261                  * this page is already dirty in another (older) snap
1262                  * context!  is it writeable now?
1263                  */
1264                 oldest = get_oldest_context(inode, NULL, NULL);
1265                 if (snapc->seq > oldest->seq) {
1266                         /* not writeable -- return it for the caller to deal with */
1267                         ceph_put_snap_context(oldest);
1268                         dout(" page %p snapc %p not current or oldest\n", page, snapc);
1269                         return ceph_get_snap_context(snapc);
1270                 }
1271                 ceph_put_snap_context(oldest);
1272
1273                 /* yay, writeable, do it now (without dropping page lock) */
1274                 dout(" page %p snapc %p not current, but oldest\n", page, snapc);
1275                 if (clear_page_dirty_for_io(page)) {
1276                         int r = writepage_nounlock(page, NULL);
1277                         if (r < 0)
1278                                 return ERR_PTR(r);
1279                 }
1280         }
1281         return NULL;
1282 }
1283
1284 static int ceph_netfs_check_write_begin(struct file *file, loff_t pos, unsigned int len,
1285                                         struct folio *folio, void **_fsdata)
1286 {
1287         struct inode *inode = file_inode(file);
1288         struct ceph_inode_info *ci = ceph_inode(inode);
1289         struct ceph_snap_context *snapc;
1290
1291         snapc = ceph_find_incompatible(folio_page(folio, 0));
1292         if (snapc) {
1293                 int r;
1294
1295                 folio_unlock(folio);
1296                 folio_put(folio);
1297                 if (IS_ERR(snapc))
1298                         return PTR_ERR(snapc);
1299
1300                 ceph_queue_writeback(inode);
1301                 r = wait_event_killable(ci->i_cap_wq,
1302                                         context_is_writeable_or_written(inode, snapc));
1303                 ceph_put_snap_context(snapc);
1304                 return r == 0 ? -EAGAIN : r;
1305         }
1306         return 0;
1307 }
1308
1309 /*
1310  * We are only allowed to write into/dirty the page if the page is
1311  * clean, or already dirty within the same snap context.
1312  */
1313 static int ceph_write_begin(struct file *file, struct address_space *mapping,
1314                             loff_t pos, unsigned len, unsigned aop_flags,
1315                             struct page **pagep, void **fsdata)
1316 {
1317         struct inode *inode = file_inode(file);
1318         struct folio *folio = NULL;
1319         int r;
1320
1321         r = netfs_write_begin(file, inode->i_mapping, pos, len, 0, &folio, NULL);
1322         if (r == 0)
1323                 folio_wait_fscache(folio);
1324         if (r < 0) {
1325                 if (folio)
1326                         folio_put(folio);
1327         } else {
1328                 WARN_ON_ONCE(!folio_test_locked(folio));
1329                 *pagep = &folio->page;
1330         }
1331         return r;
1332 }
1333
1334 /*
1335  * we don't do anything in here that simple_write_end doesn't do
1336  * except adjust dirty page accounting
1337  */
1338 static int ceph_write_end(struct file *file, struct address_space *mapping,
1339                           loff_t pos, unsigned len, unsigned copied,
1340                           struct page *subpage, void *fsdata)
1341 {
1342         struct folio *folio = page_folio(subpage);
1343         struct inode *inode = file_inode(file);
1344         bool check_cap = false;
1345
1346         dout("write_end file %p inode %p folio %p %d~%d (%d)\n", file,
1347              inode, folio, (int)pos, (int)copied, (int)len);
1348
1349         if (!folio_test_uptodate(folio)) {
1350                 /* just return that nothing was copied on a short copy */
1351                 if (copied < len) {
1352                         copied = 0;
1353                         goto out;
1354                 }
1355                 folio_mark_uptodate(folio);
1356         }
1357
1358         /* did file size increase? */
1359         if (pos+copied > i_size_read(inode))
1360                 check_cap = ceph_inode_set_size(inode, pos+copied);
1361
1362         folio_mark_dirty(folio);
1363
1364 out:
1365         folio_unlock(folio);
1366         folio_put(folio);
1367
1368         if (check_cap)
1369                 ceph_check_caps(ceph_inode(inode), CHECK_CAPS_AUTHONLY, NULL);
1370
1371         return copied;
1372 }
1373
1374 const struct address_space_operations ceph_aops = {
1375         .readpage = netfs_readpage,
1376         .readahead = netfs_readahead,
1377         .writepage = ceph_writepage,
1378         .writepages = ceph_writepages_start,
1379         .write_begin = ceph_write_begin,
1380         .write_end = ceph_write_end,
1381         .dirty_folio = ceph_dirty_folio,
1382         .invalidate_folio = ceph_invalidate_folio,
1383         .releasepage = ceph_releasepage,
1384         .direct_IO = noop_direct_IO,
1385 };
1386
1387 static void ceph_block_sigs(sigset_t *oldset)
1388 {
1389         sigset_t mask;
1390         siginitsetinv(&mask, sigmask(SIGKILL));
1391         sigprocmask(SIG_BLOCK, &mask, oldset);
1392 }
1393
1394 static void ceph_restore_sigs(sigset_t *oldset)
1395 {
1396         sigprocmask(SIG_SETMASK, oldset, NULL);
1397 }
1398
1399 /*
1400  * vm ops
1401  */
1402 static vm_fault_t ceph_filemap_fault(struct vm_fault *vmf)
1403 {
1404         struct vm_area_struct *vma = vmf->vma;
1405         struct inode *inode = file_inode(vma->vm_file);
1406         struct ceph_inode_info *ci = ceph_inode(inode);
1407         struct ceph_file_info *fi = vma->vm_file->private_data;
1408         loff_t off = (loff_t)vmf->pgoff << PAGE_SHIFT;
1409         int want, got, err;
1410         sigset_t oldset;
1411         vm_fault_t ret = VM_FAULT_SIGBUS;
1412
1413         if (ceph_inode_is_shutdown(inode))
1414                 return ret;
1415
1416         ceph_block_sigs(&oldset);
1417
1418         dout("filemap_fault %p %llx.%llx %llu trying to get caps\n",
1419              inode, ceph_vinop(inode), off);
1420         if (fi->fmode & CEPH_FILE_MODE_LAZY)
1421                 want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
1422         else
1423                 want = CEPH_CAP_FILE_CACHE;
1424
1425         got = 0;
1426         err = ceph_get_caps(vma->vm_file, CEPH_CAP_FILE_RD, want, -1, &got);
1427         if (err < 0)
1428                 goto out_restore;
1429
1430         dout("filemap_fault %p %llu got cap refs on %s\n",
1431              inode, off, ceph_cap_string(got));
1432
1433         if ((got & (CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO)) ||
1434             ci->i_inline_version == CEPH_INLINE_NONE) {
1435                 CEPH_DEFINE_RW_CONTEXT(rw_ctx, got);
1436                 ceph_add_rw_context(fi, &rw_ctx);
1437                 ret = filemap_fault(vmf);
1438                 ceph_del_rw_context(fi, &rw_ctx);
1439                 dout("filemap_fault %p %llu drop cap refs %s ret %x\n",
1440                      inode, off, ceph_cap_string(got), ret);
1441         } else
1442                 err = -EAGAIN;
1443
1444         ceph_put_cap_refs(ci, got);
1445
1446         if (err != -EAGAIN)
1447                 goto out_restore;
1448
1449         /* read inline data */
1450         if (off >= PAGE_SIZE) {
1451                 /* does not support inline data > PAGE_SIZE */
1452                 ret = VM_FAULT_SIGBUS;
1453         } else {
1454                 struct address_space *mapping = inode->i_mapping;
1455                 struct page *page;
1456
1457                 filemap_invalidate_lock_shared(mapping);
1458                 page = find_or_create_page(mapping, 0,
1459                                 mapping_gfp_constraint(mapping, ~__GFP_FS));
1460                 if (!page) {
1461                         ret = VM_FAULT_OOM;
1462                         goto out_inline;
1463                 }
1464                 err = __ceph_do_getattr(inode, page,
1465                                          CEPH_STAT_CAP_INLINE_DATA, true);
1466                 if (err < 0 || off >= i_size_read(inode)) {
1467                         unlock_page(page);
1468                         put_page(page);
1469                         ret = vmf_error(err);
1470                         goto out_inline;
1471                 }
1472                 if (err < PAGE_SIZE)
1473                         zero_user_segment(page, err, PAGE_SIZE);
1474                 else
1475                         flush_dcache_page(page);
1476                 SetPageUptodate(page);
1477                 vmf->page = page;
1478                 ret = VM_FAULT_MAJOR | VM_FAULT_LOCKED;
1479 out_inline:
1480                 filemap_invalidate_unlock_shared(mapping);
1481                 dout("filemap_fault %p %llu read inline data ret %x\n",
1482                      inode, off, ret);
1483         }
1484 out_restore:
1485         ceph_restore_sigs(&oldset);
1486         if (err < 0)
1487                 ret = vmf_error(err);
1488
1489         return ret;
1490 }
1491
1492 static vm_fault_t ceph_page_mkwrite(struct vm_fault *vmf)
1493 {
1494         struct vm_area_struct *vma = vmf->vma;
1495         struct inode *inode = file_inode(vma->vm_file);
1496         struct ceph_inode_info *ci = ceph_inode(inode);
1497         struct ceph_file_info *fi = vma->vm_file->private_data;
1498         struct ceph_cap_flush *prealloc_cf;
1499         struct page *page = vmf->page;
1500         loff_t off = page_offset(page);
1501         loff_t size = i_size_read(inode);
1502         size_t len;
1503         int want, got, err;
1504         sigset_t oldset;
1505         vm_fault_t ret = VM_FAULT_SIGBUS;
1506
1507         if (ceph_inode_is_shutdown(inode))
1508                 return ret;
1509
1510         prealloc_cf = ceph_alloc_cap_flush();
1511         if (!prealloc_cf)
1512                 return VM_FAULT_OOM;
1513
1514         sb_start_pagefault(inode->i_sb);
1515         ceph_block_sigs(&oldset);
1516
1517         if (off + thp_size(page) <= size)
1518                 len = thp_size(page);
1519         else
1520                 len = offset_in_thp(page, size);
1521
1522         dout("page_mkwrite %p %llx.%llx %llu~%zd getting caps i_size %llu\n",
1523              inode, ceph_vinop(inode), off, len, size);
1524         if (fi->fmode & CEPH_FILE_MODE_LAZY)
1525                 want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
1526         else
1527                 want = CEPH_CAP_FILE_BUFFER;
1528
1529         got = 0;
1530         err = ceph_get_caps(vma->vm_file, CEPH_CAP_FILE_WR, want, off + len, &got);
1531         if (err < 0)
1532                 goto out_free;
1533
1534         dout("page_mkwrite %p %llu~%zd got cap refs on %s\n",
1535              inode, off, len, ceph_cap_string(got));
1536
1537         /* Update time before taking page lock */
1538         file_update_time(vma->vm_file);
1539         inode_inc_iversion_raw(inode);
1540
1541         do {
1542                 struct ceph_snap_context *snapc;
1543
1544                 lock_page(page);
1545
1546                 if (page_mkwrite_check_truncate(page, inode) < 0) {
1547                         unlock_page(page);
1548                         ret = VM_FAULT_NOPAGE;
1549                         break;
1550                 }
1551
1552                 snapc = ceph_find_incompatible(page);
1553                 if (!snapc) {
1554                         /* success.  we'll keep the page locked. */
1555                         set_page_dirty(page);
1556                         ret = VM_FAULT_LOCKED;
1557                         break;
1558                 }
1559
1560                 unlock_page(page);
1561
1562                 if (IS_ERR(snapc)) {
1563                         ret = VM_FAULT_SIGBUS;
1564                         break;
1565                 }
1566
1567                 ceph_queue_writeback(inode);
1568                 err = wait_event_killable(ci->i_cap_wq,
1569                                 context_is_writeable_or_written(inode, snapc));
1570                 ceph_put_snap_context(snapc);
1571         } while (err == 0);
1572
1573         if (ret == VM_FAULT_LOCKED) {
1574                 int dirty;
1575                 spin_lock(&ci->i_ceph_lock);
1576                 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
1577                                                &prealloc_cf);
1578                 spin_unlock(&ci->i_ceph_lock);
1579                 if (dirty)
1580                         __mark_inode_dirty(inode, dirty);
1581         }
1582
1583         dout("page_mkwrite %p %llu~%zd dropping cap refs on %s ret %x\n",
1584              inode, off, len, ceph_cap_string(got), ret);
1585         ceph_put_cap_refs_async(ci, got);
1586 out_free:
1587         ceph_restore_sigs(&oldset);
1588         sb_end_pagefault(inode->i_sb);
1589         ceph_free_cap_flush(prealloc_cf);
1590         if (err < 0)
1591                 ret = vmf_error(err);
1592         return ret;
1593 }
1594
1595 void ceph_fill_inline_data(struct inode *inode, struct page *locked_page,
1596                            char *data, size_t len)
1597 {
1598         struct address_space *mapping = inode->i_mapping;
1599         struct page *page;
1600
1601         if (locked_page) {
1602                 page = locked_page;
1603         } else {
1604                 if (i_size_read(inode) == 0)
1605                         return;
1606                 page = find_or_create_page(mapping, 0,
1607                                            mapping_gfp_constraint(mapping,
1608                                            ~__GFP_FS));
1609                 if (!page)
1610                         return;
1611                 if (PageUptodate(page)) {
1612                         unlock_page(page);
1613                         put_page(page);
1614                         return;
1615                 }
1616         }
1617
1618         dout("fill_inline_data %p %llx.%llx len %zu locked_page %p\n",
1619              inode, ceph_vinop(inode), len, locked_page);
1620
1621         if (len > 0) {
1622                 void *kaddr = kmap_atomic(page);
1623                 memcpy(kaddr, data, len);
1624                 kunmap_atomic(kaddr);
1625         }
1626
1627         if (page != locked_page) {
1628                 if (len < PAGE_SIZE)
1629                         zero_user_segment(page, len, PAGE_SIZE);
1630                 else
1631                         flush_dcache_page(page);
1632
1633                 SetPageUptodate(page);
1634                 unlock_page(page);
1635                 put_page(page);
1636         }
1637 }
1638
1639 int ceph_uninline_data(struct file *file)
1640 {
1641         struct inode *inode = file_inode(file);
1642         struct ceph_inode_info *ci = ceph_inode(inode);
1643         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
1644         struct ceph_osd_request *req;
1645         struct ceph_cap_flush *prealloc_cf;
1646         struct folio *folio = NULL;
1647         u64 inline_version = CEPH_INLINE_NONE;
1648         struct page *pages[1];
1649         int err = 0;
1650         u64 len;
1651
1652         prealloc_cf = ceph_alloc_cap_flush();
1653         if (!prealloc_cf)
1654                 return -ENOMEM;
1655
1656         folio = read_mapping_folio(inode->i_mapping, 0, file);
1657         if (IS_ERR(folio)) {
1658                 err = PTR_ERR(folio);
1659                 goto out;
1660         }
1661
1662         folio_lock(folio);
1663
1664         spin_lock(&ci->i_ceph_lock);
1665         inline_version = ci->i_inline_version;
1666         spin_unlock(&ci->i_ceph_lock);
1667
1668         dout("uninline_data %p %llx.%llx inline_version %llu\n",
1669              inode, ceph_vinop(inode), inline_version);
1670
1671         if (inline_version == 1 || /* initial version, no data */
1672             inline_version == CEPH_INLINE_NONE)
1673                 goto out_unlock;
1674
1675         len = i_size_read(inode);
1676         if (len > folio_size(folio))
1677                 len = folio_size(folio);
1678
1679         req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
1680                                     ceph_vino(inode), 0, &len, 0, 1,
1681                                     CEPH_OSD_OP_CREATE, CEPH_OSD_FLAG_WRITE,
1682                                     NULL, 0, 0, false);
1683         if (IS_ERR(req)) {
1684                 err = PTR_ERR(req);
1685                 goto out_unlock;
1686         }
1687
1688         req->r_mtime = inode->i_mtime;
1689         err = ceph_osdc_start_request(&fsc->client->osdc, req, false);
1690         if (!err)
1691                 err = ceph_osdc_wait_request(&fsc->client->osdc, req);
1692         ceph_osdc_put_request(req);
1693         if (err < 0)
1694                 goto out_unlock;
1695
1696         req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
1697                                     ceph_vino(inode), 0, &len, 1, 3,
1698                                     CEPH_OSD_OP_WRITE, CEPH_OSD_FLAG_WRITE,
1699                                     NULL, ci->i_truncate_seq,
1700                                     ci->i_truncate_size, false);
1701         if (IS_ERR(req)) {
1702                 err = PTR_ERR(req);
1703                 goto out_unlock;
1704         }
1705
1706         pages[0] = folio_page(folio, 0);
1707         osd_req_op_extent_osd_data_pages(req, 1, pages, len, 0, false, false);
1708
1709         {
1710                 __le64 xattr_buf = cpu_to_le64(inline_version);
1711                 err = osd_req_op_xattr_init(req, 0, CEPH_OSD_OP_CMPXATTR,
1712                                             "inline_version", &xattr_buf,
1713                                             sizeof(xattr_buf),
1714                                             CEPH_OSD_CMPXATTR_OP_GT,
1715                                             CEPH_OSD_CMPXATTR_MODE_U64);
1716                 if (err)
1717                         goto out_put_req;
1718         }
1719
1720         {
1721                 char xattr_buf[32];
1722                 int xattr_len = snprintf(xattr_buf, sizeof(xattr_buf),
1723                                          "%llu", inline_version);
1724                 err = osd_req_op_xattr_init(req, 2, CEPH_OSD_OP_SETXATTR,
1725                                             "inline_version",
1726                                             xattr_buf, xattr_len, 0, 0);
1727                 if (err)
1728                         goto out_put_req;
1729         }
1730
1731         req->r_mtime = inode->i_mtime;
1732         err = ceph_osdc_start_request(&fsc->client->osdc, req, false);
1733         if (!err)
1734                 err = ceph_osdc_wait_request(&fsc->client->osdc, req);
1735
1736         ceph_update_write_metrics(&fsc->mdsc->metric, req->r_start_latency,
1737                                   req->r_end_latency, len, err);
1738
1739         if (!err) {
1740                 int dirty;
1741
1742                 /* Set to CAP_INLINE_NONE and dirty the caps */
1743                 down_read(&fsc->mdsc->snap_rwsem);
1744                 spin_lock(&ci->i_ceph_lock);
1745                 ci->i_inline_version = CEPH_INLINE_NONE;
1746                 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR, &prealloc_cf);
1747                 spin_unlock(&ci->i_ceph_lock);
1748                 up_read(&fsc->mdsc->snap_rwsem);
1749                 if (dirty)
1750                         __mark_inode_dirty(inode, dirty);
1751         }
1752 out_put_req:
1753         ceph_osdc_put_request(req);
1754         if (err == -ECANCELED)
1755                 err = 0;
1756 out_unlock:
1757         folio_unlock(folio);
1758         folio_put(folio);
1759 out:
1760         ceph_free_cap_flush(prealloc_cf);
1761         dout("uninline_data %p %llx.%llx inline_version %llu = %d\n",
1762              inode, ceph_vinop(inode), inline_version, err);
1763         return err;
1764 }
1765
1766 static const struct vm_operations_struct ceph_vmops = {
1767         .fault          = ceph_filemap_fault,
1768         .page_mkwrite   = ceph_page_mkwrite,
1769 };
1770
1771 int ceph_mmap(struct file *file, struct vm_area_struct *vma)
1772 {
1773         struct address_space *mapping = file->f_mapping;
1774
1775         if (!mapping->a_ops->readpage)
1776                 return -ENOEXEC;
1777         file_accessed(file);
1778         vma->vm_ops = &ceph_vmops;
1779         return 0;
1780 }
1781
1782 enum {
1783         POOL_READ       = 1,
1784         POOL_WRITE      = 2,
1785 };
1786
1787 static int __ceph_pool_perm_get(struct ceph_inode_info *ci,
1788                                 s64 pool, struct ceph_string *pool_ns)
1789 {
1790         struct ceph_fs_client *fsc = ceph_inode_to_client(&ci->vfs_inode);
1791         struct ceph_mds_client *mdsc = fsc->mdsc;
1792         struct ceph_osd_request *rd_req = NULL, *wr_req = NULL;
1793         struct rb_node **p, *parent;
1794         struct ceph_pool_perm *perm;
1795         struct page **pages;
1796         size_t pool_ns_len;
1797         int err = 0, err2 = 0, have = 0;
1798
1799         down_read(&mdsc->pool_perm_rwsem);
1800         p = &mdsc->pool_perm_tree.rb_node;
1801         while (*p) {
1802                 perm = rb_entry(*p, struct ceph_pool_perm, node);
1803                 if (pool < perm->pool)
1804                         p = &(*p)->rb_left;
1805                 else if (pool > perm->pool)
1806                         p = &(*p)->rb_right;
1807                 else {
1808                         int ret = ceph_compare_string(pool_ns,
1809                                                 perm->pool_ns,
1810                                                 perm->pool_ns_len);
1811                         if (ret < 0)
1812                                 p = &(*p)->rb_left;
1813                         else if (ret > 0)
1814                                 p = &(*p)->rb_right;
1815                         else {
1816                                 have = perm->perm;
1817                                 break;
1818                         }
1819                 }
1820         }
1821         up_read(&mdsc->pool_perm_rwsem);
1822         if (*p)
1823                 goto out;
1824
1825         if (pool_ns)
1826                 dout("__ceph_pool_perm_get pool %lld ns %.*s no perm cached\n",
1827                      pool, (int)pool_ns->len, pool_ns->str);
1828         else
1829                 dout("__ceph_pool_perm_get pool %lld no perm cached\n", pool);
1830
1831         down_write(&mdsc->pool_perm_rwsem);
1832         p = &mdsc->pool_perm_tree.rb_node;
1833         parent = NULL;
1834         while (*p) {
1835                 parent = *p;
1836                 perm = rb_entry(parent, struct ceph_pool_perm, node);
1837                 if (pool < perm->pool)
1838                         p = &(*p)->rb_left;
1839                 else if (pool > perm->pool)
1840                         p = &(*p)->rb_right;
1841                 else {
1842                         int ret = ceph_compare_string(pool_ns,
1843                                                 perm->pool_ns,
1844                                                 perm->pool_ns_len);
1845                         if (ret < 0)
1846                                 p = &(*p)->rb_left;
1847                         else if (ret > 0)
1848                                 p = &(*p)->rb_right;
1849                         else {
1850                                 have = perm->perm;
1851                                 break;
1852                         }
1853                 }
1854         }
1855         if (*p) {
1856                 up_write(&mdsc->pool_perm_rwsem);
1857                 goto out;
1858         }
1859
1860         rd_req = ceph_osdc_alloc_request(&fsc->client->osdc, NULL,
1861                                          1, false, GFP_NOFS);
1862         if (!rd_req) {
1863                 err = -ENOMEM;
1864                 goto out_unlock;
1865         }
1866
1867         rd_req->r_flags = CEPH_OSD_FLAG_READ;
1868         osd_req_op_init(rd_req, 0, CEPH_OSD_OP_STAT, 0);
1869         rd_req->r_base_oloc.pool = pool;
1870         if (pool_ns)
1871                 rd_req->r_base_oloc.pool_ns = ceph_get_string(pool_ns);
1872         ceph_oid_printf(&rd_req->r_base_oid, "%llx.00000000", ci->i_vino.ino);
1873
1874         err = ceph_osdc_alloc_messages(rd_req, GFP_NOFS);
1875         if (err)
1876                 goto out_unlock;
1877
1878         wr_req = ceph_osdc_alloc_request(&fsc->client->osdc, NULL,
1879                                          1, false, GFP_NOFS);
1880         if (!wr_req) {
1881                 err = -ENOMEM;
1882                 goto out_unlock;
1883         }
1884
1885         wr_req->r_flags = CEPH_OSD_FLAG_WRITE;
1886         osd_req_op_init(wr_req, 0, CEPH_OSD_OP_CREATE, CEPH_OSD_OP_FLAG_EXCL);
1887         ceph_oloc_copy(&wr_req->r_base_oloc, &rd_req->r_base_oloc);
1888         ceph_oid_copy(&wr_req->r_base_oid, &rd_req->r_base_oid);
1889
1890         err = ceph_osdc_alloc_messages(wr_req, GFP_NOFS);
1891         if (err)
1892                 goto out_unlock;
1893
1894         /* one page should be large enough for STAT data */
1895         pages = ceph_alloc_page_vector(1, GFP_KERNEL);
1896         if (IS_ERR(pages)) {
1897                 err = PTR_ERR(pages);
1898                 goto out_unlock;
1899         }
1900
1901         osd_req_op_raw_data_in_pages(rd_req, 0, pages, PAGE_SIZE,
1902                                      0, false, true);
1903         err = ceph_osdc_start_request(&fsc->client->osdc, rd_req, false);
1904
1905         wr_req->r_mtime = ci->vfs_inode.i_mtime;
1906         err2 = ceph_osdc_start_request(&fsc->client->osdc, wr_req, false);
1907
1908         if (!err)
1909                 err = ceph_osdc_wait_request(&fsc->client->osdc, rd_req);
1910         if (!err2)
1911                 err2 = ceph_osdc_wait_request(&fsc->client->osdc, wr_req);
1912
1913         if (err >= 0 || err == -ENOENT)
1914                 have |= POOL_READ;
1915         else if (err != -EPERM) {
1916                 if (err == -EBLOCKLISTED)
1917                         fsc->blocklisted = true;
1918                 goto out_unlock;
1919         }
1920
1921         if (err2 == 0 || err2 == -EEXIST)
1922                 have |= POOL_WRITE;
1923         else if (err2 != -EPERM) {
1924                 if (err2 == -EBLOCKLISTED)
1925                         fsc->blocklisted = true;
1926                 err = err2;
1927                 goto out_unlock;
1928         }
1929
1930         pool_ns_len = pool_ns ? pool_ns->len : 0;
1931         perm = kmalloc(sizeof(*perm) + pool_ns_len + 1, GFP_NOFS);
1932         if (!perm) {
1933                 err = -ENOMEM;
1934                 goto out_unlock;
1935         }
1936
1937         perm->pool = pool;
1938         perm->perm = have;
1939         perm->pool_ns_len = pool_ns_len;
1940         if (pool_ns_len > 0)
1941                 memcpy(perm->pool_ns, pool_ns->str, pool_ns_len);
1942         perm->pool_ns[pool_ns_len] = 0;
1943
1944         rb_link_node(&perm->node, parent, p);
1945         rb_insert_color(&perm->node, &mdsc->pool_perm_tree);
1946         err = 0;
1947 out_unlock:
1948         up_write(&mdsc->pool_perm_rwsem);
1949
1950         ceph_osdc_put_request(rd_req);
1951         ceph_osdc_put_request(wr_req);
1952 out:
1953         if (!err)
1954                 err = have;
1955         if (pool_ns)
1956                 dout("__ceph_pool_perm_get pool %lld ns %.*s result = %d\n",
1957                      pool, (int)pool_ns->len, pool_ns->str, err);
1958         else
1959                 dout("__ceph_pool_perm_get pool %lld result = %d\n", pool, err);
1960         return err;
1961 }
1962
1963 int ceph_pool_perm_check(struct inode *inode, int need)
1964 {
1965         struct ceph_inode_info *ci = ceph_inode(inode);
1966         struct ceph_string *pool_ns;
1967         s64 pool;
1968         int ret, flags;
1969
1970         /* Only need to do this for regular files */
1971         if (!S_ISREG(inode->i_mode))
1972                 return 0;
1973
1974         if (ci->i_vino.snap != CEPH_NOSNAP) {
1975                 /*
1976                  * Pool permission check needs to write to the first object.
1977                  * But for snapshot, head of the first object may have alread
1978                  * been deleted. Skip check to avoid creating orphan object.
1979                  */
1980                 return 0;
1981         }
1982
1983         if (ceph_test_mount_opt(ceph_inode_to_client(inode),
1984                                 NOPOOLPERM))
1985                 return 0;
1986
1987         spin_lock(&ci->i_ceph_lock);
1988         flags = ci->i_ceph_flags;
1989         pool = ci->i_layout.pool_id;
1990         spin_unlock(&ci->i_ceph_lock);
1991 check:
1992         if (flags & CEPH_I_POOL_PERM) {
1993                 if ((need & CEPH_CAP_FILE_RD) && !(flags & CEPH_I_POOL_RD)) {
1994                         dout("ceph_pool_perm_check pool %lld no read perm\n",
1995                              pool);
1996                         return -EPERM;
1997                 }
1998                 if ((need & CEPH_CAP_FILE_WR) && !(flags & CEPH_I_POOL_WR)) {
1999                         dout("ceph_pool_perm_check pool %lld no write perm\n",
2000                              pool);
2001                         return -EPERM;
2002                 }
2003                 return 0;
2004         }
2005
2006         pool_ns = ceph_try_get_string(ci->i_layout.pool_ns);
2007         ret = __ceph_pool_perm_get(ci, pool, pool_ns);
2008         ceph_put_string(pool_ns);
2009         if (ret < 0)
2010                 return ret;
2011
2012         flags = CEPH_I_POOL_PERM;
2013         if (ret & POOL_READ)
2014                 flags |= CEPH_I_POOL_RD;
2015         if (ret & POOL_WRITE)
2016                 flags |= CEPH_I_POOL_WR;
2017
2018         spin_lock(&ci->i_ceph_lock);
2019         if (pool == ci->i_layout.pool_id &&
2020             pool_ns == rcu_dereference_raw(ci->i_layout.pool_ns)) {
2021                 ci->i_ceph_flags |= flags;
2022         } else {
2023                 pool = ci->i_layout.pool_id;
2024                 flags = ci->i_ceph_flags;
2025         }
2026         spin_unlock(&ci->i_ceph_lock);
2027         goto check;
2028 }
2029
2030 void ceph_pool_perm_destroy(struct ceph_mds_client *mdsc)
2031 {
2032         struct ceph_pool_perm *perm;
2033         struct rb_node *n;
2034
2035         while (!RB_EMPTY_ROOT(&mdsc->pool_perm_tree)) {
2036                 n = rb_first(&mdsc->pool_perm_tree);
2037                 perm = rb_entry(n, struct ceph_pool_perm, node);
2038                 rb_erase(n, &mdsc->pool_perm_tree);
2039                 kfree(perm);
2040         }
2041 }