Linux Audio

Check our new training course

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