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#include <trace/events/netfs.h>
  17
  18#include "super.h"
  19#include "mds_client.h"
  20#include "cache.h"
  21#include "metric.h"
  22#include "crypto.h"
  23#include <linux/ceph/osd_client.h>
  24#include <linux/ceph/striper.h>
  25
  26/*
  27 * Ceph address space ops.
  28 *
  29 * There are a few funny things going on here.
  30 *
  31 * The page->private field is used to reference a struct
  32 * ceph_snap_context for _every_ dirty page.  This indicates which
  33 * snapshot the page was logically dirtied in, and thus which snap
  34 * context needs to be associated with the osd write during writeback.
  35 *
  36 * Similarly, struct ceph_inode_info maintains a set of counters to
  37 * count dirty pages on the inode.  In the absence of snapshots,
  38 * i_wrbuffer_ref == i_wrbuffer_ref_head == the dirty page count.
  39 *
  40 * When a snapshot is taken (that is, when the client receives
  41 * notification that a snapshot was taken), each inode with caps and
  42 * with dirty pages (dirty pages implies there is a cap) gets a new
  43 * ceph_cap_snap in the i_cap_snaps list (which is sorted in ascending
  44 * order, new snaps go to the tail).  The i_wrbuffer_ref_head count is
  45 * moved to capsnap->dirty. (Unless a sync write is currently in
  46 * progress.  In that case, the capsnap is said to be "pending", new
  47 * writes cannot start, and the capsnap isn't "finalized" until the
  48 * write completes (or fails) and a final size/mtime for the inode for
  49 * that snap can be settled upon.)  i_wrbuffer_ref_head is reset to 0.
  50 *
  51 * On writeback, we must submit writes to the osd IN SNAP ORDER.  So,
  52 * we look for the first capsnap in i_cap_snaps and write out pages in
  53 * that snap context _only_.  Then we move on to the next capsnap,
  54 * eventually reaching the "live" or "head" context (i.e., pages that
  55 * are not yet snapped) and are writing the most recently dirtied
  56 * pages.
  57 *
  58 * Invalidate and so forth must take care to ensure the dirty page
  59 * accounting is preserved.
  60 */
  61
  62#define CONGESTION_ON_THRESH(congestion_kb) (congestion_kb >> (PAGE_SHIFT-10))
  63#define CONGESTION_OFF_THRESH(congestion_kb)				\
  64	(CONGESTION_ON_THRESH(congestion_kb) -				\
  65	 (CONGESTION_ON_THRESH(congestion_kb) >> 2))
  66
  67static int ceph_netfs_check_write_begin(struct file *file, loff_t pos, unsigned int len,
  68					struct folio **foliop, void **_fsdata);
  69
  70static inline struct ceph_snap_context *page_snap_context(struct page *page)
  71{
  72	if (PagePrivate(page))
  73		return (void *)page->private;
  74	return NULL;
  75}
  76
  77/*
  78 * Dirty a page.  Optimistically adjust accounting, on the assumption
  79 * that we won't race with invalidate.  If we do, readjust.
  80 */
  81static bool ceph_dirty_folio(struct address_space *mapping, struct folio *folio)
  82{
  83	struct inode *inode = mapping->host;
  84	struct ceph_client *cl = ceph_inode_to_client(inode);
  85	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
  86	struct ceph_inode_info *ci;
  87	struct ceph_snap_context *snapc;
  88
  89	if (folio_test_dirty(folio)) {
  90		doutc(cl, "%llx.%llx %p idx %lu -- already dirty\n",
  91		      ceph_vinop(inode), folio, folio->index);
  92		VM_BUG_ON_FOLIO(!folio_test_private(folio), folio);
  93		return false;
  94	}
  95
  96	atomic64_inc(&mdsc->dirty_folios);
  97
  98	ci = ceph_inode(inode);
  99
 100	/* dirty the head */
 101	spin_lock(&ci->i_ceph_lock);
 102	if (__ceph_have_pending_cap_snap(ci)) {
 103		struct ceph_cap_snap *capsnap =
 104				list_last_entry(&ci->i_cap_snaps,
 105						struct ceph_cap_snap,
 106						ci_item);
 107		snapc = ceph_get_snap_context(capsnap->context);
 108		capsnap->dirty_pages++;
 109	} else {
 110		BUG_ON(!ci->i_head_snapc);
 111		snapc = ceph_get_snap_context(ci->i_head_snapc);
 112		++ci->i_wrbuffer_ref_head;
 113	}
 114	if (ci->i_wrbuffer_ref == 0)
 115		ihold(inode);
 116	++ci->i_wrbuffer_ref;
 117	doutc(cl, "%llx.%llx %p idx %lu head %d/%d -> %d/%d "
 118	      "snapc %p seq %lld (%d snaps)\n",
 119	      ceph_vinop(inode), folio, folio->index,
 120	      ci->i_wrbuffer_ref-1, ci->i_wrbuffer_ref_head-1,
 121	      ci->i_wrbuffer_ref, ci->i_wrbuffer_ref_head,
 122	      snapc, snapc->seq, snapc->num_snaps);
 123	spin_unlock(&ci->i_ceph_lock);
 124
 125	/*
 126	 * Reference snap context in folio->private.  Also set
 127	 * PagePrivate so that we get invalidate_folio callback.
 128	 */
 129	VM_WARN_ON_FOLIO(folio->private, folio);
 130	folio_attach_private(folio, snapc);
 131
 132	return ceph_fscache_dirty_folio(mapping, folio);
 133}
 134
 135/*
 136 * If we are truncating the full folio (i.e. offset == 0), adjust the
 137 * dirty folio counters appropriately.  Only called if there is private
 138 * data on the folio.
 139 */
 140static void ceph_invalidate_folio(struct folio *folio, size_t offset,
 141				size_t length)
 142{
 143	struct inode *inode = folio->mapping->host;
 144	struct ceph_client *cl = ceph_inode_to_client(inode);
 145	struct ceph_inode_info *ci = ceph_inode(inode);
 146	struct ceph_snap_context *snapc;
 147
 148
 149	if (offset != 0 || length != folio_size(folio)) {
 150		doutc(cl, "%llx.%llx idx %lu partial dirty page %zu~%zu\n",
 151		      ceph_vinop(inode), folio->index, offset, length);
 152		return;
 153	}
 154
 155	WARN_ON(!folio_test_locked(folio));
 156	if (folio_test_private(folio)) {
 157		doutc(cl, "%llx.%llx idx %lu full dirty page\n",
 158		      ceph_vinop(inode), folio->index);
 159
 160		snapc = folio_detach_private(folio);
 161		ceph_put_wrbuffer_cap_refs(ci, 1, snapc);
 162		ceph_put_snap_context(snapc);
 163	}
 164
 165	netfs_invalidate_folio(folio, offset, length);
 166}
 167
 168static void ceph_netfs_expand_readahead(struct netfs_io_request *rreq)
 169{
 170	struct inode *inode = rreq->inode;
 171	struct ceph_inode_info *ci = ceph_inode(inode);
 172	struct ceph_file_layout *lo = &ci->i_layout;
 173	unsigned long max_pages = inode->i_sb->s_bdi->ra_pages;
 174	loff_t end = rreq->start + rreq->len, new_end;
 175	struct ceph_netfs_request_data *priv = rreq->netfs_priv;
 176	unsigned long max_len;
 177	u32 blockoff;
 178
 179	if (priv) {
 180		/* Readahead is disabled by posix_fadvise POSIX_FADV_RANDOM */
 181		if (priv->file_ra_disabled)
 182			max_pages = 0;
 183		else
 184			max_pages = priv->file_ra_pages;
 185
 186	}
 187
 188	/* Readahead is disabled */
 189	if (!max_pages)
 190		return;
 191
 192	max_len = max_pages << PAGE_SHIFT;
 193
 194	/*
 195	 * Try to expand the length forward by rounding up it to the next
 196	 * block, but do not exceed the file size, unless the original
 197	 * request already exceeds it.
 198	 */
 199	new_end = umin(round_up(end, lo->stripe_unit), rreq->i_size);
 200	if (new_end > end && new_end <= rreq->start + max_len)
 201		rreq->len = new_end - rreq->start;
 202
 203	/* Try to expand the start downward */
 204	div_u64_rem(rreq->start, lo->stripe_unit, &blockoff);
 205	if (rreq->len + blockoff <= max_len) {
 206		rreq->start -= blockoff;
 207		rreq->len += blockoff;
 208	}
 209}
 210
 211static void finish_netfs_read(struct ceph_osd_request *req)
 212{
 213	struct inode *inode = req->r_inode;
 214	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
 215	struct ceph_client *cl = fsc->client;
 216	struct ceph_osd_data *osd_data = osd_req_op_extent_osd_data(req, 0);
 217	struct netfs_io_subrequest *subreq = req->r_priv;
 218	struct ceph_osd_req_op *op = &req->r_ops[0];
 219	int err = req->r_result;
 220	bool sparse = (op->op == CEPH_OSD_OP_SPARSE_READ);
 221
 222	ceph_update_read_metrics(&fsc->mdsc->metric, req->r_start_latency,
 223				 req->r_end_latency, osd_data->length, err);
 224
 225	doutc(cl, "result %d subreq->len=%zu i_size=%lld\n", req->r_result,
 226	      subreq->len, i_size_read(req->r_inode));
 227
 228	/* no object means success but no data */
 229	if (err == -ENOENT) {
 230		__set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags);
 231		__set_bit(NETFS_SREQ_MADE_PROGRESS, &subreq->flags);
 232		err = 0;
 233	} else if (err == -EBLOCKLISTED) {
 234		fsc->blocklisted = true;
 235	}
 236
 237	if (err >= 0) {
 238		if (sparse && err > 0)
 239			err = ceph_sparse_ext_map_end(op);
 240		if (err < subreq->len &&
 241		    subreq->rreq->origin != NETFS_UNBUFFERED_READ &&
 242		    subreq->rreq->origin != NETFS_DIO_READ)
 243			__set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags);
 244		if (IS_ENCRYPTED(inode) && err > 0) {
 245			err = ceph_fscrypt_decrypt_extents(inode,
 246					osd_data->pages, subreq->start,
 247					op->extent.sparse_ext,
 248					op->extent.sparse_ext_cnt);
 249			if (err > subreq->len)
 250				err = subreq->len;
 251		}
 252		if (err > 0)
 253			__set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags);
 254	}
 255
 256	if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES) {
 257		ceph_put_page_vector(osd_data->pages,
 258				     calc_pages_for(osd_data->alignment,
 259					osd_data->length), false);
 260	}
 261	if (err > 0) {
 262		subreq->transferred = err;
 263		err = 0;
 264	}
 265	subreq->error = err;
 266	trace_netfs_sreq(subreq, netfs_sreq_trace_io_progress);
 267	netfs_read_subreq_terminated(subreq);
 268	iput(req->r_inode);
 269	ceph_dec_osd_stopping_blocker(fsc->mdsc);
 270}
 271
 272static bool ceph_netfs_issue_op_inline(struct netfs_io_subrequest *subreq)
 273{
 274	struct netfs_io_request *rreq = subreq->rreq;
 275	struct inode *inode = rreq->inode;
 276	struct ceph_mds_reply_info_parsed *rinfo;
 277	struct ceph_mds_reply_info_in *iinfo;
 278	struct ceph_mds_request *req;
 279	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
 280	struct ceph_inode_info *ci = ceph_inode(inode);
 281	ssize_t err = 0;
 282	size_t len;
 283	int mode;
 284
 285	if (rreq->origin != NETFS_UNBUFFERED_READ &&
 286	    rreq->origin != NETFS_DIO_READ)
 287		__set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags);
 288	__clear_bit(NETFS_SREQ_COPY_TO_CACHE, &subreq->flags);
 289
 290	if (subreq->start >= inode->i_size)
 291		goto out;
 292
 293	/* We need to fetch the inline data. */
 294	mode = ceph_try_to_choose_auth_mds(inode, CEPH_STAT_CAP_INLINE_DATA);
 295	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_GETATTR, mode);
 296	if (IS_ERR(req)) {
 297		err = PTR_ERR(req);
 298		goto out;
 299	}
 300	req->r_ino1 = ci->i_vino;
 301	req->r_args.getattr.mask = cpu_to_le32(CEPH_STAT_CAP_INLINE_DATA);
 302	req->r_num_caps = 2;
 303
 304	trace_netfs_sreq(subreq, netfs_sreq_trace_submit);
 305	err = ceph_mdsc_do_request(mdsc, NULL, req);
 306	if (err < 0)
 307		goto out;
 308
 309	rinfo = &req->r_reply_info;
 310	iinfo = &rinfo->targeti;
 311	if (iinfo->inline_version == CEPH_INLINE_NONE) {
 312		/* The data got uninlined */
 313		ceph_mdsc_put_request(req);
 314		return false;
 315	}
 316
 317	len = min_t(size_t, iinfo->inline_len - subreq->start, subreq->len);
 318	err = copy_to_iter(iinfo->inline_data + subreq->start, len, &subreq->io_iter);
 319	if (err == 0) {
 320		err = -EFAULT;
 321	} else {
 322		subreq->transferred += err;
 323		err = 0;
 324	}
 325
 326	ceph_mdsc_put_request(req);
 327out:
 328	subreq->error = err;
 329	trace_netfs_sreq(subreq, netfs_sreq_trace_io_progress);
 330	netfs_read_subreq_terminated(subreq);
 331	return true;
 332}
 333
 334static int ceph_netfs_prepare_read(struct netfs_io_subrequest *subreq)
 335{
 336	struct netfs_io_request *rreq = subreq->rreq;
 337	struct inode *inode = rreq->inode;
 338	struct ceph_inode_info *ci = ceph_inode(inode);
 339	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
 340	u64 objno, objoff;
 341	u32 xlen;
 342
 343	/* Truncate the extent at the end of the current block */
 344	ceph_calc_file_object_mapping(&ci->i_layout, subreq->start, subreq->len,
 345				      &objno, &objoff, &xlen);
 346	rreq->io_streams[0].sreq_max_len = umin(xlen, fsc->mount_options->rsize);
 347	return 0;
 348}
 349
 350static void ceph_netfs_issue_read(struct netfs_io_subrequest *subreq)
 351{
 352	struct netfs_io_request *rreq = subreq->rreq;
 353	struct inode *inode = rreq->inode;
 354	struct ceph_inode_info *ci = ceph_inode(inode);
 355	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
 356	struct ceph_client *cl = fsc->client;
 357	struct ceph_osd_request *req = NULL;
 358	struct ceph_vino vino = ceph_vino(inode);
 359	int err;
 360	u64 len;
 361	bool sparse = IS_ENCRYPTED(inode) || ceph_test_mount_opt(fsc, SPARSEREAD);
 362	u64 off = subreq->start;
 363	int extent_cnt;
 364
 365	if (ceph_inode_is_shutdown(inode)) {
 366		err = -EIO;
 367		goto out;
 368	}
 369
 370	if (ceph_has_inline_data(ci) && ceph_netfs_issue_op_inline(subreq))
 371		return;
 372
 373	// TODO: This rounding here is slightly dodgy.  It *should* work, for
 374	// now, as the cache only deals in blocks that are a multiple of
 375	// PAGE_SIZE and fscrypt blocks are at most PAGE_SIZE.  What needs to
 376	// happen is for the fscrypt driving to be moved into netfslib and the
 377	// data in the cache also to be stored encrypted.
 378	len = subreq->len;
 379	ceph_fscrypt_adjust_off_and_len(inode, &off, &len);
 380
 381	req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout, vino,
 382			off, &len, 0, 1, sparse ? CEPH_OSD_OP_SPARSE_READ : CEPH_OSD_OP_READ,
 383			CEPH_OSD_FLAG_READ, NULL, ci->i_truncate_seq,
 384			ci->i_truncate_size, false);
 385	if (IS_ERR(req)) {
 386		err = PTR_ERR(req);
 387		req = NULL;
 388		goto out;
 389	}
 390
 391	if (sparse) {
 392		extent_cnt = __ceph_sparse_read_ext_count(inode, len);
 393		err = ceph_alloc_sparse_ext_map(&req->r_ops[0], extent_cnt);
 394		if (err)
 395			goto out;
 396	}
 397
 398	doutc(cl, "%llx.%llx pos=%llu orig_len=%zu len=%llu\n",
 399	      ceph_vinop(inode), subreq->start, subreq->len, len);
 400
 401	/*
 402	 * FIXME: For now, use CEPH_OSD_DATA_TYPE_PAGES instead of _ITER for
 403	 * encrypted inodes. We'd need infrastructure that handles an iov_iter
 404	 * instead of page arrays, and we don't have that as of yet. Once the
 405	 * dust settles on the write helpers and encrypt/decrypt routines for
 406	 * netfs, we should be able to rework this.
 407	 */
 408	if (IS_ENCRYPTED(inode)) {
 409		struct page **pages;
 410		size_t page_off;
 411
 412		/*
 413		 * FIXME: io_iter.count needs to be corrected to aligned
 414		 * length. Otherwise, iov_iter_get_pages_alloc2() operates
 415		 * with the initial unaligned length value. As a result,
 416		 * ceph_msg_data_cursor_init() triggers BUG_ON() in the case
 417		 * if msg->sparse_read_total > msg->data_length.
 418		 */
 419		subreq->io_iter.count = len;
 420
 421		err = iov_iter_get_pages_alloc2(&subreq->io_iter, &pages, len, &page_off);
 422		if (err < 0) {
 423			doutc(cl, "%llx.%llx failed to allocate pages, %d\n",
 424			      ceph_vinop(inode), err);
 425			goto out;
 426		}
 427
 428		/* should always give us a page-aligned read */
 429		WARN_ON_ONCE(page_off);
 430		len = err;
 431		err = 0;
 432
 433		osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0, false,
 434						 false);
 435	} else {
 436		osd_req_op_extent_osd_iter(req, 0, &subreq->io_iter);
 437	}
 438	if (!ceph_inc_osd_stopping_blocker(fsc->mdsc)) {
 439		err = -EIO;
 440		goto out;
 441	}
 442	req->r_callback = finish_netfs_read;
 443	req->r_priv = subreq;
 444	req->r_inode = inode;
 445	ihold(inode);
 446
 447	trace_netfs_sreq(subreq, netfs_sreq_trace_submit);
 448	ceph_osdc_start_request(req->r_osdc, req);
 449out:
 450	ceph_osdc_put_request(req);
 451	if (err) {
 452		subreq->error = err;
 453		netfs_read_subreq_terminated(subreq);
 454	}
 455	doutc(cl, "%llx.%llx result %d\n", ceph_vinop(inode), err);
 456}
 457
 458static int ceph_init_request(struct netfs_io_request *rreq, struct file *file)
 459{
 460	struct inode *inode = rreq->inode;
 461	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
 462	struct ceph_client *cl = ceph_inode_to_client(inode);
 463	int got = 0, want = CEPH_CAP_FILE_CACHE;
 464	struct ceph_netfs_request_data *priv;
 465	int ret = 0;
 466
 467	/* [DEPRECATED] Use PG_private_2 to mark folio being written to the cache. */
 468	__set_bit(NETFS_RREQ_USE_PGPRIV2, &rreq->flags);
 469
 470	if (rreq->origin != NETFS_READAHEAD)
 471		return 0;
 472
 473	priv = kzalloc(sizeof(*priv), GFP_NOFS);
 474	if (!priv)
 475		return -ENOMEM;
 476
 477	if (file) {
 478		struct ceph_rw_context *rw_ctx;
 479		struct ceph_file_info *fi = file->private_data;
 480
 481		priv->file_ra_pages = file->f_ra.ra_pages;
 482		priv->file_ra_disabled = file->f_mode & FMODE_RANDOM;
 483
 484		rw_ctx = ceph_find_rw_context(fi);
 485		if (rw_ctx) {
 486			rreq->netfs_priv = priv;
 487			return 0;
 488		}
 489	}
 490
 491	/*
 492	 * readahead callers do not necessarily hold Fcb caps
 493	 * (e.g. fadvise, madvise).
 494	 */
 495	ret = ceph_try_get_caps(inode, CEPH_CAP_FILE_RD, want, true, &got);
 496	if (ret < 0) {
 497		doutc(cl, "%llx.%llx, error getting cap\n", ceph_vinop(inode));
 498		goto out;
 499	}
 500
 501	if (!(got & want)) {
 502		doutc(cl, "%llx.%llx, no cache cap\n", ceph_vinop(inode));
 503		ret = -EACCES;
 504		goto out;
 505	}
 506	if (ret == 0) {
 507		ret = -EACCES;
 508		goto out;
 509	}
 510
 511	priv->caps = got;
 512	rreq->netfs_priv = priv;
 513	rreq->io_streams[0].sreq_max_len = fsc->mount_options->rsize;
 514
 515out:
 516	if (ret < 0) {
 517		if (got)
 518			ceph_put_cap_refs(ceph_inode(inode), got);
 519		kfree(priv);
 520	}
 521
 522	return ret;
 523}
 524
 525static void ceph_netfs_free_request(struct netfs_io_request *rreq)
 526{
 527	struct ceph_netfs_request_data *priv = rreq->netfs_priv;
 528
 529	if (!priv)
 530		return;
 531
 532	if (priv->caps)
 533		ceph_put_cap_refs(ceph_inode(rreq->inode), priv->caps);
 534	kfree(priv);
 535	rreq->netfs_priv = NULL;
 536}
 537
 538const struct netfs_request_ops ceph_netfs_ops = {
 539	.init_request		= ceph_init_request,
 540	.free_request		= ceph_netfs_free_request,
 541	.prepare_read		= ceph_netfs_prepare_read,
 542	.issue_read		= ceph_netfs_issue_read,
 543	.expand_readahead	= ceph_netfs_expand_readahead,
 544	.check_write_begin	= ceph_netfs_check_write_begin,
 545};
 546
 547#ifdef CONFIG_CEPH_FSCACHE
 548static void ceph_set_page_fscache(struct page *page)
 549{
 550	folio_start_private_2(page_folio(page)); /* [DEPRECATED] */
 551}
 552
 553static void ceph_fscache_write_terminated(void *priv, ssize_t error)
 554{
 555	struct inode *inode = priv;
 556
 557	if (IS_ERR_VALUE(error) && error != -ENOBUFS)
 558		ceph_fscache_invalidate(inode, false);
 559}
 560
 561static void ceph_fscache_write_to_cache(struct inode *inode, u64 off, u64 len, bool caching)
 562{
 563	struct ceph_inode_info *ci = ceph_inode(inode);
 564	struct fscache_cookie *cookie = ceph_fscache_cookie(ci);
 565
 566	fscache_write_to_cache(cookie, inode->i_mapping, off, len, i_size_read(inode),
 567			       ceph_fscache_write_terminated, inode, true, caching);
 568}
 569#else
 570static inline void ceph_set_page_fscache(struct page *page)
 571{
 572}
 573
 574static inline void ceph_fscache_write_to_cache(struct inode *inode, u64 off, u64 len, bool caching)
 575{
 576}
 577#endif /* CONFIG_CEPH_FSCACHE */
 578
 579struct ceph_writeback_ctl
 580{
 581	loff_t i_size;
 582	u64 truncate_size;
 583	u32 truncate_seq;
 584	bool size_stable;
 585
 586	bool head_snapc;
 587	struct ceph_snap_context *snapc;
 588	struct ceph_snap_context *last_snapc;
 589
 590	bool done;
 591	bool should_loop;
 592	bool range_whole;
 593	pgoff_t start_index;
 594	pgoff_t index;
 595	pgoff_t end;
 596	xa_mark_t tag;
 597
 598	pgoff_t strip_unit_end;
 599	unsigned int wsize;
 600	unsigned int nr_folios;
 601	unsigned int max_pages;
 602	unsigned int locked_pages;
 603
 604	int op_idx;
 605	int num_ops;
 606	u64 offset;
 607	u64 len;
 608
 609	struct folio_batch fbatch;
 610	unsigned int processed_in_fbatch;
 611
 612	bool from_pool;
 613	struct page **pages;
 614	struct page **data_pages;
 615};
 616
 617/*
 618 * Get ref for the oldest snapc for an inode with dirty data... that is, the
 619 * only snap context we are allowed to write back.
 620 */
 621static struct ceph_snap_context *
 622get_oldest_context(struct inode *inode, struct ceph_writeback_ctl *ctl,
 623		   struct ceph_snap_context *page_snapc)
 624{
 625	struct ceph_inode_info *ci = ceph_inode(inode);
 626	struct ceph_client *cl = ceph_inode_to_client(inode);
 627	struct ceph_snap_context *snapc = NULL;
 628	struct ceph_cap_snap *capsnap = NULL;
 629
 630	spin_lock(&ci->i_ceph_lock);
 631	list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
 632		doutc(cl, " capsnap %p snapc %p has %d dirty pages\n",
 633		      capsnap, capsnap->context, capsnap->dirty_pages);
 634		if (!capsnap->dirty_pages)
 635			continue;
 636
 637		/* get i_size, truncate_{seq,size} for page_snapc? */
 638		if (snapc && capsnap->context != page_snapc)
 639			continue;
 640
 641		if (ctl) {
 642			if (capsnap->writing) {
 643				ctl->i_size = i_size_read(inode);
 644				ctl->size_stable = false;
 645			} else {
 646				ctl->i_size = capsnap->size;
 647				ctl->size_stable = true;
 648			}
 649			ctl->truncate_size = capsnap->truncate_size;
 650			ctl->truncate_seq = capsnap->truncate_seq;
 651			ctl->head_snapc = false;
 652		}
 653
 654		if (snapc)
 655			break;
 656
 657		snapc = ceph_get_snap_context(capsnap->context);
 658		if (!page_snapc ||
 659		    page_snapc == snapc ||
 660		    page_snapc->seq > snapc->seq)
 661			break;
 662	}
 663	if (!snapc && ci->i_wrbuffer_ref_head) {
 664		snapc = ceph_get_snap_context(ci->i_head_snapc);
 665		doutc(cl, " head snapc %p has %d dirty pages\n", snapc,
 666		      ci->i_wrbuffer_ref_head);
 667		if (ctl) {
 668			ctl->i_size = i_size_read(inode);
 669			ctl->truncate_size = ci->i_truncate_size;
 670			ctl->truncate_seq = ci->i_truncate_seq;
 671			ctl->size_stable = false;
 672			ctl->head_snapc = true;
 673		}
 674	}
 675	spin_unlock(&ci->i_ceph_lock);
 676	return snapc;
 677}
 678
 679static u64 get_writepages_data_length(struct inode *inode,
 680				      struct page *page, u64 start)
 681{
 682	struct ceph_inode_info *ci = ceph_inode(inode);
 683	struct ceph_snap_context *snapc;
 684	struct ceph_cap_snap *capsnap = NULL;
 685	u64 end = i_size_read(inode);
 686	u64 ret;
 687
 688	snapc = page_snap_context(ceph_fscrypt_pagecache_page(page));
 689	if (snapc != ci->i_head_snapc) {
 690		bool found = false;
 691		spin_lock(&ci->i_ceph_lock);
 692		list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
 693			if (capsnap->context == snapc) {
 694				if (!capsnap->writing)
 695					end = capsnap->size;
 696				found = true;
 697				break;
 698			}
 699		}
 700		spin_unlock(&ci->i_ceph_lock);
 701		WARN_ON(!found);
 702	}
 703	if (end > ceph_fscrypt_page_offset(page) + thp_size(page))
 704		end = ceph_fscrypt_page_offset(page) + thp_size(page);
 705	ret = end > start ? end - start : 0;
 706	if (ret && fscrypt_is_bounce_page(page))
 707		ret = round_up(ret, CEPH_FSCRYPT_BLOCK_SIZE);
 708	return ret;
 709}
 710
 711/*
 712 * Write a folio, but leave it locked.
 713 *
 714 * If we get a write error, mark the mapping for error, but still adjust the
 715 * dirty page accounting (i.e., folio is no longer dirty).
 716 */
 717static int write_folio_nounlock(struct folio *folio,
 718		struct writeback_control *wbc)
 719{
 720	struct page *page = &folio->page;
 721	struct inode *inode = folio->mapping->host;
 722	struct ceph_inode_info *ci = ceph_inode(inode);
 723	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
 724	struct ceph_client *cl = fsc->client;
 725	struct ceph_snap_context *snapc, *oldest;
 726	loff_t page_off = folio_pos(folio);
 727	int err;
 728	loff_t len = folio_size(folio);
 729	loff_t wlen;
 730	struct ceph_writeback_ctl ceph_wbc;
 731	struct ceph_osd_client *osdc = &fsc->client->osdc;
 732	struct ceph_osd_request *req;
 733	bool caching = ceph_is_cache_enabled(inode);
 734	struct page *bounce_page = NULL;
 735
 736	doutc(cl, "%llx.%llx folio %p idx %lu\n", ceph_vinop(inode), folio,
 737	      folio->index);
 738
 739	if (ceph_inode_is_shutdown(inode))
 740		return -EIO;
 741
 742	/* verify this is a writeable snap context */
 743	snapc = page_snap_context(&folio->page);
 744	if (!snapc) {
 745		doutc(cl, "%llx.%llx folio %p not dirty?\n", ceph_vinop(inode),
 746		      folio);
 747		return 0;
 748	}
 749	oldest = get_oldest_context(inode, &ceph_wbc, snapc);
 750	if (snapc->seq > oldest->seq) {
 751		doutc(cl, "%llx.%llx folio %p snapc %p not writeable - noop\n",
 752		      ceph_vinop(inode), folio, snapc);
 753		/* we should only noop if called by kswapd */
 754		WARN_ON(!(current->flags & PF_MEMALLOC));
 755		ceph_put_snap_context(oldest);
 756		folio_redirty_for_writepage(wbc, folio);
 757		return 0;
 758	}
 759	ceph_put_snap_context(oldest);
 760
 761	/* is this a partial page at end of file? */
 762	if (page_off >= ceph_wbc.i_size) {
 763		doutc(cl, "%llx.%llx folio at %lu beyond eof %llu\n",
 764		      ceph_vinop(inode), folio->index, ceph_wbc.i_size);
 765		folio_invalidate(folio, 0, folio_size(folio));
 766		return 0;
 767	}
 768
 769	if (ceph_wbc.i_size < page_off + len)
 770		len = ceph_wbc.i_size - page_off;
 771
 772	wlen = IS_ENCRYPTED(inode) ? round_up(len, CEPH_FSCRYPT_BLOCK_SIZE) : len;
 773	doutc(cl, "%llx.%llx folio %p index %lu on %llu~%llu snapc %p seq %lld\n",
 774	      ceph_vinop(inode), folio, folio->index, page_off, wlen, snapc,
 775	      snapc->seq);
 776
 777	if (atomic_long_inc_return(&fsc->writeback_count) >
 778	    CONGESTION_ON_THRESH(fsc->mount_options->congestion_kb))
 779		fsc->write_congested = true;
 780
 781	req = ceph_osdc_new_request(osdc, &ci->i_layout, ceph_vino(inode),
 782				    page_off, &wlen, 0, 1, CEPH_OSD_OP_WRITE,
 783				    CEPH_OSD_FLAG_WRITE, snapc,
 784				    ceph_wbc.truncate_seq,
 785				    ceph_wbc.truncate_size, true);
 786	if (IS_ERR(req)) {
 787		folio_redirty_for_writepage(wbc, folio);
 788		return PTR_ERR(req);
 789	}
 790
 791	if (wlen < len)
 792		len = wlen;
 793
 794	folio_start_writeback(folio);
 795	if (caching)
 796		ceph_set_page_fscache(&folio->page);
 797	ceph_fscache_write_to_cache(inode, page_off, len, caching);
 798
 799	if (IS_ENCRYPTED(inode)) {
 800		bounce_page = fscrypt_encrypt_pagecache_blocks(folio,
 801						    CEPH_FSCRYPT_BLOCK_SIZE, 0,
 802						    GFP_NOFS);
 803		if (IS_ERR(bounce_page)) {
 804			folio_redirty_for_writepage(wbc, folio);
 805			folio_end_writeback(folio);
 806			ceph_osdc_put_request(req);
 807			return PTR_ERR(bounce_page);
 808		}
 809	}
 810
 811	/* it may be a short write due to an object boundary */
 812	WARN_ON_ONCE(len > folio_size(folio));
 813	osd_req_op_extent_osd_data_pages(req, 0,
 814			bounce_page ? &bounce_page : &page, wlen, 0,
 815			false, false);
 816	doutc(cl, "%llx.%llx %llu~%llu (%llu bytes, %sencrypted)\n",
 817	      ceph_vinop(inode), page_off, len, wlen,
 818	      IS_ENCRYPTED(inode) ? "" : "not ");
 819
 820	req->r_mtime = inode_get_mtime(inode);
 821	ceph_osdc_start_request(osdc, req);
 822	err = ceph_osdc_wait_request(osdc, req);
 823
 824	ceph_update_write_metrics(&fsc->mdsc->metric, req->r_start_latency,
 825				  req->r_end_latency, len, err);
 826	fscrypt_free_bounce_page(bounce_page);
 827	ceph_osdc_put_request(req);
 828	if (err == 0)
 829		err = len;
 830
 831	if (err < 0) {
 832		struct writeback_control tmp_wbc;
 833		if (!wbc)
 834			wbc = &tmp_wbc;
 835		if (err == -ERESTARTSYS) {
 836			/* killed by SIGKILL */
 837			doutc(cl, "%llx.%llx interrupted page %p\n",
 838			      ceph_vinop(inode), folio);
 839			folio_redirty_for_writepage(wbc, folio);
 840			folio_end_writeback(folio);
 841			return err;
 842		}
 843		if (err == -EBLOCKLISTED)
 844			fsc->blocklisted = true;
 845		doutc(cl, "%llx.%llx setting mapping error %d %p\n",
 846		      ceph_vinop(inode), err, folio);
 847		mapping_set_error(&inode->i_data, err);
 848		wbc->pages_skipped++;
 849	} else {
 850		doutc(cl, "%llx.%llx cleaned page %p\n",
 851		      ceph_vinop(inode), folio);
 852		err = 0;  /* vfs expects us to return 0 */
 853	}
 854	oldest = folio_detach_private(folio);
 855	WARN_ON_ONCE(oldest != snapc);
 856	folio_end_writeback(folio);
 857	ceph_put_wrbuffer_cap_refs(ci, 1, snapc);
 858	ceph_put_snap_context(snapc);  /* page's reference */
 859
 860	if (atomic_long_dec_return(&fsc->writeback_count) <
 861	    CONGESTION_OFF_THRESH(fsc->mount_options->congestion_kb))
 862		fsc->write_congested = false;
 863
 864	return err;
 865}
 866
 867/*
 868 * async writeback completion handler.
 869 *
 870 * If we get an error, set the mapping error bit, but not the individual
 871 * page error bits.
 872 */
 873static void writepages_finish(struct ceph_osd_request *req)
 874{
 875	struct inode *inode = req->r_inode;
 876	struct ceph_inode_info *ci = ceph_inode(inode);
 877	struct ceph_client *cl = ceph_inode_to_client(inode);
 878	struct ceph_osd_data *osd_data;
 879	struct page *page;
 880	int num_pages, total_pages = 0;
 881	int i, j;
 882	int rc = req->r_result;
 883	struct ceph_snap_context *snapc = req->r_snapc;
 884	struct address_space *mapping = inode->i_mapping;
 885	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
 886	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
 887	unsigned int len = 0;
 888	bool remove_page;
 889
 890	doutc(cl, "%llx.%llx rc %d\n", ceph_vinop(inode), rc);
 891	if (rc < 0) {
 892		mapping_set_error(mapping, rc);
 893		ceph_set_error_write(ci);
 894		if (rc == -EBLOCKLISTED)
 895			fsc->blocklisted = true;
 896	} else {
 897		ceph_clear_error_write(ci);
 898	}
 899
 900	/*
 901	 * We lost the cache cap, need to truncate the page before
 902	 * it is unlocked, otherwise we'd truncate it later in the
 903	 * page truncation thread, possibly losing some data that
 904	 * raced its way in
 905	 */
 906	remove_page = !(ceph_caps_issued(ci) &
 907			(CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO));
 908
 909	/* clean all pages */
 910	for (i = 0; i < req->r_num_ops; i++) {
 911		if (req->r_ops[i].op != CEPH_OSD_OP_WRITE) {
 912			pr_warn_client(cl,
 913				"%llx.%llx incorrect op %d req %p index %d tid %llu\n",
 914				ceph_vinop(inode), req->r_ops[i].op, req, i,
 915				req->r_tid);
 916			break;
 917		}
 918
 919		osd_data = osd_req_op_extent_osd_data(req, i);
 920		BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGES);
 921		len += osd_data->length;
 922		num_pages = calc_pages_for((u64)osd_data->alignment,
 923					   (u64)osd_data->length);
 924		total_pages += num_pages;
 925		for (j = 0; j < num_pages; j++) {
 926			page = osd_data->pages[j];
 927			if (fscrypt_is_bounce_page(page)) {
 928				page = fscrypt_pagecache_page(page);
 929				fscrypt_free_bounce_page(osd_data->pages[j]);
 930				osd_data->pages[j] = page;
 931			}
 932			BUG_ON(!page);
 933			WARN_ON(!PageUptodate(page));
 934
 935			if (atomic_long_dec_return(&fsc->writeback_count) <
 936			     CONGESTION_OFF_THRESH(
 937					fsc->mount_options->congestion_kb))
 938				fsc->write_congested = false;
 939
 940			ceph_put_snap_context(detach_page_private(page));
 941			end_page_writeback(page);
 942
 943			if (atomic64_dec_return(&mdsc->dirty_folios) <= 0) {
 944				wake_up_all(&mdsc->flush_end_wq);
 945				WARN_ON(atomic64_read(&mdsc->dirty_folios) < 0);
 946			}
 947
 948			doutc(cl, "unlocking %p\n", page);
 949
 950			if (remove_page)
 951				generic_error_remove_folio(inode->i_mapping,
 952							  page_folio(page));
 953
 954			unlock_page(page);
 955		}
 956		doutc(cl, "%llx.%llx wrote %llu bytes cleaned %d pages\n",
 957		      ceph_vinop(inode), osd_data->length,
 958		      rc >= 0 ? num_pages : 0);
 959
 960		release_pages(osd_data->pages, num_pages);
 961	}
 962
 963	ceph_update_write_metrics(&fsc->mdsc->metric, req->r_start_latency,
 964				  req->r_end_latency, len, rc);
 965
 966	ceph_put_wrbuffer_cap_refs(ci, total_pages, snapc);
 967
 968	osd_data = osd_req_op_extent_osd_data(req, 0);
 969	if (osd_data->pages_from_pool)
 970		mempool_free(osd_data->pages, ceph_wb_pagevec_pool);
 971	else
 972		kfree(osd_data->pages);
 973	ceph_osdc_put_request(req);
 974	ceph_dec_osd_stopping_blocker(fsc->mdsc);
 975}
 976
 977static inline
 978bool is_forced_umount(struct address_space *mapping)
 979{
 980	struct inode *inode = mapping->host;
 981	struct ceph_inode_info *ci = ceph_inode(inode);
 982	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
 983	struct ceph_client *cl = fsc->client;
 984
 985	if (ceph_inode_is_shutdown(inode)) {
 986		if (ci->i_wrbuffer_ref > 0) {
 987			pr_warn_ratelimited_client(cl,
 988				"%llx.%llx %lld forced umount\n",
 989				ceph_vinop(inode), ceph_ino(inode));
 990		}
 991		mapping_set_error(mapping, -EIO);
 992		return true;
 993	}
 994
 995	return false;
 996}
 997
 998static inline
 999unsigned int ceph_define_write_size(struct address_space *mapping)
1000{
1001	struct inode *inode = mapping->host;
1002	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1003	struct ceph_inode_info *ci = ceph_inode(inode);
1004	unsigned int wsize = ci->i_layout.stripe_unit;
1005
1006	if (fsc->mount_options->wsize < wsize)
1007		wsize = fsc->mount_options->wsize;
1008
1009	return wsize;
1010}
1011
1012static inline
1013void ceph_folio_batch_init(struct ceph_writeback_ctl *ceph_wbc)
1014{
1015	folio_batch_init(&ceph_wbc->fbatch);
1016	ceph_wbc->processed_in_fbatch = 0;
1017}
1018
1019static inline
1020void ceph_folio_batch_reinit(struct ceph_writeback_ctl *ceph_wbc)
1021{
1022	folio_batch_release(&ceph_wbc->fbatch);
1023	ceph_folio_batch_init(ceph_wbc);
1024}
1025
1026static inline
1027void ceph_init_writeback_ctl(struct address_space *mapping,
1028			     struct writeback_control *wbc,
1029			     struct ceph_writeback_ctl *ceph_wbc)
1030{
1031	ceph_wbc->snapc = NULL;
1032	ceph_wbc->last_snapc = NULL;
1033
1034	ceph_wbc->strip_unit_end = 0;
1035	ceph_wbc->wsize = ceph_define_write_size(mapping);
1036
1037	ceph_wbc->nr_folios = 0;
1038	ceph_wbc->max_pages = 0;
1039	ceph_wbc->locked_pages = 0;
1040
1041	ceph_wbc->done = false;
1042	ceph_wbc->should_loop = false;
1043	ceph_wbc->range_whole = false;
1044
1045	ceph_wbc->start_index = wbc->range_cyclic ? mapping->writeback_index : 0;
1046	ceph_wbc->index = ceph_wbc->start_index;
1047	ceph_wbc->end = -1;
1048
1049	ceph_wbc->tag = wbc_to_tag(wbc);
1050
1051	ceph_wbc->op_idx = -1;
1052	ceph_wbc->num_ops = 0;
1053	ceph_wbc->offset = 0;
1054	ceph_wbc->len = 0;
1055	ceph_wbc->from_pool = false;
1056
1057	ceph_folio_batch_init(ceph_wbc);
1058
1059	ceph_wbc->pages = NULL;
1060	ceph_wbc->data_pages = NULL;
1061}
1062
1063static inline
1064int ceph_define_writeback_range(struct address_space *mapping,
1065				struct writeback_control *wbc,
1066				struct ceph_writeback_ctl *ceph_wbc)
1067{
1068	struct inode *inode = mapping->host;
1069	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1070	struct ceph_client *cl = fsc->client;
1071
1072	/* find oldest snap context with dirty data */
1073	ceph_wbc->snapc = get_oldest_context(inode, ceph_wbc, NULL);
1074	if (!ceph_wbc->snapc) {
1075		/* hmm, why does writepages get called when there
1076		   is no dirty data? */
1077		doutc(cl, " no snap context with dirty data?\n");
1078		return -ENODATA;
1079	}
1080
1081	doutc(cl, " oldest snapc is %p seq %lld (%d snaps)\n",
1082	      ceph_wbc->snapc, ceph_wbc->snapc->seq,
1083	      ceph_wbc->snapc->num_snaps);
1084
1085	ceph_wbc->should_loop = false;
1086
1087	if (ceph_wbc->head_snapc && ceph_wbc->snapc != ceph_wbc->last_snapc) {
1088		/* where to start/end? */
1089		if (wbc->range_cyclic) {
1090			ceph_wbc->index = ceph_wbc->start_index;
1091			ceph_wbc->end = -1;
1092			if (ceph_wbc->index > 0)
1093				ceph_wbc->should_loop = true;
1094			doutc(cl, " cyclic, start at %lu\n", ceph_wbc->index);
1095		} else {
1096			ceph_wbc->index = wbc->range_start >> PAGE_SHIFT;
1097			ceph_wbc->end = wbc->range_end >> PAGE_SHIFT;
1098			if (wbc->range_start == 0 && wbc->range_end == LLONG_MAX)
1099				ceph_wbc->range_whole = true;
1100			doutc(cl, " not cyclic, %lu to %lu\n",
1101				ceph_wbc->index, ceph_wbc->end);
1102		}
1103	} else if (!ceph_wbc->head_snapc) {
1104		/* Do not respect wbc->range_{start,end}. Dirty pages
1105		 * in that range can be associated with newer snapc.
1106		 * They are not writeable until we write all dirty pages
1107		 * associated with 'snapc' get written */
1108		if (ceph_wbc->index > 0)
1109			ceph_wbc->should_loop = true;
1110		doutc(cl, " non-head snapc, range whole\n");
1111	}
1112
1113	ceph_put_snap_context(ceph_wbc->last_snapc);
1114	ceph_wbc->last_snapc = ceph_wbc->snapc;
1115
1116	return 0;
1117}
1118
1119static inline
1120bool has_writeback_done(struct ceph_writeback_ctl *ceph_wbc)
1121{
1122	return ceph_wbc->done && ceph_wbc->index > ceph_wbc->end;
1123}
1124
1125static inline
1126bool can_next_page_be_processed(struct ceph_writeback_ctl *ceph_wbc,
1127				unsigned index)
1128{
1129	return index < ceph_wbc->nr_folios &&
1130		ceph_wbc->locked_pages < ceph_wbc->max_pages;
1131}
1132
1133static
1134int ceph_check_page_before_write(struct address_space *mapping,
1135				 struct writeback_control *wbc,
1136				 struct ceph_writeback_ctl *ceph_wbc,
1137				 struct folio *folio)
1138{
1139	struct inode *inode = mapping->host;
1140	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1141	struct ceph_client *cl = fsc->client;
1142	struct ceph_snap_context *pgsnapc;
1143
1144	/* only dirty folios, or our accounting breaks */
1145	if (unlikely(!folio_test_dirty(folio) || folio->mapping != mapping)) {
1146		doutc(cl, "!dirty or !mapping %p\n", folio);
1147		return -ENODATA;
1148	}
1149
1150	/* only if matching snap context */
1151	pgsnapc = page_snap_context(&folio->page);
1152	if (pgsnapc != ceph_wbc->snapc) {
1153		doutc(cl, "folio snapc %p %lld != oldest %p %lld\n",
1154		      pgsnapc, pgsnapc->seq,
1155		      ceph_wbc->snapc, ceph_wbc->snapc->seq);
1156
1157		if (!ceph_wbc->should_loop && !ceph_wbc->head_snapc &&
1158		    wbc->sync_mode != WB_SYNC_NONE)
1159			ceph_wbc->should_loop = true;
1160
1161		return -ENODATA;
1162	}
1163
1164	if (folio_pos(folio) >= ceph_wbc->i_size) {
1165		doutc(cl, "folio at %lu beyond eof %llu\n",
1166		      folio->index, ceph_wbc->i_size);
1167
1168		if ((ceph_wbc->size_stable ||
1169		    folio_pos(folio) >= i_size_read(inode)) &&
1170		    folio_clear_dirty_for_io(folio))
1171			folio_invalidate(folio, 0, folio_size(folio));
1172
1173		return -ENODATA;
1174	}
1175
1176	if (ceph_wbc->strip_unit_end &&
1177	    (folio->index > ceph_wbc->strip_unit_end)) {
1178		doutc(cl, "end of strip unit %p\n", folio);
1179		return -E2BIG;
1180	}
1181
1182	return 0;
1183}
1184
1185static inline
1186void __ceph_allocate_page_array(struct ceph_writeback_ctl *ceph_wbc,
1187				unsigned int max_pages)
1188{
1189	ceph_wbc->pages = kmalloc_array(max_pages,
1190					sizeof(*ceph_wbc->pages),
1191					GFP_NOFS);
1192	if (!ceph_wbc->pages) {
1193		ceph_wbc->from_pool = true;
1194		ceph_wbc->pages = mempool_alloc(ceph_wb_pagevec_pool, GFP_NOFS);
1195		BUG_ON(!ceph_wbc->pages);
1196	}
1197}
1198
1199static inline
1200void ceph_allocate_page_array(struct address_space *mapping,
1201			      struct ceph_writeback_ctl *ceph_wbc,
1202			      struct folio *folio)
1203{
1204	struct inode *inode = mapping->host;
1205	struct ceph_inode_info *ci = ceph_inode(inode);
1206	u64 objnum;
1207	u64 objoff;
1208	u32 xlen;
1209
1210	/* prepare async write request */
1211	ceph_wbc->offset = (u64)folio_pos(folio);
1212	ceph_calc_file_object_mapping(&ci->i_layout,
1213					ceph_wbc->offset, ceph_wbc->wsize,
1214					&objnum, &objoff, &xlen);
1215
1216	ceph_wbc->num_ops = 1;
1217	ceph_wbc->strip_unit_end = folio->index + ((xlen - 1) >> PAGE_SHIFT);
1218
1219	BUG_ON(ceph_wbc->pages);
1220	ceph_wbc->max_pages = calc_pages_for(0, (u64)xlen);
1221	__ceph_allocate_page_array(ceph_wbc, ceph_wbc->max_pages);
1222
1223	ceph_wbc->len = 0;
1224}
1225
1226static inline
1227bool is_folio_index_contiguous(const struct ceph_writeback_ctl *ceph_wbc,
1228			      const struct folio *folio)
1229{
1230	return folio->index == (ceph_wbc->offset + ceph_wbc->len) >> PAGE_SHIFT;
1231}
1232
1233static inline
1234bool is_num_ops_too_big(struct ceph_writeback_ctl *ceph_wbc)
1235{
1236	return ceph_wbc->num_ops >=
1237		(ceph_wbc->from_pool ?  CEPH_OSD_SLAB_OPS : CEPH_OSD_MAX_OPS);
1238}
1239
1240static inline
1241bool is_write_congestion_happened(struct ceph_fs_client *fsc)
1242{
1243	return atomic_long_inc_return(&fsc->writeback_count) >
1244		CONGESTION_ON_THRESH(fsc->mount_options->congestion_kb);
1245}
1246
1247static inline int move_dirty_folio_in_page_array(struct address_space *mapping,
1248		struct writeback_control *wbc,
1249		struct ceph_writeback_ctl *ceph_wbc, struct folio *folio)
1250{
1251	struct inode *inode = mapping->host;
1252	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1253	struct ceph_client *cl = fsc->client;
1254	struct page **pages = ceph_wbc->pages;
1255	unsigned int index = ceph_wbc->locked_pages;
1256	gfp_t gfp_flags = ceph_wbc->locked_pages ? GFP_NOWAIT : GFP_NOFS;
1257
1258	if (IS_ENCRYPTED(inode)) {
1259		pages[index] = fscrypt_encrypt_pagecache_blocks(folio,
1260								PAGE_SIZE,
1261								0,
1262								gfp_flags);
1263		if (IS_ERR(pages[index])) {
1264			int err = PTR_ERR(pages[index]);
1265
1266			if (err == -EINVAL) {
1267				pr_err_client(cl, "inode->i_blkbits=%hhu\n",
1268						inode->i_blkbits);
1269			}
1270
1271			/* better not fail on first page! */
1272			BUG_ON(ceph_wbc->locked_pages == 0);
1273
1274			pages[index] = NULL;
1275			return err;
1276		}
1277	} else {
1278		pages[index] = &folio->page;
1279	}
1280
1281	ceph_wbc->locked_pages++;
1282
1283	return 0;
1284}
1285
1286static
1287int ceph_process_folio_batch(struct address_space *mapping,
1288			     struct writeback_control *wbc,
1289			     struct ceph_writeback_ctl *ceph_wbc)
1290{
1291	struct inode *inode = mapping->host;
1292	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1293	struct ceph_client *cl = fsc->client;
1294	struct folio *folio = NULL;
1295	unsigned i;
1296	int rc = 0;
1297
1298	for (i = 0; can_next_page_be_processed(ceph_wbc, i); i++) {
1299		folio = ceph_wbc->fbatch.folios[i];
1300
1301		if (!folio)
1302			continue;
1303
1304		doutc(cl, "? %p idx %lu, folio_test_writeback %#x, "
1305			"folio_test_dirty %#x, folio_test_locked %#x\n",
1306			folio, folio->index, folio_test_writeback(folio),
1307			folio_test_dirty(folio),
1308			folio_test_locked(folio));
1309
1310		if (folio_test_writeback(folio) ||
1311		    folio_test_private_2(folio) /* [DEPRECATED] */) {
1312			doutc(cl, "waiting on writeback %p\n", folio);
1313			folio_wait_writeback(folio);
1314			folio_wait_private_2(folio); /* [DEPRECATED] */
1315			continue;
1316		}
1317
1318		if (ceph_wbc->locked_pages == 0)
1319			folio_lock(folio);
1320		else if (!folio_trylock(folio))
1321			break;
1322
1323		rc = ceph_check_page_before_write(mapping, wbc,
1324						  ceph_wbc, folio);
1325		if (rc == -ENODATA) {
1326			rc = 0;
1327			folio_unlock(folio);
1328			ceph_wbc->fbatch.folios[i] = NULL;
1329			continue;
1330		} else if (rc == -E2BIG) {
1331			rc = 0;
1332			folio_unlock(folio);
1333			break;
1334		}
1335
1336		if (!folio_clear_dirty_for_io(folio)) {
1337			doutc(cl, "%p !folio_clear_dirty_for_io\n", folio);
1338			folio_unlock(folio);
1339			ceph_wbc->fbatch.folios[i] = NULL;
1340			continue;
1341		}
1342
1343		/*
1344		 * We have something to write.  If this is
1345		 * the first locked page this time through,
1346		 * calculate max possible write size and
1347		 * allocate a page array
1348		 */
1349		if (ceph_wbc->locked_pages == 0) {
1350			ceph_allocate_page_array(mapping, ceph_wbc, folio);
1351		} else if (!is_folio_index_contiguous(ceph_wbc, folio)) {
1352			if (is_num_ops_too_big(ceph_wbc)) {
1353				folio_redirty_for_writepage(wbc, folio);
1354				folio_unlock(folio);
1355				break;
1356			}
1357
1358			ceph_wbc->num_ops++;
1359			ceph_wbc->offset = (u64)folio_pos(folio);
1360			ceph_wbc->len = 0;
1361		}
1362
1363		/* note position of first page in fbatch */
1364		doutc(cl, "%llx.%llx will write folio %p idx %lu\n",
1365		      ceph_vinop(inode), folio, folio->index);
1366
1367		fsc->write_congested = is_write_congestion_happened(fsc);
1368
1369		rc = move_dirty_folio_in_page_array(mapping, wbc, ceph_wbc,
1370				folio);
1371		if (rc) {
1372			rc = 0;
1373			folio_redirty_for_writepage(wbc, folio);
1374			folio_unlock(folio);
1375			break;
1376		}
1377
1378		ceph_wbc->fbatch.folios[i] = NULL;
1379		ceph_wbc->len += folio_size(folio);
1380	}
1381
1382	ceph_wbc->processed_in_fbatch = i;
1383
1384	return rc;
1385}
1386
1387static inline
1388void ceph_shift_unused_folios_left(struct folio_batch *fbatch)
1389{
1390	unsigned j, n = 0;
1391
1392	/* shift unused page to beginning of fbatch */
1393	for (j = 0; j < folio_batch_count(fbatch); j++) {
1394		if (!fbatch->folios[j])
1395			continue;
1396
1397		if (n < j) {
1398			fbatch->folios[n] = fbatch->folios[j];
1399		}
1400
1401		n++;
1402	}
1403
1404	fbatch->nr = n;
1405}
1406
1407static
1408int ceph_submit_write(struct address_space *mapping,
1409			struct writeback_control *wbc,
1410			struct ceph_writeback_ctl *ceph_wbc)
1411{
1412	struct inode *inode = mapping->host;
1413	struct ceph_inode_info *ci = ceph_inode(inode);
1414	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1415	struct ceph_client *cl = fsc->client;
1416	struct ceph_vino vino = ceph_vino(inode);
1417	struct ceph_osd_request *req = NULL;
1418	struct page *page = NULL;
1419	bool caching = ceph_is_cache_enabled(inode);
1420	u64 offset;
1421	u64 len;
1422	unsigned i;
1423
1424new_request:
1425	offset = ceph_fscrypt_page_offset(ceph_wbc->pages[0]);
1426	len = ceph_wbc->wsize;
1427
1428	req = ceph_osdc_new_request(&fsc->client->osdc,
1429				    &ci->i_layout, vino,
1430				    offset, &len, 0, ceph_wbc->num_ops,
1431				    CEPH_OSD_OP_WRITE, CEPH_OSD_FLAG_WRITE,
1432				    ceph_wbc->snapc, ceph_wbc->truncate_seq,
1433				    ceph_wbc->truncate_size, false);
1434	if (IS_ERR(req)) {
1435		req = ceph_osdc_new_request(&fsc->client->osdc,
1436					    &ci->i_layout, vino,
1437					    offset, &len, 0,
1438					    min(ceph_wbc->num_ops,
1439						CEPH_OSD_SLAB_OPS),
1440					    CEPH_OSD_OP_WRITE,
1441					    CEPH_OSD_FLAG_WRITE,
1442					    ceph_wbc->snapc,
1443					    ceph_wbc->truncate_seq,
1444					    ceph_wbc->truncate_size,
1445					    true);
1446		BUG_ON(IS_ERR(req));
1447	}
1448
1449	page = ceph_wbc->pages[ceph_wbc->locked_pages - 1];
1450	BUG_ON(len < ceph_fscrypt_page_offset(page) + thp_size(page) - offset);
1451
1452	if (!ceph_inc_osd_stopping_blocker(fsc->mdsc)) {
1453		for (i = 0; i < folio_batch_count(&ceph_wbc->fbatch); i++) {
1454			struct folio *folio = ceph_wbc->fbatch.folios[i];
1455
1456			if (!folio)
1457				continue;
1458
1459			page = &folio->page;
1460			redirty_page_for_writepage(wbc, page);
1461			unlock_page(page);
1462		}
1463
1464		for (i = 0; i < ceph_wbc->locked_pages; i++) {
1465			page = ceph_fscrypt_pagecache_page(ceph_wbc->pages[i]);
1466
1467			if (!page)
1468				continue;
1469
1470			redirty_page_for_writepage(wbc, page);
1471			unlock_page(page);
1472		}
1473
1474		ceph_osdc_put_request(req);
1475		return -EIO;
1476	}
1477
1478	req->r_callback = writepages_finish;
1479	req->r_inode = inode;
1480
1481	/* Format the osd request message and submit the write */
1482	len = 0;
1483	ceph_wbc->data_pages = ceph_wbc->pages;
1484	ceph_wbc->op_idx = 0;
1485	for (i = 0; i < ceph_wbc->locked_pages; i++) {
1486		u64 cur_offset;
1487
1488		page = ceph_fscrypt_pagecache_page(ceph_wbc->pages[i]);
1489		cur_offset = page_offset(page);
1490
1491		/*
1492		 * Discontinuity in page range? Ceph can handle that by just passing
1493		 * multiple extents in the write op.
1494		 */
1495		if (offset + len != cur_offset) {
1496			/* If it's full, stop here */
1497			if (ceph_wbc->op_idx + 1 == req->r_num_ops)
1498				break;
1499
1500			/* Kick off an fscache write with what we have so far. */
1501			ceph_fscache_write_to_cache(inode, offset, len, caching);
1502
1503			/* Start a new extent */
1504			osd_req_op_extent_dup_last(req, ceph_wbc->op_idx,
1505						   cur_offset - offset);
1506
1507			doutc(cl, "got pages at %llu~%llu\n", offset, len);
1508
1509			osd_req_op_extent_osd_data_pages(req, ceph_wbc->op_idx,
1510							 ceph_wbc->data_pages,
1511							 len, 0,
1512							 ceph_wbc->from_pool,
1513							 false);
1514			osd_req_op_extent_update(req, ceph_wbc->op_idx, len);
1515
1516			len = 0;
1517			offset = cur_offset;
1518			ceph_wbc->data_pages = ceph_wbc->pages + i;
1519			ceph_wbc->op_idx++;
1520		}
1521
1522		set_page_writeback(page);
1523
1524		if (caching)
1525			ceph_set_page_fscache(page);
1526
1527		len += thp_size(page);
1528	}
1529
1530	ceph_fscache_write_to_cache(inode, offset, len, caching);
1531
1532	if (ceph_wbc->size_stable) {
1533		len = min(len, ceph_wbc->i_size - offset);
1534	} else if (i == ceph_wbc->locked_pages) {
1535		/* writepages_finish() clears writeback pages
1536		 * according to the data length, so make sure
1537		 * data length covers all locked pages */
1538		u64 min_len = len + 1 - thp_size(page);
1539		len = get_writepages_data_length(inode,
1540						 ceph_wbc->pages[i - 1],
1541						 offset);
1542		len = max(len, min_len);
1543	}
1544
1545	if (IS_ENCRYPTED(inode))
1546		len = round_up(len, CEPH_FSCRYPT_BLOCK_SIZE);
1547
1548	doutc(cl, "got pages at %llu~%llu\n", offset, len);
1549
1550	if (IS_ENCRYPTED(inode) &&
1551	    ((offset | len) & ~CEPH_FSCRYPT_BLOCK_MASK)) {
1552		pr_warn_client(cl,
1553			"bad encrypted write offset=%lld len=%llu\n",
1554			offset, len);
1555	}
1556
1557	osd_req_op_extent_osd_data_pages(req, ceph_wbc->op_idx,
1558					 ceph_wbc->data_pages, len,
1559					 0, ceph_wbc->from_pool, false);
1560	osd_req_op_extent_update(req, ceph_wbc->op_idx, len);
1561
1562	BUG_ON(ceph_wbc->op_idx + 1 != req->r_num_ops);
1563
1564	ceph_wbc->from_pool = false;
1565	if (i < ceph_wbc->locked_pages) {
1566		BUG_ON(ceph_wbc->num_ops <= req->r_num_ops);
1567		ceph_wbc->num_ops -= req->r_num_ops;
1568		ceph_wbc->locked_pages -= i;
1569
1570		/* allocate new pages array for next request */
1571		ceph_wbc->data_pages = ceph_wbc->pages;
1572		__ceph_allocate_page_array(ceph_wbc, ceph_wbc->locked_pages);
1573		memcpy(ceph_wbc->pages, ceph_wbc->data_pages + i,
1574			ceph_wbc->locked_pages * sizeof(*ceph_wbc->pages));
1575		memset(ceph_wbc->data_pages + i, 0,
1576			ceph_wbc->locked_pages * sizeof(*ceph_wbc->pages));
1577	} else {
1578		BUG_ON(ceph_wbc->num_ops != req->r_num_ops);
1579		/* request message now owns the pages array */
1580		ceph_wbc->pages = NULL;
1581	}
1582
1583	req->r_mtime = inode_get_mtime(inode);
1584	ceph_osdc_start_request(&fsc->client->osdc, req);
1585	req = NULL;
1586
1587	wbc->nr_to_write -= i;
1588	if (ceph_wbc->pages)
1589		goto new_request;
1590
1591	return 0;
1592}
1593
1594static
1595void ceph_wait_until_current_writes_complete(struct address_space *mapping,
1596					     struct writeback_control *wbc,
1597					     struct ceph_writeback_ctl *ceph_wbc)
1598{
1599	struct page *page;
1600	unsigned i, nr;
1601
1602	if (wbc->sync_mode != WB_SYNC_NONE &&
1603	    ceph_wbc->start_index == 0 && /* all dirty pages were checked */
1604	    !ceph_wbc->head_snapc) {
1605		ceph_wbc->index = 0;
1606
1607		while ((ceph_wbc->index <= ceph_wbc->end) &&
1608			(nr = filemap_get_folios_tag(mapping,
1609						     &ceph_wbc->index,
1610						     (pgoff_t)-1,
1611						     PAGECACHE_TAG_WRITEBACK,
1612						     &ceph_wbc->fbatch))) {
1613			for (i = 0; i < nr; i++) {
1614				page = &ceph_wbc->fbatch.folios[i]->page;
1615				if (page_snap_context(page) != ceph_wbc->snapc)
1616					continue;
1617				wait_on_page_writeback(page);
1618			}
1619
1620			folio_batch_release(&ceph_wbc->fbatch);
1621			cond_resched();
1622		}
1623	}
1624}
1625
1626/*
1627 * initiate async writeback
1628 */
1629static int ceph_writepages_start(struct address_space *mapping,
1630				 struct writeback_control *wbc)
1631{
1632	struct inode *inode = mapping->host;
1633	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1634	struct ceph_client *cl = fsc->client;
1635	struct ceph_writeback_ctl ceph_wbc;
1636	int rc = 0;
1637
1638	if (wbc->sync_mode == WB_SYNC_NONE && fsc->write_congested)
1639		return 0;
1640
1641	doutc(cl, "%llx.%llx (mode=%s)\n", ceph_vinop(inode),
1642	      wbc->sync_mode == WB_SYNC_NONE ? "NONE" :
1643	      (wbc->sync_mode == WB_SYNC_ALL ? "ALL" : "HOLD"));
1644
1645	if (is_forced_umount(mapping)) {
1646		/* we're in a forced umount, don't write! */
1647		return -EIO;
1648	}
1649
1650	ceph_init_writeback_ctl(mapping, wbc, &ceph_wbc);
1651
1652	if (!ceph_inc_osd_stopping_blocker(fsc->mdsc)) {
1653		rc = -EIO;
1654		goto out;
1655	}
1656
1657retry:
1658	rc = ceph_define_writeback_range(mapping, wbc, &ceph_wbc);
1659	if (rc == -ENODATA) {
1660		/* hmm, why does writepages get called when there
1661		   is no dirty data? */
1662		rc = 0;
1663		goto dec_osd_stopping_blocker;
1664	}
1665
1666	if (wbc->sync_mode == WB_SYNC_ALL || wbc->tagged_writepages)
1667		tag_pages_for_writeback(mapping, ceph_wbc.index, ceph_wbc.end);
1668
1669	while (!has_writeback_done(&ceph_wbc)) {
1670		ceph_wbc.locked_pages = 0;
1671		ceph_wbc.max_pages = ceph_wbc.wsize >> PAGE_SHIFT;
1672
1673get_more_pages:
1674		ceph_folio_batch_reinit(&ceph_wbc);
1675
1676		ceph_wbc.nr_folios = filemap_get_folios_tag(mapping,
1677							    &ceph_wbc.index,
1678							    ceph_wbc.end,
1679							    ceph_wbc.tag,
1680							    &ceph_wbc.fbatch);
1681		doutc(cl, "pagevec_lookup_range_tag for tag %#x got %d\n",
1682			ceph_wbc.tag, ceph_wbc.nr_folios);
1683
1684		if (!ceph_wbc.nr_folios && !ceph_wbc.locked_pages)
1685			break;
1686
1687process_folio_batch:
1688		rc = ceph_process_folio_batch(mapping, wbc, &ceph_wbc);
1689		ceph_shift_unused_folios_left(&ceph_wbc.fbatch);
1690		if (rc)
1691			goto release_folios;
1692
1693		/* did we get anything? */
1694		if (!ceph_wbc.locked_pages)
1695			goto release_folios;
1696
1697		if (ceph_wbc.processed_in_fbatch) {
1698			if (folio_batch_count(&ceph_wbc.fbatch) == 0 &&
1699			    ceph_wbc.locked_pages < ceph_wbc.max_pages) {
1700				doutc(cl, "reached end fbatch, trying for more\n");
1701				goto get_more_pages;
1702			}
1703		}
1704
1705		rc = ceph_submit_write(mapping, wbc, &ceph_wbc);
1706		if (rc)
1707			goto release_folios;
1708
1709		ceph_wbc.locked_pages = 0;
1710		ceph_wbc.strip_unit_end = 0;
1711
1712		if (folio_batch_count(&ceph_wbc.fbatch) > 0) {
1713			ceph_wbc.nr_folios =
1714				folio_batch_count(&ceph_wbc.fbatch);
1715			goto process_folio_batch;
1716		}
1717
1718		/*
1719		 * We stop writing back only if we are not doing
1720		 * integrity sync. In case of integrity sync we have to
1721		 * keep going until we have written all the pages
1722		 * we tagged for writeback prior to entering this loop.
1723		 */
1724		if (wbc->nr_to_write <= 0 && wbc->sync_mode == WB_SYNC_NONE)
1725			ceph_wbc.done = true;
1726
1727release_folios:
1728		doutc(cl, "folio_batch release on %d folios (%p)\n",
1729		      (int)ceph_wbc.fbatch.nr,
1730		      ceph_wbc.fbatch.nr ? ceph_wbc.fbatch.folios[0] : NULL);
1731		folio_batch_release(&ceph_wbc.fbatch);
1732	}
1733
1734	if (ceph_wbc.should_loop && !ceph_wbc.done) {
1735		/* more to do; loop back to beginning of file */
1736		doutc(cl, "looping back to beginning of file\n");
1737		/* OK even when start_index == 0 */
1738		ceph_wbc.end = ceph_wbc.start_index - 1;
1739
1740		/* to write dirty pages associated with next snapc,
1741		 * we need to wait until current writes complete */
1742		ceph_wait_until_current_writes_complete(mapping, wbc, &ceph_wbc);
1743
1744		ceph_wbc.start_index = 0;
1745		ceph_wbc.index = 0;
1746		goto retry;
1747	}
1748
1749	if (wbc->range_cyclic || (ceph_wbc.range_whole && wbc->nr_to_write > 0))
1750		mapping->writeback_index = ceph_wbc.index;
1751
1752dec_osd_stopping_blocker:
1753	ceph_dec_osd_stopping_blocker(fsc->mdsc);
1754
1755out:
1756	ceph_put_snap_context(ceph_wbc.last_snapc);
1757	doutc(cl, "%llx.%llx dend - startone, rc = %d\n", ceph_vinop(inode),
1758	      rc);
1759
1760	return rc;
1761}
1762
1763/*
1764 * See if a given @snapc is either writeable, or already written.
1765 */
1766static int context_is_writeable_or_written(struct inode *inode,
1767					   struct ceph_snap_context *snapc)
1768{
1769	struct ceph_snap_context *oldest = get_oldest_context(inode, NULL, NULL);
1770	int ret = !oldest || snapc->seq <= oldest->seq;
1771
1772	ceph_put_snap_context(oldest);
1773	return ret;
1774}
1775
1776/**
1777 * ceph_find_incompatible - find an incompatible context and return it
1778 * @folio: folio being dirtied
1779 *
1780 * We are only allowed to write into/dirty a folio if the folio is
1781 * clean, or already dirty within the same snap context. Returns a
1782 * conflicting context if there is one, NULL if there isn't, or a
1783 * negative error code on other errors.
1784 *
1785 * Must be called with folio lock held.
1786 */
1787static struct ceph_snap_context *
1788ceph_find_incompatible(struct folio *folio)
1789{
1790	struct inode *inode = folio->mapping->host;
1791	struct ceph_client *cl = ceph_inode_to_client(inode);
1792	struct ceph_inode_info *ci = ceph_inode(inode);
1793
1794	if (ceph_inode_is_shutdown(inode)) {
1795		doutc(cl, " %llx.%llx folio %p is shutdown\n",
1796		      ceph_vinop(inode), folio);
1797		return ERR_PTR(-ESTALE);
1798	}
1799
1800	for (;;) {
1801		struct ceph_snap_context *snapc, *oldest;
1802
1803		folio_wait_writeback(folio);
1804
1805		snapc = page_snap_context(&folio->page);
1806		if (!snapc || snapc == ci->i_head_snapc)
1807			break;
1808
1809		/*
1810		 * this folio is already dirty in another (older) snap
1811		 * context!  is it writeable now?
1812		 */
1813		oldest = get_oldest_context(inode, NULL, NULL);
1814		if (snapc->seq > oldest->seq) {
1815			/* not writeable -- return it for the caller to deal with */
1816			ceph_put_snap_context(oldest);
1817			doutc(cl, " %llx.%llx folio %p snapc %p not current or oldest\n",
1818			      ceph_vinop(inode), folio, snapc);
1819			return ceph_get_snap_context(snapc);
1820		}
1821		ceph_put_snap_context(oldest);
1822
1823		/* yay, writeable, do it now (without dropping folio lock) */
1824		doutc(cl, " %llx.%llx folio %p snapc %p not current, but oldest\n",
1825		      ceph_vinop(inode), folio, snapc);
1826		if (folio_clear_dirty_for_io(folio)) {
1827			int r = write_folio_nounlock(folio, NULL);
1828			if (r < 0)
1829				return ERR_PTR(r);
1830		}
1831	}
1832	return NULL;
1833}
1834
1835static int ceph_netfs_check_write_begin(struct file *file, loff_t pos, unsigned int len,
1836					struct folio **foliop, void **_fsdata)
1837{
1838	struct inode *inode = file_inode(file);
1839	struct ceph_inode_info *ci = ceph_inode(inode);
1840	struct ceph_snap_context *snapc;
1841
1842	snapc = ceph_find_incompatible(*foliop);
1843	if (snapc) {
1844		int r;
1845
1846		folio_unlock(*foliop);
1847		folio_put(*foliop);
1848		*foliop = NULL;
1849		if (IS_ERR(snapc))
1850			return PTR_ERR(snapc);
1851
1852		ceph_queue_writeback(inode);
1853		r = wait_event_killable(ci->i_cap_wq,
1854					context_is_writeable_or_written(inode, snapc));
1855		ceph_put_snap_context(snapc);
1856		return r == 0 ? -EAGAIN : r;
1857	}
1858	return 0;
1859}
1860
1861/*
1862 * We are only allowed to write into/dirty the page if the page is
1863 * clean, or already dirty within the same snap context.
1864 */
1865static int ceph_write_begin(const struct kiocb *iocb,
1866			    struct address_space *mapping,
1867			    loff_t pos, unsigned len,
1868			    struct folio **foliop, void **fsdata)
1869{
1870	struct file *file = iocb->ki_filp;
1871	struct inode *inode = file_inode(file);
1872	struct ceph_inode_info *ci = ceph_inode(inode);
1873	int r;
1874
1875	r = netfs_write_begin(&ci->netfs, file, inode->i_mapping, pos, len, foliop, NULL);
1876	if (r < 0)
1877		return r;
1878
1879	folio_wait_private_2(*foliop); /* [DEPRECATED] */
1880	WARN_ON_ONCE(!folio_test_locked(*foliop));
1881	return 0;
1882}
1883
1884/*
1885 * we don't do anything in here that simple_write_end doesn't do
1886 * except adjust dirty page accounting
1887 */
1888static int ceph_write_end(const struct kiocb *iocb,
1889			  struct address_space *mapping, loff_t pos,
1890			  unsigned len, unsigned copied,
1891			  struct folio *folio, void *fsdata)
1892{
1893	struct file *file = iocb->ki_filp;
1894	struct inode *inode = file_inode(file);
1895	struct ceph_client *cl = ceph_inode_to_client(inode);
1896	bool check_cap = false;
1897
1898	doutc(cl, "%llx.%llx file %p folio %p %d~%d (%d)\n", ceph_vinop(inode),
1899	      file, folio, (int)pos, (int)copied, (int)len);
1900
1901	if (!folio_test_uptodate(folio)) {
1902		/* just return that nothing was copied on a short copy */
1903		if (copied < len) {
1904			copied = 0;
1905			goto out;
1906		}
1907		folio_mark_uptodate(folio);
1908	}
1909
1910	/* did file size increase? */
1911	if (pos+copied > i_size_read(inode))
1912		check_cap = ceph_inode_set_size(inode, pos+copied);
1913
1914	folio_mark_dirty(folio);
1915
1916out:
1917	folio_unlock(folio);
1918	folio_put(folio);
1919
1920	if (check_cap)
1921		ceph_check_caps(ceph_inode(inode), CHECK_CAPS_AUTHONLY);
1922
1923	return copied;
1924}
1925
1926const struct address_space_operations ceph_aops = {
1927	.read_folio = netfs_read_folio,
1928	.readahead = netfs_readahead,
1929	.writepages = ceph_writepages_start,
1930	.write_begin = ceph_write_begin,
1931	.write_end = ceph_write_end,
1932	.dirty_folio = ceph_dirty_folio,
1933	.invalidate_folio = ceph_invalidate_folio,
1934	.release_folio = netfs_release_folio,
1935	.direct_IO = noop_direct_IO,
1936	.migrate_folio = filemap_migrate_folio,
1937};
1938
1939static void ceph_block_sigs(sigset_t *oldset)
1940{
1941	sigset_t mask;
1942	siginitsetinv(&mask, sigmask(SIGKILL));
1943	sigprocmask(SIG_BLOCK, &mask, oldset);
1944}
1945
1946static void ceph_restore_sigs(sigset_t *oldset)
1947{
1948	sigprocmask(SIG_SETMASK, oldset, NULL);
1949}
1950
1951/*
1952 * vm ops
1953 */
1954static vm_fault_t ceph_filemap_fault(struct vm_fault *vmf)
1955{
1956	struct vm_area_struct *vma = vmf->vma;
1957	struct inode *inode = file_inode(vma->vm_file);
1958	struct ceph_inode_info *ci = ceph_inode(inode);
1959	struct ceph_client *cl = ceph_inode_to_client(inode);
1960	struct ceph_file_info *fi = vma->vm_file->private_data;
1961	loff_t off = (loff_t)vmf->pgoff << PAGE_SHIFT;
1962	int want, got, err;
1963	sigset_t oldset;
1964	vm_fault_t ret = VM_FAULT_SIGBUS;
1965
1966	if (ceph_inode_is_shutdown(inode))
1967		return ret;
1968
1969	ceph_block_sigs(&oldset);
1970
1971	doutc(cl, "%llx.%llx %llu trying to get caps\n",
1972	      ceph_vinop(inode), off);
1973	if (fi->fmode & CEPH_FILE_MODE_LAZY)
1974		want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
1975	else
1976		want = CEPH_CAP_FILE_CACHE;
1977
1978	got = 0;
1979	err = ceph_get_caps(vma->vm_file, CEPH_CAP_FILE_RD, want, -1, &got);
1980	if (err < 0)
1981		goto out_restore;
1982
1983	doutc(cl, "%llx.%llx %llu got cap refs on %s\n", ceph_vinop(inode),
1984	      off, ceph_cap_string(got));
1985
1986	if ((got & (CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO)) ||
1987	    !ceph_has_inline_data(ci)) {
1988		CEPH_DEFINE_RW_CONTEXT(rw_ctx, got);
1989		ceph_add_rw_context(fi, &rw_ctx);
1990		ret = filemap_fault(vmf);
1991		ceph_del_rw_context(fi, &rw_ctx);
1992		doutc(cl, "%llx.%llx %llu drop cap refs %s ret %x\n",
1993		      ceph_vinop(inode), off, ceph_cap_string(got), ret);
1994	} else
1995		err = -EAGAIN;
1996
1997	ceph_put_cap_refs(ci, got);
1998
1999	if (err != -EAGAIN)
2000		goto out_restore;
2001
2002	/* read inline data */
2003	if (off >= PAGE_SIZE) {
2004		/* does not support inline data > PAGE_SIZE */
2005		ret = VM_FAULT_SIGBUS;
2006	} else {
2007		struct address_space *mapping = inode->i_mapping;
2008		struct page *page;
2009
2010		filemap_invalidate_lock_shared(mapping);
2011		page = find_or_create_page(mapping, 0,
2012				mapping_gfp_constraint(mapping, ~__GFP_FS));
2013		if (!page) {
2014			ret = VM_FAULT_OOM;
2015			goto out_inline;
2016		}
2017		err = __ceph_do_getattr(inode, page,
2018					 CEPH_STAT_CAP_INLINE_DATA, true);
2019		if (err < 0 || off >= i_size_read(inode)) {
2020			unlock_page(page);
2021			put_page(page);
2022			ret = vmf_error(err);
2023			goto out_inline;
2024		}
2025		if (err < PAGE_SIZE)
2026			zero_user_segment(page, err, PAGE_SIZE);
2027		else
2028			flush_dcache_page(page);
2029		SetPageUptodate(page);
2030		vmf->page = page;
2031		ret = VM_FAULT_MAJOR | VM_FAULT_LOCKED;
2032out_inline:
2033		filemap_invalidate_unlock_shared(mapping);
2034		doutc(cl, "%llx.%llx %llu read inline data ret %x\n",
2035		      ceph_vinop(inode), off, ret);
2036	}
2037out_restore:
2038	ceph_restore_sigs(&oldset);
2039	if (err < 0)
2040		ret = vmf_error(err);
2041
2042	return ret;
2043}
2044
2045static vm_fault_t ceph_page_mkwrite(struct vm_fault *vmf)
2046{
2047	struct vm_area_struct *vma = vmf->vma;
2048	struct inode *inode = file_inode(vma->vm_file);
2049	struct ceph_client *cl = ceph_inode_to_client(inode);
2050	struct ceph_inode_info *ci = ceph_inode(inode);
2051	struct ceph_file_info *fi = vma->vm_file->private_data;
2052	struct ceph_cap_flush *prealloc_cf;
2053	struct folio *folio = page_folio(vmf->page);
2054	loff_t off = folio_pos(folio);
2055	loff_t size = i_size_read(inode);
2056	size_t len;
2057	int want, got, err;
2058	sigset_t oldset;
2059	vm_fault_t ret = VM_FAULT_SIGBUS;
2060
2061	if (ceph_inode_is_shutdown(inode))
2062		return ret;
2063
2064	prealloc_cf = ceph_alloc_cap_flush();
2065	if (!prealloc_cf)
2066		return VM_FAULT_OOM;
2067
2068	sb_start_pagefault(inode->i_sb);
2069	ceph_block_sigs(&oldset);
2070
2071	if (off + folio_size(folio) <= size)
2072		len = folio_size(folio);
2073	else
2074		len = offset_in_folio(folio, size);
2075
2076	doutc(cl, "%llx.%llx %llu~%zd getting caps i_size %llu\n",
2077	      ceph_vinop(inode), off, len, size);
2078	if (fi->fmode & CEPH_FILE_MODE_LAZY)
2079		want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
2080	else
2081		want = CEPH_CAP_FILE_BUFFER;
2082
2083	got = 0;
2084	err = ceph_get_caps(vma->vm_file, CEPH_CAP_FILE_WR, want, off + len, &got);
2085	if (err < 0)
2086		goto out_free;
2087
2088	doutc(cl, "%llx.%llx %llu~%zd got cap refs on %s\n", ceph_vinop(inode),
2089	      off, len, ceph_cap_string(got));
2090
2091	/* Update time before taking folio lock */
2092	file_update_time(vma->vm_file);
2093	inode_inc_iversion_raw(inode);
2094
2095	do {
2096		struct ceph_snap_context *snapc;
2097
2098		folio_lock(folio);
2099
2100		if (folio_mkwrite_check_truncate(folio, inode) < 0) {
2101			folio_unlock(folio);
2102			ret = VM_FAULT_NOPAGE;
2103			break;
2104		}
2105
2106		snapc = ceph_find_incompatible(folio);
2107		if (!snapc) {
2108			/* success.  we'll keep the folio locked. */
2109			folio_mark_dirty(folio);
2110			ret = VM_FAULT_LOCKED;
2111			break;
2112		}
2113
2114		folio_unlock(folio);
2115
2116		if (IS_ERR(snapc)) {
2117			ret = VM_FAULT_SIGBUS;
2118			break;
2119		}
2120
2121		ceph_queue_writeback(inode);
2122		err = wait_event_killable(ci->i_cap_wq,
2123				context_is_writeable_or_written(inode, snapc));
2124		ceph_put_snap_context(snapc);
2125	} while (err == 0);
2126
2127	if (ret == VM_FAULT_LOCKED) {
2128		int dirty;
2129		spin_lock(&ci->i_ceph_lock);
2130		dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
2131					       &prealloc_cf);
2132		spin_unlock(&ci->i_ceph_lock);
2133		if (dirty)
2134			__mark_inode_dirty(inode, dirty);
2135	}
2136
2137	doutc(cl, "%llx.%llx %llu~%zd dropping cap refs on %s ret %x\n",
2138	      ceph_vinop(inode), off, len, ceph_cap_string(got), ret);
2139	ceph_put_cap_refs_async(ci, got);
2140out_free:
2141	ceph_restore_sigs(&oldset);
2142	sb_end_pagefault(inode->i_sb);
2143	ceph_free_cap_flush(prealloc_cf);
2144	if (err < 0)
2145		ret = vmf_error(err);
2146	return ret;
2147}
2148
2149void ceph_fill_inline_data(struct inode *inode, struct page *locked_page,
2150			   char	*data, size_t len)
2151{
2152	struct ceph_client *cl = ceph_inode_to_client(inode);
2153	struct address_space *mapping = inode->i_mapping;
2154	struct page *page;
2155
2156	if (locked_page) {
2157		page = locked_page;
2158	} else {
2159		if (i_size_read(inode) == 0)
2160			return;
2161		page = find_or_create_page(mapping, 0,
2162					   mapping_gfp_constraint(mapping,
2163					   ~__GFP_FS));
2164		if (!page)
2165			return;
2166		if (PageUptodate(page)) {
2167			unlock_page(page);
2168			put_page(page);
2169			return;
2170		}
2171	}
2172
2173	doutc(cl, "%p %llx.%llx len %zu locked_page %p\n", inode,
2174	      ceph_vinop(inode), len, locked_page);
2175
2176	if (len > 0) {
2177		void *kaddr = kmap_atomic(page);
2178		memcpy(kaddr, data, len);
2179		kunmap_atomic(kaddr);
2180	}
2181
2182	if (page != locked_page) {
2183		if (len < PAGE_SIZE)
2184			zero_user_segment(page, len, PAGE_SIZE);
2185		else
2186			flush_dcache_page(page);
2187
2188		SetPageUptodate(page);
2189		unlock_page(page);
2190		put_page(page);
2191	}
2192}
2193
2194int ceph_uninline_data(struct file *file)
2195{
2196	struct inode *inode = file_inode(file);
2197	struct ceph_inode_info *ci = ceph_inode(inode);
2198	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
2199	struct ceph_client *cl = fsc->client;
2200	struct ceph_osd_request *req = NULL;
2201	struct ceph_cap_flush *prealloc_cf = NULL;
2202	struct folio *folio = NULL;
2203	struct ceph_snap_context *snapc = NULL;
2204	u64 inline_version = CEPH_INLINE_NONE;
2205	struct page *pages[1];
2206	int err = 0;
2207	u64 len;
2208
2209	spin_lock(&ci->i_ceph_lock);
2210	inline_version = ci->i_inline_version;
2211	spin_unlock(&ci->i_ceph_lock);
2212
2213	doutc(cl, "%llx.%llx inline_version %llu\n", ceph_vinop(inode),
2214	      inline_version);
2215
2216	if (ceph_inode_is_shutdown(inode)) {
2217		err = -EIO;
2218		goto out;
2219	}
2220
2221	if (inline_version == CEPH_INLINE_NONE)
2222		return 0;
2223
2224	prealloc_cf = ceph_alloc_cap_flush();
2225	if (!prealloc_cf)
2226		return -ENOMEM;
2227
2228	if (inline_version == 1) /* initial version, no data */
2229		goto out_uninline;
2230
2231	down_read(&fsc->mdsc->snap_rwsem);
2232	spin_lock(&ci->i_ceph_lock);
2233	if (__ceph_have_pending_cap_snap(ci)) {
2234		struct ceph_cap_snap *capsnap =
2235				list_last_entry(&ci->i_cap_snaps,
2236						struct ceph_cap_snap,
2237						ci_item);
2238		snapc = ceph_get_snap_context(capsnap->context);
2239	} else {
2240		if (!ci->i_head_snapc) {
2241			ci->i_head_snapc = ceph_get_snap_context(
2242				ci->i_snap_realm->cached_context);
2243		}
2244		snapc = ceph_get_snap_context(ci->i_head_snapc);
2245	}
2246	spin_unlock(&ci->i_ceph_lock);
2247	up_read(&fsc->mdsc->snap_rwsem);
2248
2249	folio = read_mapping_folio(inode->i_mapping, 0, file);
2250	if (IS_ERR(folio)) {
2251		err = PTR_ERR(folio);
2252		goto out;
2253	}
2254
2255	folio_lock(folio);
2256
2257	len = i_size_read(inode);
2258	if (len > folio_size(folio))
2259		len = folio_size(folio);
2260
2261	req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
2262				    ceph_vino(inode), 0, &len, 0, 1,
2263				    CEPH_OSD_OP_CREATE, CEPH_OSD_FLAG_WRITE,
2264				    snapc, 0, 0, false);
2265	if (IS_ERR(req)) {
2266		err = PTR_ERR(req);
2267		goto out_unlock;
2268	}
2269
2270	req->r_mtime = inode_get_mtime(inode);
2271	ceph_osdc_start_request(&fsc->client->osdc, req);
2272	err = ceph_osdc_wait_request(&fsc->client->osdc, req);
2273	ceph_osdc_put_request(req);
2274	if (err < 0)
2275		goto out_unlock;
2276
2277	req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
2278				    ceph_vino(inode), 0, &len, 1, 3,
2279				    CEPH_OSD_OP_WRITE, CEPH_OSD_FLAG_WRITE,
2280				    snapc, ci->i_truncate_seq,
2281				    ci->i_truncate_size, false);
2282	if (IS_ERR(req)) {
2283		err = PTR_ERR(req);
2284		goto out_unlock;
2285	}
2286
2287	pages[0] = folio_page(folio, 0);
2288	osd_req_op_extent_osd_data_pages(req, 1, pages, len, 0, false, false);
2289
2290	{
2291		__le64 xattr_buf = cpu_to_le64(inline_version);
2292		err = osd_req_op_xattr_init(req, 0, CEPH_OSD_OP_CMPXATTR,
2293					    "inline_version", &xattr_buf,
2294					    sizeof(xattr_buf),
2295					    CEPH_OSD_CMPXATTR_OP_GT,
2296					    CEPH_OSD_CMPXATTR_MODE_U64);
2297		if (err)
2298			goto out_put_req;
2299	}
2300
2301	{
2302		char xattr_buf[32];
2303		int xattr_len = snprintf(xattr_buf, sizeof(xattr_buf),
2304					 "%llu", inline_version);
2305		err = osd_req_op_xattr_init(req, 2, CEPH_OSD_OP_SETXATTR,
2306					    "inline_version",
2307					    xattr_buf, xattr_len, 0, 0);
2308		if (err)
2309			goto out_put_req;
2310	}
2311
2312	req->r_mtime = inode_get_mtime(inode);
2313	ceph_osdc_start_request(&fsc->client->osdc, req);
2314	err = ceph_osdc_wait_request(&fsc->client->osdc, req);
2315
2316	ceph_update_write_metrics(&fsc->mdsc->metric, req->r_start_latency,
2317				  req->r_end_latency, len, err);
2318
2319out_uninline:
2320	if (!err) {
2321		int dirty;
2322
2323		/* Set to CAP_INLINE_NONE and dirty the caps */
2324		down_read(&fsc->mdsc->snap_rwsem);
2325		spin_lock(&ci->i_ceph_lock);
2326		ci->i_inline_version = CEPH_INLINE_NONE;
2327		dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR, &prealloc_cf);
2328		spin_unlock(&ci->i_ceph_lock);
2329		up_read(&fsc->mdsc->snap_rwsem);
2330		if (dirty)
2331			__mark_inode_dirty(inode, dirty);
2332	}
2333out_put_req:
2334	ceph_osdc_put_request(req);
2335	if (err == -ECANCELED)
2336		err = 0;
2337out_unlock:
2338	if (folio) {
2339		folio_unlock(folio);
2340		folio_put(folio);
2341	}
2342out:
2343	ceph_put_snap_context(snapc);
2344	ceph_free_cap_flush(prealloc_cf);
2345	doutc(cl, "%llx.%llx inline_version %llu = %d\n",
2346	      ceph_vinop(inode), inline_version, err);
2347	return err;
2348}
2349
2350static const struct vm_operations_struct ceph_vmops = {
2351	.fault		= ceph_filemap_fault,
2352	.page_mkwrite	= ceph_page_mkwrite,
2353};
2354
2355int ceph_mmap_prepare(struct vm_area_desc *desc)
2356{
2357	struct address_space *mapping = desc->file->f_mapping;
2358
2359	if (!mapping->a_ops->read_folio)
2360		return -ENOEXEC;
2361	desc->vm_ops = &ceph_vmops;
2362	return 0;
2363}
2364
2365enum {
2366	POOL_READ	= 1,
2367	POOL_WRITE	= 2,
2368};
2369
2370static int __ceph_pool_perm_get(struct ceph_inode_info *ci,
2371				s64 pool, struct ceph_string *pool_ns)
2372{
2373	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(&ci->netfs.inode);
2374	struct ceph_mds_client *mdsc = fsc->mdsc;
2375	struct ceph_client *cl = fsc->client;
2376	struct ceph_osd_request *rd_req = NULL, *wr_req = NULL;
2377	struct rb_node **p, *parent;
2378	struct ceph_pool_perm *perm;
2379	struct page **pages;
2380	size_t pool_ns_len;
2381	int err = 0, err2 = 0, have = 0;
2382
2383	down_read(&mdsc->pool_perm_rwsem);
2384	p = &mdsc->pool_perm_tree.rb_node;
2385	while (*p) {
2386		perm = rb_entry(*p, struct ceph_pool_perm, node);
2387		if (pool < perm->pool)
2388			p = &(*p)->rb_left;
2389		else if (pool > perm->pool)
2390			p = &(*p)->rb_right;
2391		else {
2392			int ret = ceph_compare_string(pool_ns,
2393						perm->pool_ns,
2394						perm->pool_ns_len);
2395			if (ret < 0)
2396				p = &(*p)->rb_left;
2397			else if (ret > 0)
2398				p = &(*p)->rb_right;
2399			else {
2400				have = perm->perm;
2401				break;
2402			}
2403		}
2404	}
2405	up_read(&mdsc->pool_perm_rwsem);
2406	if (*p)
2407		goto out;
2408
2409	if (pool_ns)
2410		doutc(cl, "pool %lld ns %.*s no perm cached\n", pool,
2411		      (int)pool_ns->len, pool_ns->str);
2412	else
2413		doutc(cl, "pool %lld no perm cached\n", pool);
2414
2415	down_write(&mdsc->pool_perm_rwsem);
2416	p = &mdsc->pool_perm_tree.rb_node;
2417	parent = NULL;
2418	while (*p) {
2419		parent = *p;
2420		perm = rb_entry(parent, struct ceph_pool_perm, node);
2421		if (pool < perm->pool)
2422			p = &(*p)->rb_left;
2423		else if (pool > perm->pool)
2424			p = &(*p)->rb_right;
2425		else {
2426			int ret = ceph_compare_string(pool_ns,
2427						perm->pool_ns,
2428						perm->pool_ns_len);
2429			if (ret < 0)
2430				p = &(*p)->rb_left;
2431			else if (ret > 0)
2432				p = &(*p)->rb_right;
2433			else {
2434				have = perm->perm;
2435				break;
2436			}
2437		}
2438	}
2439	if (*p) {
2440		up_write(&mdsc->pool_perm_rwsem);
2441		goto out;
2442	}
2443
2444	rd_req = ceph_osdc_alloc_request(&fsc->client->osdc, NULL,
2445					 1, false, GFP_NOFS);
2446	if (!rd_req) {
2447		err = -ENOMEM;
2448		goto out_unlock;
2449	}
2450
2451	rd_req->r_flags = CEPH_OSD_FLAG_READ;
2452	osd_req_op_init(rd_req, 0, CEPH_OSD_OP_STAT, 0);
2453	rd_req->r_base_oloc.pool = pool;
2454	if (pool_ns)
2455		rd_req->r_base_oloc.pool_ns = ceph_get_string(pool_ns);
2456	ceph_oid_printf(&rd_req->r_base_oid, "%llx.00000000", ci->i_vino.ino);
2457
2458	err = ceph_osdc_alloc_messages(rd_req, GFP_NOFS);
2459	if (err)
2460		goto out_unlock;
2461
2462	wr_req = ceph_osdc_alloc_request(&fsc->client->osdc, NULL,
2463					 1, false, GFP_NOFS);
2464	if (!wr_req) {
2465		err = -ENOMEM;
2466		goto out_unlock;
2467	}
2468
2469	wr_req->r_flags = CEPH_OSD_FLAG_WRITE;
2470	osd_req_op_init(wr_req, 0, CEPH_OSD_OP_CREATE, CEPH_OSD_OP_FLAG_EXCL);
2471	ceph_oloc_copy(&wr_req->r_base_oloc, &rd_req->r_base_oloc);
2472	ceph_oid_copy(&wr_req->r_base_oid, &rd_req->r_base_oid);
2473
2474	err = ceph_osdc_alloc_messages(wr_req, GFP_NOFS);
2475	if (err)
2476		goto out_unlock;
2477
2478	/* one page should be large enough for STAT data */
2479	pages = ceph_alloc_page_vector(1, GFP_KERNEL);
2480	if (IS_ERR(pages)) {
2481		err = PTR_ERR(pages);
2482		goto out_unlock;
2483	}
2484
2485	osd_req_op_raw_data_in_pages(rd_req, 0, pages, PAGE_SIZE,
2486				     0, false, true);
2487	ceph_osdc_start_request(&fsc->client->osdc, rd_req);
2488
2489	wr_req->r_mtime = inode_get_mtime(&ci->netfs.inode);
2490	ceph_osdc_start_request(&fsc->client->osdc, wr_req);
2491
2492	err = ceph_osdc_wait_request(&fsc->client->osdc, rd_req);
2493	err2 = ceph_osdc_wait_request(&fsc->client->osdc, wr_req);
2494
2495	if (err >= 0 || err == -ENOENT)
2496		have |= POOL_READ;
2497	else if (err != -EPERM) {
2498		if (err == -EBLOCKLISTED)
2499			fsc->blocklisted = true;
2500		goto out_unlock;
2501	}
2502
2503	if (err2 == 0 || err2 == -EEXIST)
2504		have |= POOL_WRITE;
2505	else if (err2 != -EPERM) {
2506		if (err2 == -EBLOCKLISTED)
2507			fsc->blocklisted = true;
2508		err = err2;
2509		goto out_unlock;
2510	}
2511
2512	pool_ns_len = pool_ns ? pool_ns->len : 0;
2513	perm = kmalloc(struct_size(perm, pool_ns, pool_ns_len + 1), GFP_NOFS);
2514	if (!perm) {
2515		err = -ENOMEM;
2516		goto out_unlock;
2517	}
2518
2519	perm->pool = pool;
2520	perm->perm = have;
2521	perm->pool_ns_len = pool_ns_len;
2522	if (pool_ns_len > 0)
2523		memcpy(perm->pool_ns, pool_ns->str, pool_ns_len);
2524	perm->pool_ns[pool_ns_len] = 0;
2525
2526	rb_link_node(&perm->node, parent, p);
2527	rb_insert_color(&perm->node, &mdsc->pool_perm_tree);
2528	err = 0;
2529out_unlock:
2530	up_write(&mdsc->pool_perm_rwsem);
2531
2532	ceph_osdc_put_request(rd_req);
2533	ceph_osdc_put_request(wr_req);
2534out:
2535	if (!err)
2536		err = have;
2537	if (pool_ns)
2538		doutc(cl, "pool %lld ns %.*s result = %d\n", pool,
2539		      (int)pool_ns->len, pool_ns->str, err);
2540	else
2541		doutc(cl, "pool %lld result = %d\n", pool, err);
2542	return err;
2543}
2544
2545int ceph_pool_perm_check(struct inode *inode, int need)
2546{
2547	struct ceph_client *cl = ceph_inode_to_client(inode);
2548	struct ceph_inode_info *ci = ceph_inode(inode);
2549	struct ceph_string *pool_ns;
2550	s64 pool;
2551	int ret, flags;
2552
2553	/* Only need to do this for regular files */
2554	if (!S_ISREG(inode->i_mode))
2555		return 0;
2556
2557	if (ci->i_vino.snap != CEPH_NOSNAP) {
2558		/*
2559		 * Pool permission check needs to write to the first object.
2560		 * But for snapshot, head of the first object may have already
2561		 * been deleted. Skip check to avoid creating orphan object.
2562		 */
2563		return 0;
2564	}
2565
2566	if (ceph_test_mount_opt(ceph_inode_to_fs_client(inode),
2567				NOPOOLPERM))
2568		return 0;
2569
2570	spin_lock(&ci->i_ceph_lock);
2571	flags = ci->i_ceph_flags;
2572	pool = ci->i_layout.pool_id;
2573	spin_unlock(&ci->i_ceph_lock);
2574check:
2575	if (flags & CEPH_I_POOL_PERM) {
2576		if ((need & CEPH_CAP_FILE_RD) && !(flags & CEPH_I_POOL_RD)) {
2577			doutc(cl, "pool %lld no read perm\n", pool);
2578			return -EPERM;
2579		}
2580		if ((need & CEPH_CAP_FILE_WR) && !(flags & CEPH_I_POOL_WR)) {
2581			doutc(cl, "pool %lld no write perm\n", pool);
2582			return -EPERM;
2583		}
2584		return 0;
2585	}
2586
2587	pool_ns = ceph_try_get_string(ci->i_layout.pool_ns);
2588	ret = __ceph_pool_perm_get(ci, pool, pool_ns);
2589	ceph_put_string(pool_ns);
2590	if (ret < 0)
2591		return ret;
2592
2593	flags = CEPH_I_POOL_PERM;
2594	if (ret & POOL_READ)
2595		flags |= CEPH_I_POOL_RD;
2596	if (ret & POOL_WRITE)
2597		flags |= CEPH_I_POOL_WR;
2598
2599	spin_lock(&ci->i_ceph_lock);
2600	if (pool == ci->i_layout.pool_id &&
2601	    pool_ns == rcu_dereference_raw(ci->i_layout.pool_ns)) {
2602		ci->i_ceph_flags |= flags;
2603        } else {
2604		pool = ci->i_layout.pool_id;
2605		flags = ci->i_ceph_flags;
2606	}
2607	spin_unlock(&ci->i_ceph_lock);
2608	goto check;
2609}
2610
2611void ceph_pool_perm_destroy(struct ceph_mds_client *mdsc)
2612{
2613	struct ceph_pool_perm *perm;
2614	struct rb_node *n;
2615
2616	while (!RB_EMPTY_ROOT(&mdsc->pool_perm_tree)) {
2617		n = rb_first(&mdsc->pool_perm_tree);
2618		perm = rb_entry(n, struct ceph_pool_perm, node);
2619		rb_erase(n, &mdsc->pool_perm_tree);
2620		kfree(perm);
2621	}
2622}