addr.c 53.3 KB
Newer Older
1
// SPDX-License-Identifier: GPL-2.0
2
#include <linux/ceph/ceph_debug.h>
S
Sage Weil 已提交
3 4 5 6 7 8

#include <linux/backing-dev.h>
#include <linux/fs.h>
#include <linux/mm.h>
#include <linux/pagemap.h>
#include <linux/writeback.h>	/* generic_writepages */
9
#include <linux/slab.h>
S
Sage Weil 已提交
10 11
#include <linux/pagevec.h>
#include <linux/task_io_accounting_ops.h>
12
#include <linux/signal.h>
S
Sage Weil 已提交
13 14

#include "super.h"
15
#include "mds_client.h"
16
#include "cache.h"
17
#include <linux/ceph/osd_client.h>
18
#include <linux/ceph/striper.h>
S
Sage Weil 已提交
19 20 21 22 23 24 25 26 27 28 29 30

/*
 * Ceph address space ops.
 *
 * There are a few funny things going on here.
 *
 * The page->private field is used to reference a struct
 * ceph_snap_context for _every_ dirty page.  This indicates which
 * snapshot the page was logically dirtied in, and thus which snap
 * context needs to be associated with the osd write during writeback.
 *
 * Similarly, struct ceph_inode_info maintains a set of counters to
L
Lucas De Marchi 已提交
31
 * count dirty pages on the inode.  In the absence of snapshots,
S
Sage Weil 已提交
32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55
 * i_wrbuffer_ref == i_wrbuffer_ref_head == the dirty page count.
 *
 * When a snapshot is taken (that is, when the client receives
 * notification that a snapshot was taken), each inode with caps and
 * with dirty pages (dirty pages implies there is a cap) gets a new
 * ceph_cap_snap in the i_cap_snaps list (which is sorted in ascending
 * order, new snaps go to the tail).  The i_wrbuffer_ref_head count is
 * moved to capsnap->dirty. (Unless a sync write is currently in
 * progress.  In that case, the capsnap is said to be "pending", new
 * writes cannot start, and the capsnap isn't "finalized" until the
 * write completes (or fails) and a final size/mtime for the inode for
 * that snap can be settled upon.)  i_wrbuffer_ref_head is reset to 0.
 *
 * On writeback, we must submit writes to the osd IN SNAP ORDER.  So,
 * we look for the first capsnap in i_cap_snaps and write out pages in
 * that snap context _only_.  Then we move on to the next capsnap,
 * eventually reaching the "live" or "head" context (i.e., pages that
 * are not yet snapped) and are writing the most recently dirtied
 * pages.
 *
 * Invalidate and so forth must take care to ensure the dirty page
 * accounting is preserved.
 */

Y
Yehuda Sadeh 已提交
56 57 58 59 60
#define CONGESTION_ON_THRESH(congestion_kb) (congestion_kb >> (PAGE_SHIFT-10))
#define CONGESTION_OFF_THRESH(congestion_kb)				\
	(CONGESTION_ON_THRESH(congestion_kb) -				\
	 (CONGESTION_ON_THRESH(congestion_kb) >> 2))

61 62 63 64 65 66
static inline struct ceph_snap_context *page_snap_context(struct page *page)
{
	if (PagePrivate(page))
		return (void *)page->private;
	return NULL;
}
S
Sage Weil 已提交
67 68 69 70 71 72 73 74 75 76 77

/*
 * Dirty a page.  Optimistically adjust accounting, on the assumption
 * that we won't race with invalidate.  If we do, readjust.
 */
static int ceph_set_page_dirty(struct page *page)
{
	struct address_space *mapping = page->mapping;
	struct inode *inode;
	struct ceph_inode_info *ci;
	struct ceph_snap_context *snapc;
78
	int ret;
S
Sage Weil 已提交
79 80 81 82

	if (unlikely(!mapping))
		return !TestSetPageDirty(page);

83
	if (PageDirty(page)) {
S
Sage Weil 已提交
84 85
		dout("%p set_page_dirty %p idx %lu -- already dirty\n",
		     mapping->host, page, page->index);
86
		BUG_ON(!PagePrivate(page));
S
Sage Weil 已提交
87 88 89 90 91 92 93
		return 0;
	}

	inode = mapping->host;
	ci = ceph_inode(inode);

	/* dirty the head */
94
	spin_lock(&ci->i_ceph_lock);
95 96 97 98 99 100 101 102 103 104 105 106 107
	BUG_ON(ci->i_wr_ref == 0); // caller should hold Fw reference
	if (__ceph_have_pending_cap_snap(ci)) {
		struct ceph_cap_snap *capsnap =
				list_last_entry(&ci->i_cap_snaps,
						struct ceph_cap_snap,
						ci_item);
		snapc = ceph_get_snap_context(capsnap->context);
		capsnap->dirty_pages++;
	} else {
		BUG_ON(!ci->i_head_snapc);
		snapc = ceph_get_snap_context(ci->i_head_snapc);
		++ci->i_wrbuffer_ref_head;
	}
S
Sage Weil 已提交
108
	if (ci->i_wrbuffer_ref == 0)
109
		ihold(inode);
S
Sage Weil 已提交
110 111 112 113 114 115 116
	++ci->i_wrbuffer_ref;
	dout("%p set_page_dirty %p idx %lu head %d/%d -> %d/%d "
	     "snapc %p seq %lld (%d snaps)\n",
	     mapping->host, page, page->index,
	     ci->i_wrbuffer_ref-1, ci->i_wrbuffer_ref_head-1,
	     ci->i_wrbuffer_ref, ci->i_wrbuffer_ref_head,
	     snapc, snapc->seq, snapc->num_snaps);
117
	spin_unlock(&ci->i_ceph_lock);
S
Sage Weil 已提交
118

119 120 121 122 123 124 125
	/*
	 * Reference snap context in page->private.  Also set
	 * PagePrivate so that we get invalidatepage callback.
	 */
	BUG_ON(PagePrivate(page));
	page->private = (unsigned long)snapc;
	SetPagePrivate(page);
S
Sage Weil 已提交
126

127 128 129
	ret = __set_page_dirty_nobuffers(page);
	WARN_ON(!PageLocked(page));
	WARN_ON(!page->mapping);
S
Sage Weil 已提交
130

131
	return ret;
S
Sage Weil 已提交
132 133 134 135 136 137 138
}

/*
 * If we are truncating the full page (i.e. offset == 0), adjust the
 * dirty page counters appropriately.  Only called if there is private
 * data on the page.
 */
139 140
static void ceph_invalidatepage(struct page *page, unsigned int offset,
				unsigned int length)
S
Sage Weil 已提交
141
{
142
	struct inode *inode;
S
Sage Weil 已提交
143
	struct ceph_inode_info *ci;
144
	struct ceph_snap_context *snapc = page_snap_context(page);
S
Sage Weil 已提交
145

146
	inode = page->mapping->host;
147 148
	ci = ceph_inode(inode);

149
	if (offset != 0 || length != PAGE_SIZE) {
150 151 152 153
		dout("%p invalidatepage %p idx %lu partial dirty page %u~%u\n",
		     inode, page, page->index, offset, length);
		return;
	}
154

155 156
	ceph_invalidate_fscache_page(inode, page);

157
	WARN_ON(!PageLocked(page));
158 159 160
	if (!PagePrivate(page))
		return;

161
	ClearPageChecked(page);
S
Sage Weil 已提交
162

163 164 165 166 167 168 169
	dout("%p invalidatepage %p idx %lu full dirty page\n",
	     inode, page, page->index);

	ceph_put_wrbuffer_cap_refs(ci, 1, snapc);
	ceph_put_snap_context(snapc);
	page->private = 0;
	ClearPagePrivate(page);
S
Sage Weil 已提交
170 171 172 173
}

static int ceph_releasepage(struct page *page, gfp_t g)
{
174 175
	dout("%p releasepage %p idx %lu (%sdirty)\n", page->mapping->host,
	     page, page->index, PageDirty(page) ? "" : "not ");
176 177 178 179 180 181

	/* Can we release the page from the cache? */
	if (!ceph_release_fscache_page(page, g))
		return 0;

	return !PagePrivate(page);
S
Sage Weil 已提交
182 183 184 185 186
}

/*
 * read a single page, without unlocking it.
 */
Y
Yan, Zheng 已提交
187
static int ceph_do_readpage(struct file *filp, struct page *page)
S
Sage Weil 已提交
188
{
A
Al Viro 已提交
189
	struct inode *inode = file_inode(filp);
S
Sage Weil 已提交
190
	struct ceph_inode_info *ci = ceph_inode(inode);
191
	struct ceph_osd_client *osdc =
192
		&ceph_inode_to_client(inode)->client->osdc;
S
Sage Weil 已提交
193
	int err = 0;
Y
Yan, Zheng 已提交
194
	u64 off = page_offset(page);
195
	u64 len = PAGE_SIZE;
S
Sage Weil 已提交
196

Y
Yan, Zheng 已提交
197
	if (off >= i_size_read(inode)) {
198
		zero_user_segment(page, 0, PAGE_SIZE);
Y
Yan, Zheng 已提交
199 200 201
		SetPageUptodate(page);
		return 0;
	}
202

203 204 205 206 207 208 209
	if (ci->i_inline_version != CEPH_INLINE_NONE) {
		/*
		 * Uptodate inline data should have been added
		 * into page cache while getting Fcr caps.
		 */
		if (off == 0)
			return -EINVAL;
210
		zero_user_segment(page, 0, PAGE_SIZE);
211 212 213
		SetPageUptodate(page);
		return 0;
	}
Y
Yan, Zheng 已提交
214 215

	err = ceph_readpage_from_fscache(inode, page);
216
	if (err == 0)
Y
Yan, Zheng 已提交
217
		return -EINPROGRESS;
218

S
Sage Weil 已提交
219 220 221
	dout("readpage inode %p file %p page %p index %lu\n",
	     inode, filp, page, page->index);
	err = ceph_osdc_readpages(osdc, ceph_vino(inode), &ci->i_layout,
Y
Yan, Zheng 已提交
222
				  off, &len,
S
Sage Weil 已提交
223
				  ci->i_truncate_seq, ci->i_truncate_size,
224
				  &page, 1, 0);
S
Sage Weil 已提交
225 226 227 228
	if (err == -ENOENT)
		err = 0;
	if (err < 0) {
		SetPageError(page);
229
		ceph_fscache_readpage_cancel(inode, page);
S
Sage Weil 已提交
230 231
		goto out;
	}
232
	if (err < PAGE_SIZE)
233
		/* zero fill remainder of page */
234
		zero_user_segment(page, err, PAGE_SIZE);
235 236
	else
		flush_dcache_page(page);
S
Sage Weil 已提交
237

238 239
	SetPageUptodate(page);
	ceph_readpage_to_fscache(inode, page);
240

S
Sage Weil 已提交
241 242 243 244 245 246
out:
	return err < 0 ? err : 0;
}

static int ceph_readpage(struct file *filp, struct page *page)
{
Y
Yan, Zheng 已提交
247 248 249 250 251
	int r = ceph_do_readpage(filp, page);
	if (r != -EINPROGRESS)
		unlock_page(page);
	else
		r = 0;
S
Sage Weil 已提交
252 253 254 255
	return r;
}

/*
S
Sage Weil 已提交
256
 * Finish an async read(ahead) op.
S
Sage Weil 已提交
257
 */
258
static void finish_read(struct ceph_osd_request *req)
S
Sage Weil 已提交
259
{
S
Sage Weil 已提交
260
	struct inode *inode = req->r_inode;
261
	struct ceph_osd_data *osd_data;
262 263
	int rc = req->r_result <= 0 ? req->r_result : 0;
	int bytes = req->r_result >= 0 ? req->r_result : 0;
264
	int num_pages;
S
Sage Weil 已提交
265
	int i;
S
Sage Weil 已提交
266

S
Sage Weil 已提交
267 268 269
	dout("finish_read %p req %p rc %d bytes %d\n", inode, req, rc, bytes);

	/* unlock all pages, zeroing any data we didn't read */
270
	osd_data = osd_req_op_extent_osd_data(req, 0);
271 272 273
	BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGES);
	num_pages = calc_pages_for((u64)osd_data->alignment,
					(u64)osd_data->length);
274
	for (i = 0; i < num_pages; i++) {
275
		struct page *page = osd_data->pages[i];
S
Sage Weil 已提交
276

277 278
		if (rc < 0 && rc != -ENOENT) {
			ceph_fscache_readpage_cancel(inode, page);
279
			goto unlock;
280
		}
281
		if (bytes < (int)PAGE_SIZE) {
S
Sage Weil 已提交
282 283
			/* zero (remainder of) page */
			int s = bytes < 0 ? 0 : bytes;
284
			zero_user_segment(page, s, PAGE_SIZE);
S
Sage Weil 已提交
285
		}
S
Sage Weil 已提交
286 287 288 289
 		dout("finish_read %p uptodate %p idx %lu\n", inode, page,
		     page->index);
		flush_dcache_page(page);
		SetPageUptodate(page);
290
		ceph_readpage_to_fscache(inode, page);
291
unlock:
S
Sage Weil 已提交
292
		unlock_page(page);
293 294
		put_page(page);
		bytes -= PAGE_SIZE;
S
Sage Weil 已提交
295
	}
296
	kfree(osd_data->pages);
S
Sage Weil 已提交
297 298 299
}

/*
S
Sage Weil 已提交
300 301
 * start an async read(ahead) operation.  return nr_pages we submitted
 * a read for on success, or negative error code.
S
Sage Weil 已提交
302
 */
303 304
static int start_read(struct inode *inode, struct ceph_rw_context *rw_ctx,
		      struct list_head *page_list, int max)
S
Sage Weil 已提交
305
{
306 307
	struct ceph_osd_client *osdc =
		&ceph_inode_to_client(inode)->client->osdc;
S
Sage Weil 已提交
308 309
	struct ceph_inode_info *ci = ceph_inode(inode);
	struct page *page = list_entry(page_list->prev, struct page, lru);
310
	struct ceph_vino vino;
S
Sage Weil 已提交
311 312
	struct ceph_osd_request *req;
	u64 off;
S
Sage Weil 已提交
313
	u64 len;
S
Sage Weil 已提交
314 315 316 317
	int i;
	struct page **pages;
	pgoff_t next_index;
	int nr_pages = 0;
318 319 320
	int got = 0;
	int ret = 0;

321
	if (!rw_ctx) {
322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343
		/* caller of readpages does not hold buffer and read caps
		 * (fadvise, madvise and readahead cases) */
		int want = CEPH_CAP_FILE_CACHE;
		ret = ceph_try_get_caps(ci, CEPH_CAP_FILE_RD, want, &got);
		if (ret < 0) {
			dout("start_read %p, error getting cap\n", inode);
		} else if (!(got & want)) {
			dout("start_read %p, no cache cap\n", inode);
			ret = 0;
		}
		if (ret <= 0) {
			if (got)
				ceph_put_cap_refs(ci, got);
			while (!list_empty(page_list)) {
				page = list_entry(page_list->prev,
						  struct page, lru);
				list_del(&page->lru);
				put_page(page);
			}
			return ret;
		}
	}
S
Sage Weil 已提交
344

345
	off = (u64) page_offset(page);
S
Sage Weil 已提交
346

S
Sage Weil 已提交
347 348 349 350 351 352 353
	/* count pages */
	next_index = page->index;
	list_for_each_entry_reverse(page, page_list, lru) {
		if (page->index != next_index)
			break;
		nr_pages++;
		next_index++;
354 355
		if (max && nr_pages == max)
			break;
S
Sage Weil 已提交
356
	}
357
	len = nr_pages << PAGE_SHIFT;
S
Sage Weil 已提交
358 359
	dout("start_read %p nr_pages %d is %lld~%lld\n", inode, nr_pages,
	     off, len);
360 361
	vino = ceph_vino(inode);
	req = ceph_osdc_new_request(osdc, &ci->i_layout, vino, off, &len,
362
				    0, 1, CEPH_OSD_OP_READ,
363
				    CEPH_OSD_FLAG_READ, NULL,
S
Sage Weil 已提交
364
				    ci->i_truncate_seq, ci->i_truncate_size,
365
				    false);
366 367 368 369
	if (IS_ERR(req)) {
		ret = PTR_ERR(req);
		goto out;
	}
S
Sage Weil 已提交
370

S
Sage Weil 已提交
371
	/* build page vector */
372
	nr_pages = calc_pages_for(0, len);
373
	pages = kmalloc_array(nr_pages, sizeof(*pages), GFP_KERNEL);
374 375 376 377
	if (!pages) {
		ret = -ENOMEM;
		goto out_put;
	}
S
Sage Weil 已提交
378 379 380
	for (i = 0; i < nr_pages; ++i) {
		page = list_entry(page_list->prev, struct page, lru);
		BUG_ON(PageLocked(page));
S
Sage Weil 已提交
381
		list_del(&page->lru);
382

S
Sage Weil 已提交
383 384 385
 		dout("start_read %p adding %p idx %lu\n", inode, page,
		     page->index);
		if (add_to_page_cache_lru(page, &inode->i_data, page->index,
386
					  GFP_KERNEL)) {
387
			ceph_fscache_uncache_page(inode, page);
388
			put_page(page);
S
Sage Weil 已提交
389
			dout("start_read %p add_to_page_cache failed %p\n",
S
Sage Weil 已提交
390
			     inode, page);
S
Sage Weil 已提交
391
			nr_pages = i;
392 393
			if (nr_pages > 0) {
				len = nr_pages << PAGE_SHIFT;
394
				osd_req_op_extent_update(req, 0, len);
395 396
				break;
			}
S
Sage Weil 已提交
397
			goto out_pages;
S
Sage Weil 已提交
398
		}
S
Sage Weil 已提交
399
		pages[i] = page;
S
Sage Weil 已提交
400
	}
401
	osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0, false, false);
S
Sage Weil 已提交
402 403 404 405 406 407 408 409
	req->r_callback = finish_read;
	req->r_inode = inode;

	dout("start_read %p starting %p %lld~%lld\n", inode, req, off, len);
	ret = ceph_osdc_start_request(osdc, req, false);
	if (ret < 0)
		goto out_pages;
	ceph_osdc_put_request(req);
410 411 412 413 414 415

	/* After adding locked pages to page cache, the inode holds cache cap.
	 * So we can drop our cap refs. */
	if (got)
		ceph_put_cap_refs(ci, got);

S
Sage Weil 已提交
416 417 418
	return nr_pages;

out_pages:
419 420 421 422 423
	for (i = 0; i < nr_pages; ++i) {
		ceph_fscache_readpage_cancel(inode, pages[i]);
		unlock_page(pages[i]);
	}
	ceph_put_page_vector(pages, nr_pages, false);
424
out_put:
S
Sage Weil 已提交
425
	ceph_osdc_put_request(req);
426 427 428
out:
	if (got)
		ceph_put_cap_refs(ci, got);
S
Sage Weil 已提交
429 430
	return ret;
}
S
Sage Weil 已提交
431

S
Sage Weil 已提交
432 433 434 435 436 437 438 439

/*
 * Read multiple pages.  Leave pages we don't read + unlock in page_list;
 * the caller (VM) cleans them up.
 */
static int ceph_readpages(struct file *file, struct address_space *mapping,
			  struct list_head *page_list, unsigned nr_pages)
{
A
Al Viro 已提交
440
	struct inode *inode = file_inode(file);
441
	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
442
	struct ceph_file_info *fi = file->private_data;
443
	struct ceph_rw_context *rw_ctx;
S
Sage Weil 已提交
444
	int rc = 0;
445 446
	int max = 0;

Y
Yan, Zheng 已提交
447 448 449
	if (ceph_inode(inode)->i_inline_version != CEPH_INLINE_NONE)
		return -EINVAL;

450 451 452 453 454 455
	rc = ceph_readpages_from_fscache(mapping->host, mapping, page_list,
					 &nr_pages);

	if (rc == 0)
		goto out;

456
	rw_ctx = ceph_find_rw_context(fi);
457
	max = fsc->mount_options->rsize >> PAGE_SHIFT;
458 459
	dout("readpages %p file %p ctx %p nr_pages %d max %d\n",
	     inode, file, rw_ctx, nr_pages, max);
S
Sage Weil 已提交
460
	while (!list_empty(page_list)) {
461
		rc = start_read(inode, rw_ctx, page_list, max);
S
Sage Weil 已提交
462 463 464
		if (rc < 0)
			goto out;
	}
S
Sage Weil 已提交
465
out:
466 467
	ceph_fscache_readpages_cancel(inode, page_list);

S
Sage Weil 已提交
468
	dout("readpages %p file %p ret %d\n", inode, file, rc);
S
Sage Weil 已提交
469 470 471
	return rc;
}

472 473 474 475 476 477
struct ceph_writeback_ctl
{
	loff_t i_size;
	u64 truncate_size;
	u32 truncate_seq;
	bool size_stable;
478
	bool head_snapc;
479 480
};

S
Sage Weil 已提交
481 482 483 484
/*
 * Get ref for the oldest snapc for an inode with dirty data... that is, the
 * only snap context we are allowed to write back.
 */
485
static struct ceph_snap_context *
486 487
get_oldest_context(struct inode *inode, struct ceph_writeback_ctl *ctl,
		   struct ceph_snap_context *page_snapc)
S
Sage Weil 已提交
488 489 490 491 492
{
	struct ceph_inode_info *ci = ceph_inode(inode);
	struct ceph_snap_context *snapc = NULL;
	struct ceph_cap_snap *capsnap = NULL;

493
	spin_lock(&ci->i_ceph_lock);
S
Sage Weil 已提交
494 495 496
	list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
		dout(" cap_snap %p snapc %p has %d dirty pages\n", capsnap,
		     capsnap->context, capsnap->dirty_pages);
497 498 499 500 501 502 503 504 505 506 507 508 509 510
		if (!capsnap->dirty_pages)
			continue;

		/* get i_size, truncate_{seq,size} for page_snapc? */
		if (snapc && capsnap->context != page_snapc)
			continue;

		if (ctl) {
			if (capsnap->writing) {
				ctl->i_size = i_size_read(inode);
				ctl->size_stable = false;
			} else {
				ctl->i_size = capsnap->size;
				ctl->size_stable = true;
511
			}
512 513
			ctl->truncate_size = capsnap->truncate_size;
			ctl->truncate_seq = capsnap->truncate_seq;
514
			ctl->head_snapc = false;
S
Sage Weil 已提交
515
		}
516 517 518 519 520 521 522 523 524

		if (snapc)
			break;

		snapc = ceph_get_snap_context(capsnap->context);
		if (!page_snapc ||
		    page_snapc == snapc ||
		    page_snapc->seq > snapc->seq)
			break;
S
Sage Weil 已提交
525
	}
526
	if (!snapc && ci->i_wrbuffer_ref_head) {
527
		snapc = ceph_get_snap_context(ci->i_head_snapc);
S
Sage Weil 已提交
528 529
		dout(" head snapc %p has %d dirty pages\n",
		     snapc, ci->i_wrbuffer_ref_head);
530 531 532 533 534
		if (ctl) {
			ctl->i_size = i_size_read(inode);
			ctl->truncate_size = ci->i_truncate_size;
			ctl->truncate_seq = ci->i_truncate_seq;
			ctl->size_stable = false;
535
			ctl->head_snapc = true;
536
		}
S
Sage Weil 已提交
537
	}
538
	spin_unlock(&ci->i_ceph_lock);
S
Sage Weil 已提交
539 540 541
	return snapc;
}

542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568
static u64 get_writepages_data_length(struct inode *inode,
				      struct page *page, u64 start)
{
	struct ceph_inode_info *ci = ceph_inode(inode);
	struct ceph_snap_context *snapc = page_snap_context(page);
	struct ceph_cap_snap *capsnap = NULL;
	u64 end = i_size_read(inode);

	if (snapc != ci->i_head_snapc) {
		bool found = false;
		spin_lock(&ci->i_ceph_lock);
		list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
			if (capsnap->context == snapc) {
				if (!capsnap->writing)
					end = capsnap->size;
				found = true;
				break;
			}
		}
		spin_unlock(&ci->i_ceph_lock);
		WARN_ON(!found);
	}
	if (end > page_offset(page) + PAGE_SIZE)
		end = page_offset(page) + PAGE_SIZE;
	return end > start ? end - start : 0;
}

S
Sage Weil 已提交
569 570 571 572 573 574 575 576 577 578
/*
 * Write a single page, but leave the page locked.
 *
 * If we get a write error, set the page error bit, but still adjust the
 * dirty page accounting (i.e., page is no longer dirty).
 */
static int writepage_nounlock(struct page *page, struct writeback_control *wbc)
{
	struct inode *inode;
	struct ceph_inode_info *ci;
579
	struct ceph_fs_client *fsc;
580
	struct ceph_snap_context *snapc, *oldest;
581
	loff_t page_off = page_offset(page);
Y
Yan, Zheng 已提交
582
	int err, len = PAGE_SIZE;
583
	struct ceph_writeback_ctl ceph_wbc;
S
Sage Weil 已提交
584 585 586 587 588

	dout("writepage %p idx %lu\n", page, page->index);

	inode = page->mapping->host;
	ci = ceph_inode(inode);
589
	fsc = ceph_inode_to_client(inode);
S
Sage Weil 已提交
590 591

	/* verify this is a writeable snap context */
592
	snapc = page_snap_context(page);
593
	if (!snapc) {
S
Sage Weil 已提交
594
		dout("writepage %p page %p not dirty?\n", inode, page);
Y
Yan, Zheng 已提交
595
		return 0;
S
Sage Weil 已提交
596
	}
597
	oldest = get_oldest_context(inode, &ceph_wbc, snapc);
598
	if (snapc->seq > oldest->seq) {
S
Sage Weil 已提交
599
		dout("writepage %p page %p snapc %p not writeable - noop\n",
600
		     inode, page, snapc);
S
Sage Weil 已提交
601
		/* we should only noop if called by kswapd */
602
		WARN_ON(!(current->flags & PF_MEMALLOC));
603
		ceph_put_snap_context(oldest);
604
		redirty_page_for_writepage(wbc, page);
Y
Yan, Zheng 已提交
605
		return 0;
S
Sage Weil 已提交
606
	}
607
	ceph_put_snap_context(oldest);
S
Sage Weil 已提交
608 609

	/* is this a partial page at end of file? */
610 611
	if (page_off >= ceph_wbc.i_size) {
		dout("%p page eof %llu\n", page, ceph_wbc.i_size);
612
		page->mapping->a_ops->invalidatepage(page, 0, PAGE_SIZE);
Y
Yan, Zheng 已提交
613
		return 0;
614
	}
Y
Yan, Zheng 已提交
615

616 617
	if (ceph_wbc.i_size < page_off + len)
		len = ceph_wbc.i_size - page_off;
S
Sage Weil 已提交
618

619 620
	dout("writepage %p page %p index %lu on %llu~%u snapc %p seq %lld\n",
	     inode, page, page->index, page_off, len, snapc, snapc->seq);
S
Sage Weil 已提交
621

622
	if (atomic_long_inc_return(&fsc->writeback_count) >
623
	    CONGESTION_ON_THRESH(fsc->mount_options->congestion_kb))
624
		set_bdi_congested(inode_to_bdi(inode), BLK_RW_ASYNC);
Y
Yehuda Sadeh 已提交
625

S
Sage Weil 已提交
626
	set_page_writeback(page);
627 628 629 630
	err = ceph_osdc_writepages(&fsc->client->osdc, ceph_vino(inode),
				   &ci->i_layout, snapc, page_off, len,
				   ceph_wbc.truncate_seq,
				   ceph_wbc.truncate_size,
631
				   &inode->i_mtime, &page, 1);
S
Sage Weil 已提交
632
	if (err < 0) {
633 634 635 636 637 638 639 640
		struct writeback_control tmp_wbc;
		if (!wbc)
			wbc = &tmp_wbc;
		if (err == -ERESTARTSYS) {
			/* killed by SIGKILL */
			dout("writepage interrupted page %p\n", page);
			redirty_page_for_writepage(wbc, page);
			end_page_writeback(page);
Y
Yan, Zheng 已提交
641
			return err;
642 643 644
		}
		dout("writepage setting page/mapping error %d %p\n",
		     err, page);
S
Sage Weil 已提交
645 646
		SetPageError(page);
		mapping_set_error(&inode->i_data, err);
647
		wbc->pages_skipped++;
S
Sage Weil 已提交
648 649 650 651 652 653 654 655
	} else {
		dout("writepage cleaned page %p\n", page);
		err = 0;  /* vfs expects us to return 0 */
	}
	page->private = 0;
	ClearPagePrivate(page);
	end_page_writeback(page);
	ceph_put_wrbuffer_cap_refs(ci, 1, snapc);
656
	ceph_put_snap_context(snapc);  /* page's reference */
657 658 659 660 661

	if (atomic_long_dec_return(&fsc->writeback_count) <
	    CONGESTION_OFF_THRESH(fsc->mount_options->congestion_kb))
		clear_bdi_congested(inode_to_bdi(inode), BLK_RW_ASYNC);

S
Sage Weil 已提交
662 663 664 665 666
	return err;
}

static int ceph_writepage(struct page *page, struct writeback_control *wbc)
{
667 668 669
	int err;
	struct inode *inode = page->mapping->host;
	BUG_ON(!inode);
670
	ihold(inode);
671
	err = writepage_nounlock(page, wbc);
672 673 674 675 676
	if (err == -ERESTARTSYS) {
		/* direct memory reclaimer was killed by SIGKILL. return 0
		 * to prevent caller from setting mapping/page error */
		err = 0;
	}
S
Sage Weil 已提交
677
	unlock_page(page);
678
	iput(inode);
S
Sage Weil 已提交
679 680 681 682 683 684 685 686 687 688 689 690
	return err;
}

/*
 * lame release_pages helper.  release_pages() isn't exported to
 * modules.
 */
static void ceph_release_pages(struct page **pages, int num)
{
	struct pagevec pvec;
	int i;

691
	pagevec_init(&pvec);
S
Sage Weil 已提交
692 693 694 695 696 697 698 699 700 701 702 703 704
	for (i = 0; i < num; i++) {
		if (pagevec_add(&pvec, pages[i]) == 0)
			pagevec_release(&pvec);
	}
	pagevec_release(&pvec);
}

/*
 * async writeback completion handler.
 *
 * If we get an error, set the mapping error bit, but not the individual
 * page error bits.
 */
705
static void writepages_finish(struct ceph_osd_request *req)
S
Sage Weil 已提交
706 707 708
{
	struct inode *inode = req->r_inode;
	struct ceph_inode_info *ci = ceph_inode(inode);
709
	struct ceph_osd_data *osd_data;
S
Sage Weil 已提交
710
	struct page *page;
Y
Yan, Zheng 已提交
711 712 713
	int num_pages, total_pages = 0;
	int i, j;
	int rc = req->r_result;
S
Sage Weil 已提交
714 715
	struct ceph_snap_context *snapc = req->r_snapc;
	struct address_space *mapping = inode->i_mapping;
716
	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
Y
Yan, Zheng 已提交
717
	bool remove_page;
S
Sage Weil 已提交
718

Y
Yan, Zheng 已提交
719
	dout("writepages_finish %p rc %d\n", inode, rc);
720
	if (rc < 0) {
S
Sage Weil 已提交
721
		mapping_set_error(mapping, rc);
722 723 724 725
		ceph_set_error_write(ci);
	} else {
		ceph_clear_error_write(ci);
	}
Y
Yan, Zheng 已提交
726 727 728 729 730 731 732 733 734

	/*
	 * We lost the cache cap, need to truncate the page before
	 * it is unlocked, otherwise we'd truncate it later in the
	 * page truncation thread, possibly losing some data that
	 * raced its way in
	 */
	remove_page = !(ceph_caps_issued(ci) &
			(CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO));
S
Sage Weil 已提交
735 736

	/* clean all pages */
Y
Yan, Zheng 已提交
737 738 739
	for (i = 0; i < req->r_num_ops; i++) {
		if (req->r_ops[i].op != CEPH_OSD_OP_WRITE)
			break;
740

Y
Yan, Zheng 已提交
741 742 743 744 745 746 747 748 749 750 751 752 753
		osd_data = osd_req_op_extent_osd_data(req, i);
		BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGES);
		num_pages = calc_pages_for((u64)osd_data->alignment,
					   (u64)osd_data->length);
		total_pages += num_pages;
		for (j = 0; j < num_pages; j++) {
			page = osd_data->pages[j];
			BUG_ON(!page);
			WARN_ON(!PageUptodate(page));

			if (atomic_long_dec_return(&fsc->writeback_count) <
			     CONGESTION_OFF_THRESH(
					fsc->mount_options->congestion_kb))
754
				clear_bdi_congested(inode_to_bdi(inode),
Y
Yan, Zheng 已提交
755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770
						    BLK_RW_ASYNC);

			ceph_put_snap_context(page_snap_context(page));
			page->private = 0;
			ClearPagePrivate(page);
			dout("unlocking %p\n", page);
			end_page_writeback(page);

			if (remove_page)
				generic_error_remove_page(inode->i_mapping,
							  page);

			unlock_page(page);
		}
		dout("writepages_finish %p wrote %llu bytes cleaned %d pages\n",
		     inode, osd_data->length, rc >= 0 ? num_pages : 0);
771

Y
Yan, Zheng 已提交
772
		ceph_release_pages(osd_data->pages, num_pages);
S
Sage Weil 已提交
773 774
	}

Y
Yan, Zheng 已提交
775 776 777
	ceph_put_wrbuffer_cap_refs(ci, total_pages, snapc);

	osd_data = osd_req_op_extent_osd_data(req, 0);
778 779
	if (osd_data->pages_from_pool)
		mempool_free(osd_data->pages,
780
			     ceph_sb_to_client(inode->i_sb)->wb_pagevec_pool);
S
Sage Weil 已提交
781
	else
782
		kfree(osd_data->pages);
S
Sage Weil 已提交
783 784 785 786 787 788 789 790 791 792 793
	ceph_osdc_put_request(req);
}

/*
 * initiate async writeback
 */
static int ceph_writepages_start(struct address_space *mapping,
				 struct writeback_control *wbc)
{
	struct inode *inode = mapping->host;
	struct ceph_inode_info *ci = ceph_inode(inode);
794 795
	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
	struct ceph_vino vino = ceph_vino(inode);
796
	pgoff_t index, start_index, end = -1;
797
	struct ceph_snap_context *snapc = NULL, *last_snapc = NULL, *pgsnapc;
S
Sage Weil 已提交
798 799
	struct pagevec pvec;
	int rc = 0;
F
Fabian Frederick 已提交
800
	unsigned int wsize = i_blocksize(inode);
S
Sage Weil 已提交
801
	struct ceph_osd_request *req = NULL;
802
	struct ceph_writeback_ctl ceph_wbc;
803
	bool should_loop, range_whole = false;
804
	bool done = false;
S
Sage Weil 已提交
805

Y
Yanhu Cao 已提交
806
	dout("writepages_start %p (mode=%s)\n", inode,
S
Sage Weil 已提交
807 808 809
	     wbc->sync_mode == WB_SYNC_NONE ? "NONE" :
	     (wbc->sync_mode == WB_SYNC_ALL ? "ALL" : "HOLD"));

810
	if (READ_ONCE(fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
811 812 813 814 815
		if (ci->i_wrbuffer_ref > 0) {
			pr_warn_ratelimited(
				"writepage_start %p %lld forced umount\n",
				inode, ceph_ino(inode));
		}
816
		mapping_set_error(mapping, -EIO);
S
Sage Weil 已提交
817 818
		return -EIO; /* we're in a forced umount, don't write! */
	}
Y
Yan, Zheng 已提交
819
	if (fsc->mount_options->wsize < wsize)
820
		wsize = fsc->mount_options->wsize;
S
Sage Weil 已提交
821

822
	pagevec_init(&pvec);
S
Sage Weil 已提交
823

824
	start_index = wbc->range_cyclic ? mapping->writeback_index : 0;
825
	index = start_index;
S
Sage Weil 已提交
826 827 828

retry:
	/* find oldest snap context with dirty data */
829
	snapc = get_oldest_context(inode, &ceph_wbc, NULL);
S
Sage Weil 已提交
830 831 832 833 834 835 836 837
	if (!snapc) {
		/* hmm, why does writepages get called when there
		   is no dirty data? */
		dout(" no snap context with dirty data?\n");
		goto out;
	}
	dout(" oldest snapc is %p seq %lld (%d snaps)\n",
	     snapc, snapc->seq, snapc->num_snaps);
838

839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859
	should_loop = false;
	if (ceph_wbc.head_snapc && snapc != last_snapc) {
		/* where to start/end? */
		if (wbc->range_cyclic) {
			index = start_index;
			end = -1;
			if (index > 0)
				should_loop = true;
			dout(" cyclic, start at %lu\n", index);
		} else {
			index = wbc->range_start >> PAGE_SHIFT;
			end = wbc->range_end >> PAGE_SHIFT;
			if (wbc->range_start == 0 && wbc->range_end == LLONG_MAX)
				range_whole = true;
			dout(" not cyclic, %lu to %lu\n", index, end);
		}
	} else if (!ceph_wbc.head_snapc) {
		/* Do not respect wbc->range_{start,end}. Dirty pages
		 * in that range can be associated with newer snapc.
		 * They are not writeable until we write all dirty pages
		 * associated with 'snapc' get written */
860
		if (index > 0)
861 862
			should_loop = true;
		dout(" non-head snapc, range whole\n");
S
Sage Weil 已提交
863
	}
864 865

	ceph_put_snap_context(last_snapc);
S
Sage Weil 已提交
866 867
	last_snapc = snapc;

868
	while (!done && index <= end) {
Y
Yan, Zheng 已提交
869
		int num_ops = 0, op_idx;
870
		unsigned i, pvec_pages, max_pages, locked_pages = 0;
Y
Yan, Zheng 已提交
871
		struct page **pages = NULL, **data_pages;
872
		mempool_t *pool = NULL;	/* Becomes non-null if mempool used */
S
Sage Weil 已提交
873
		struct page *page;
874
		pgoff_t strip_unit_end = 0;
Y
Yan, Zheng 已提交
875
		u64 offset = 0, len = 0;
S
Sage Weil 已提交
876

877
		max_pages = wsize >> PAGE_SHIFT;
S
Sage Weil 已提交
878 879

get_more_pages:
J
Jan Kara 已提交
880
		pvec_pages = pagevec_lookup_range_nr_tag(&pvec, mapping, &index,
J
Jan Kara 已提交
881
						end, PAGECACHE_TAG_DIRTY,
J
Jan Kara 已提交
882
						max_pages - locked_pages);
J
Jan Kara 已提交
883
		dout("pagevec_lookup_range_tag got %d\n", pvec_pages);
S
Sage Weil 已提交
884 885 886 887 888 889 890 891 892 893 894 895 896 897 898
		if (!pvec_pages && !locked_pages)
			break;
		for (i = 0; i < pvec_pages && locked_pages < max_pages; i++) {
			page = pvec.pages[i];
			dout("? %p idx %lu\n", page, page->index);
			if (locked_pages == 0)
				lock_page(page);  /* first page */
			else if (!trylock_page(page))
				break;

			/* only dirty pages, or our accounting breaks */
			if (unlikely(!PageDirty(page)) ||
			    unlikely(page->mapping != mapping)) {
				dout("!dirty or !mapping %p\n", page);
				unlock_page(page);
899
				continue;
S
Sage Weil 已提交
900
			}
901 902 903 904 905
			/* only if matching snap context */
			pgsnapc = page_snap_context(page);
			if (pgsnapc != snapc) {
				dout("page snapc %p %lld != oldest %p %lld\n",
				     pgsnapc, pgsnapc->seq, snapc, snapc->seq);
906 907 908 909
				if (!should_loop &&
				    !ceph_wbc.head_snapc &&
				    wbc->sync_mode != WB_SYNC_NONE)
					should_loop = true;
S
Sage Weil 已提交
910
				unlock_page(page);
911
				continue;
S
Sage Weil 已提交
912
			}
913 914 915
			if (page_offset(page) >= ceph_wbc.i_size) {
				dout("%p page eof %llu\n",
				     page, ceph_wbc.i_size);
916 917 918
				if ((ceph_wbc.size_stable ||
				    page_offset(page) >= i_size_read(inode)) &&
				    clear_page_dirty_for_io(page))
919 920 921 922 923 924 925
					mapping->a_ops->invalidatepage(page,
								0, PAGE_SIZE);
				unlock_page(page);
				continue;
			}
			if (strip_unit_end && (page->index > strip_unit_end)) {
				dout("end of strip unit %p\n", page);
S
Sage Weil 已提交
926 927 928 929
				unlock_page(page);
				break;
			}
			if (PageWriteback(page)) {
930 931 932 933 934 935 936
				if (wbc->sync_mode == WB_SYNC_NONE) {
					dout("%p under writeback\n", page);
					unlock_page(page);
					continue;
				}
				dout("waiting on writeback %p\n", page);
				wait_on_page_writeback(page);
S
Sage Weil 已提交
937 938 939 940 941
			}

			if (!clear_page_dirty_for_io(page)) {
				dout("%p !clear_page_dirty_for_io\n", page);
				unlock_page(page);
942
				continue;
S
Sage Weil 已提交
943 944
			}

945 946 947
			/*
			 * We have something to write.  If this is
			 * the first locked page this time through,
Y
Yan, Zheng 已提交
948 949
			 * calculate max possinle write size and
			 * allocate a page array
950
			 */
S
Sage Weil 已提交
951
			if (locked_pages == 0) {
Y
Yan, Zheng 已提交
952 953
				u64 objnum;
				u64 objoff;
954
				u32 xlen;
Y
Yan, Zheng 已提交
955

S
Sage Weil 已提交
956
				/* prepare async write request */
957
				offset = (u64)page_offset(page);
958 959 960 961 962
				ceph_calc_file_object_mapping(&ci->i_layout,
							      offset, wsize,
							      &objnum, &objoff,
							      &xlen);
				len = xlen;
963

Y
Yanhu Cao 已提交
964
				num_ops = 1;
Y
Yan, Zheng 已提交
965
				strip_unit_end = page->index +
966
					((len - 1) >> PAGE_SHIFT);
A
Alex Elder 已提交
967

Y
Yan, Zheng 已提交
968
				BUG_ON(pages);
A
Alex Elder 已提交
969
				max_pages = calc_pages_for(0, (u64)len);
970 971 972
				pages = kmalloc_array(max_pages,
						      sizeof(*pages),
						      GFP_NOFS);
A
Alex Elder 已提交
973 974 975
				if (!pages) {
					pool = fsc->wb_pagevec_pool;
					pages = mempool_alloc(pool, GFP_NOFS);
976
					BUG_ON(!pages);
A
Alex Elder 已提交
977
				}
Y
Yan, Zheng 已提交
978 979 980

				len = 0;
			} else if (page->index !=
981
				   (offset + len) >> PAGE_SHIFT) {
Y
Yan, Zheng 已提交
982 983 984 985 986 987 988 989 990 991
				if (num_ops >= (pool ?  CEPH_OSD_SLAB_OPS :
							CEPH_OSD_MAX_OPS)) {
					redirty_page_for_writepage(wbc, page);
					unlock_page(page);
					break;
				}

				num_ops++;
				offset = (u64)page_offset(page);
				len = 0;
S
Sage Weil 已提交
992 993 994 995 996
			}

			/* note position of first page in pvec */
			dout("%p will write page %p idx %lu\n",
			     inode, page, page->index);
Y
Yehuda Sadeh 已提交
997

Y
Yan, Zheng 已提交
998 999
			if (atomic_long_inc_return(&fsc->writeback_count) >
			    CONGESTION_ON_THRESH(
1000
				    fsc->mount_options->congestion_kb)) {
1001
				set_bdi_congested(inode_to_bdi(inode),
S
Sage Weil 已提交
1002
						  BLK_RW_ASYNC);
Y
Yehuda Sadeh 已提交
1003 1004
			}

1005 1006 1007 1008

			pages[locked_pages++] = page;
			pvec.pages[i] = NULL;

1009
			len += PAGE_SIZE;
S
Sage Weil 已提交
1010 1011 1012 1013 1014 1015
		}

		/* did we get anything? */
		if (!locked_pages)
			goto release_pvec_pages;
		if (i) {
1016 1017 1018 1019 1020 1021 1022 1023 1024 1025
			unsigned j, n = 0;
			/* shift unused page to beginning of pvec */
			for (j = 0; j < pvec_pages; j++) {
				if (!pvec.pages[j])
					continue;
				if (n < j)
					pvec.pages[n] = pvec.pages[j];
				n++;
			}
			pvec.nr = n;
S
Sage Weil 已提交
1026 1027 1028 1029

			if (pvec_pages && i == pvec_pages &&
			    locked_pages < max_pages) {
				dout("reached end pvec, trying for more\n");
1030
				pagevec_release(&pvec);
S
Sage Weil 已提交
1031 1032 1033 1034
				goto get_more_pages;
			}
		}

Y
Yan, Zheng 已提交
1035
new_request:
1036
		offset = page_offset(pages[0]);
Y
Yan, Zheng 已提交
1037 1038 1039 1040 1041
		len = wsize;

		req = ceph_osdc_new_request(&fsc->client->osdc,
					&ci->i_layout, vino,
					offset, &len, 0, num_ops,
1042 1043 1044
					CEPH_OSD_OP_WRITE, CEPH_OSD_FLAG_WRITE,
					snapc, ceph_wbc.truncate_seq,
					ceph_wbc.truncate_size, false);
Y
Yan, Zheng 已提交
1045 1046 1047 1048 1049 1050 1051
		if (IS_ERR(req)) {
			req = ceph_osdc_new_request(&fsc->client->osdc,
						&ci->i_layout, vino,
						offset, &len, 0,
						min(num_ops,
						    CEPH_OSD_SLAB_OPS),
						CEPH_OSD_OP_WRITE,
1052
						CEPH_OSD_FLAG_WRITE,
1053 1054
						snapc, ceph_wbc.truncate_seq,
						ceph_wbc.truncate_size, true);
Y
Yan, Zheng 已提交
1055
			BUG_ON(IS_ERR(req));
Y
Yan, Zheng 已提交
1056
		}
Y
Yan, Zheng 已提交
1057
		BUG_ON(len < page_offset(pages[locked_pages - 1]) +
1058
			     PAGE_SIZE - offset);
Y
Yan, Zheng 已提交
1059 1060 1061

		req->r_callback = writepages_finish;
		req->r_inode = inode;
S
Sage Weil 已提交
1062

Y
Yan, Zheng 已提交
1063 1064 1065 1066 1067 1068 1069
		/* Format the osd request message and submit the write */
		len = 0;
		data_pages = pages;
		op_idx = 0;
		for (i = 0; i < locked_pages; i++) {
			u64 cur_offset = page_offset(pages[i]);
			if (offset + len != cur_offset) {
Y
Yanhu Cao 已提交
1070
				if (op_idx + 1 == req->r_num_ops)
Y
Yan, Zheng 已提交
1071 1072 1073 1074 1075 1076 1077
					break;
				osd_req_op_extent_dup_last(req, op_idx,
							   cur_offset - offset);
				dout("writepages got pages at %llu~%llu\n",
				     offset, len);
				osd_req_op_extent_osd_data_pages(req, op_idx,
							data_pages, len, 0,
1078
							!!pool, false);
Y
Yan, Zheng 已提交
1079
				osd_req_op_extent_update(req, op_idx, len);
1080

Y
Yan, Zheng 已提交
1081 1082 1083 1084 1085 1086 1087
				len = 0;
				offset = cur_offset; 
				data_pages = pages + i;
				op_idx++;
			}

			set_page_writeback(pages[i]);
1088
			len += PAGE_SIZE;
Y
Yan, Zheng 已提交
1089 1090
		}

1091 1092
		if (ceph_wbc.size_stable) {
			len = min(len, ceph_wbc.i_size - offset);
Y
Yan, Zheng 已提交
1093 1094 1095 1096
		} else if (i == locked_pages) {
			/* writepages_finish() clears writeback pages
			 * according to the data length, so make sure
			 * data length covers all locked pages */
1097
			u64 min_len = len + 1 - PAGE_SIZE;
1098 1099
			len = get_writepages_data_length(inode, pages[i - 1],
							 offset);
Y
Yan, Zheng 已提交
1100 1101 1102
			len = max(len, min_len);
		}
		dout("writepages got pages at %llu~%llu\n", offset, len);
1103

Y
Yan, Zheng 已提交
1104 1105 1106
		osd_req_op_extent_osd_data_pages(req, op_idx, data_pages, len,
						 0, !!pool, false);
		osd_req_op_extent_update(req, op_idx, len);
1107

Y
Yan, Zheng 已提交
1108 1109 1110 1111 1112 1113 1114 1115 1116 1117
		BUG_ON(op_idx + 1 != req->r_num_ops);

		pool = NULL;
		if (i < locked_pages) {
			BUG_ON(num_ops <= req->r_num_ops);
			num_ops -= req->r_num_ops;
			locked_pages -= i;

			/* allocate new pages array for next request */
			data_pages = pages;
1118 1119
			pages = kmalloc_array(locked_pages, sizeof(*pages),
					      GFP_NOFS);
Y
Yan, Zheng 已提交
1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134
			if (!pages) {
				pool = fsc->wb_pagevec_pool;
				pages = mempool_alloc(pool, GFP_NOFS);
				BUG_ON(!pages);
			}
			memcpy(pages, data_pages + i,
			       locked_pages * sizeof(*pages));
			memset(data_pages + i, 0,
			       locked_pages * sizeof(*pages));
		} else {
			BUG_ON(num_ops != req->r_num_ops);
			index = pages[i - 1]->index + 1;
			/* request message now owns the pages array */
			pages = NULL;
		}
1135

1136
		req->r_mtime = inode->i_mtime;
1137 1138
		rc = ceph_osdc_start_request(&fsc->client->osdc, req, true);
		BUG_ON(rc);
S
Sage Weil 已提交
1139 1140
		req = NULL;

Y
Yan, Zheng 已提交
1141 1142 1143 1144
		wbc->nr_to_write -= i;
		if (pages)
			goto new_request;

1145 1146 1147 1148 1149 1150 1151
		/*
		 * We stop writing back only if we are not doing
		 * integrity sync. In case of integrity sync we have to
		 * keep going until we have written all the pages
		 * we tagged for writeback prior to entering this loop.
		 */
		if (wbc->nr_to_write <= 0 && wbc->sync_mode == WB_SYNC_NONE)
1152
			done = true;
S
Sage Weil 已提交
1153 1154 1155 1156 1157 1158 1159 1160 1161 1162

release_pvec_pages:
		dout("pagevec_release on %d pages (%p)\n", (int)pvec.nr,
		     pvec.nr ? pvec.pages[0] : NULL);
		pagevec_release(&pvec);
	}

	if (should_loop && !done) {
		/* more to do; loop back to beginning of file */
		dout("writepages looping back to beginning of file\n");
1163
		end = start_index - 1; /* OK even when start_index == 0 */
1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174

		/* to write dirty pages associated with next snapc,
		 * we need to wait until current writes complete */
		if (wbc->sync_mode != WB_SYNC_NONE &&
		    start_index == 0 && /* all dirty pages were checked */
		    !ceph_wbc.head_snapc) {
			struct page *page;
			unsigned i, nr;
			index = 0;
			while ((index <= end) &&
			       (nr = pagevec_lookup_tag(&pvec, mapping, &index,
1175
						PAGECACHE_TAG_WRITEBACK))) {
1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186
				for (i = 0; i < nr; i++) {
					page = pvec.pages[i];
					if (page_snap_context(page) != snapc)
						continue;
					wait_on_page_writeback(page);
				}
				pagevec_release(&pvec);
				cond_resched();
			}
		}

1187
		start_index = 0;
S
Sage Weil 已提交
1188 1189 1190 1191 1192 1193 1194 1195
		index = 0;
		goto retry;
	}

	if (wbc->range_cyclic || (range_whole && wbc->nr_to_write > 0))
		mapping->writeback_index = index;

out:
1196
	ceph_osdc_put_request(req);
1197 1198
	ceph_put_snap_context(last_snapc);
	dout("writepages dend - startone, rc = %d\n", rc);
S
Sage Weil 已提交
1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209
	return rc;
}



/*
 * See if a given @snapc is either writeable, or already written.
 */
static int context_is_writeable_or_written(struct inode *inode,
					   struct ceph_snap_context *snapc)
{
1210
	struct ceph_snap_context *oldest = get_oldest_context(inode, NULL, NULL);
1211 1212 1213 1214
	int ret = !oldest || snapc->seq <= oldest->seq;

	ceph_put_snap_context(oldest);
	return ret;
S
Sage Weil 已提交
1215 1216 1217 1218 1219
}

/*
 * We are only allowed to write into/dirty the page if the page is
 * clean, or already dirty within the same snap context.
1220 1221 1222 1223
 *
 * called with page locked.
 * return success with page locked,
 * or any failure (incl -EAGAIN) with page unlocked.
S
Sage Weil 已提交
1224
 */
1225 1226 1227
static int ceph_update_writeable_page(struct file *file,
			    loff_t pos, unsigned len,
			    struct page *page)
S
Sage Weil 已提交
1228
{
A
Al Viro 已提交
1229
	struct inode *inode = file_inode(file);
1230
	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
S
Sage Weil 已提交
1231
	struct ceph_inode_info *ci = ceph_inode(inode);
1232 1233
	loff_t page_off = pos & PAGE_MASK;
	int pos_in_page = pos & ~PAGE_MASK;
S
Sage Weil 已提交
1234 1235 1236
	int end_in_page = pos_in_page + len;
	loff_t i_size;
	int r;
1237
	struct ceph_snap_context *snapc, *oldest;
S
Sage Weil 已提交
1238

1239
	if (READ_ONCE(fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
1240 1241 1242 1243 1244
		dout(" page %p forced umount\n", page);
		unlock_page(page);
		return -EIO;
	}

S
Sage Weil 已提交
1245 1246 1247 1248
retry_locked:
	/* writepages currently holds page lock, but if we change that later, */
	wait_on_page_writeback(page);

1249
	snapc = page_snap_context(page);
1250
	if (snapc && snapc != ci->i_head_snapc) {
S
Sage Weil 已提交
1251 1252 1253 1254
		/*
		 * this page is already dirty in another (older) snap
		 * context!  is it writeable now?
		 */
1255
		oldest = get_oldest_context(inode, NULL, NULL);
1256
		if (snapc->seq > oldest->seq) {
1257
			ceph_put_snap_context(oldest);
S
Sage Weil 已提交
1258
			dout(" page %p snapc %p not current or oldest\n",
1259
			     page, snapc);
S
Sage Weil 已提交
1260 1261 1262 1263
			/*
			 * queue for writeback, and wait for snapc to
			 * be writeable or written
			 */
1264
			snapc = ceph_get_snap_context(snapc);
S
Sage Weil 已提交
1265
			unlock_page(page);
1266
			ceph_queue_writeback(inode);
1267
			r = wait_event_killable(ci->i_cap_wq,
S
Sage Weil 已提交
1268 1269
			       context_is_writeable_or_written(inode, snapc));
			ceph_put_snap_context(snapc);
1270 1271
			if (r == -ERESTARTSYS)
				return r;
1272
			return -EAGAIN;
S
Sage Weil 已提交
1273
		}
1274
		ceph_put_snap_context(oldest);
S
Sage Weil 已提交
1275 1276 1277 1278 1279 1280 1281 1282

		/* yay, writeable, do it now (without dropping page lock) */
		dout(" page %p snapc %p not current, but oldest\n",
		     page, snapc);
		if (!clear_page_dirty_for_io(page))
			goto retry_locked;
		r = writepage_nounlock(page, NULL);
		if (r < 0)
Y
Yan, Zheng 已提交
1283
			goto fail_unlock;
S
Sage Weil 已提交
1284 1285 1286 1287 1288 1289 1290 1291 1292
		goto retry_locked;
	}

	if (PageUptodate(page)) {
		dout(" page %p already uptodate\n", page);
		return 0;
	}

	/* full page? */
1293
	if (pos_in_page == 0 && len == PAGE_SIZE)
S
Sage Weil 已提交
1294 1295 1296
		return 0;

	/* past end of file? */
1297
	i_size = i_size_read(inode);
S
Sage Weil 已提交
1298 1299 1300

	if (page_off >= i_size ||
	    (pos_in_page == 0 && (pos+len) >= i_size &&
1301
	     end_in_page - pos_in_page != PAGE_SIZE)) {
S
Sage Weil 已提交
1302
		dout(" zeroing %p 0 - %d and %d - %d\n",
1303
		     page, pos_in_page, end_in_page, (int)PAGE_SIZE);
S
Sage Weil 已提交
1304 1305
		zero_user_segments(page,
				   0, pos_in_page,
1306
				   end_in_page, PAGE_SIZE);
S
Sage Weil 已提交
1307 1308 1309 1310
		return 0;
	}

	/* we need to read it. */
Y
Yan, Zheng 已提交
1311 1312 1313 1314 1315 1316
	r = ceph_do_readpage(file, page);
	if (r < 0) {
		if (r == -EINPROGRESS)
			return -EAGAIN;
		goto fail_unlock;
	}
S
Sage Weil 已提交
1317
	goto retry_locked;
Y
Yan, Zheng 已提交
1318
fail_unlock:
S
Sage Weil 已提交
1319 1320 1321 1322
	unlock_page(page);
	return r;
}

1323 1324 1325 1326 1327 1328 1329 1330
/*
 * We are only allowed to write into/dirty the page if the page is
 * clean, or already dirty within the same snap context.
 */
static int ceph_write_begin(struct file *file, struct address_space *mapping,
			    loff_t pos, unsigned len, unsigned flags,
			    struct page **pagep, void **fsdata)
{
A
Al Viro 已提交
1331
	struct inode *inode = file_inode(file);
1332
	struct page *page;
1333
	pgoff_t index = pos >> PAGE_SHIFT;
S
Sage Weil 已提交
1334
	int r;
1335 1336

	do {
1337
		/* get a page */
1338
		page = grab_cache_page_write_begin(mapping, index, 0);
S
Sage Weil 已提交
1339 1340
		if (!page)
			return -ENOMEM;
1341 1342

		dout("write_begin file %p inode %p page %p %d~%d\n", file,
S
Sage Weil 已提交
1343
		     inode, page, (int)pos, (int)len);
1344 1345

		r = ceph_update_writeable_page(file, pos, len, page);
1346
		if (r < 0)
1347
			put_page(page);
1348 1349
		else
			*pagep = page;
1350 1351 1352 1353 1354
	} while (r == -EAGAIN);

	return r;
}

S
Sage Weil 已提交
1355 1356
/*
 * we don't do anything in here that simple_write_end doesn't do
1357
 * except adjust dirty page accounting
S
Sage Weil 已提交
1358 1359 1360 1361 1362
 */
static int ceph_write_end(struct file *file, struct address_space *mapping,
			  loff_t pos, unsigned len, unsigned copied,
			  struct page *page, void *fsdata)
{
A
Al Viro 已提交
1363
	struct inode *inode = file_inode(file);
1364
	bool check_cap = false;
S
Sage Weil 已提交
1365 1366 1367 1368 1369

	dout("write_end file %p inode %p page %p %d~%d (%d)\n", file,
	     inode, page, (int)pos, (int)copied, (int)len);

	/* zero the stale part of the page if we did a short copy */
A
Al Viro 已提交
1370 1371 1372 1373 1374 1375 1376
	if (!PageUptodate(page)) {
		if (copied < len) {
			copied = 0;
			goto out;
		}
		SetPageUptodate(page);
	}
S
Sage Weil 已提交
1377 1378

	/* did file size increase? */
1379
	if (pos+copied > i_size_read(inode))
S
Sage Weil 已提交
1380 1381 1382 1383
		check_cap = ceph_inode_set_size(inode, pos+copied);

	set_page_dirty(page);

A
Al Viro 已提交
1384
out:
S
Sage Weil 已提交
1385
	unlock_page(page);
1386
	put_page(page);
S
Sage Weil 已提交
1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398

	if (check_cap)
		ceph_check_caps(ceph_inode(inode), CHECK_CAPS_AUTHONLY, NULL);

	return copied;
}

/*
 * we set .direct_IO to indicate direct io is supported, but since we
 * intercept O_DIRECT reads and writes early, this function should
 * never get called.
 */
1399
static ssize_t ceph_direct_io(struct kiocb *iocb, struct iov_iter *iter)
S
Sage Weil 已提交
1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417
{
	WARN_ON(1);
	return -EINVAL;
}

const struct address_space_operations ceph_aops = {
	.readpage = ceph_readpage,
	.readpages = ceph_readpages,
	.writepage = ceph_writepage,
	.writepages = ceph_writepages_start,
	.write_begin = ceph_write_begin,
	.write_end = ceph_write_end,
	.set_page_dirty = ceph_set_page_dirty,
	.invalidatepage = ceph_invalidatepage,
	.releasepage = ceph_releasepage,
	.direct_IO = ceph_direct_io,
};

1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428
static void ceph_block_sigs(sigset_t *oldset)
{
	sigset_t mask;
	siginitsetinv(&mask, sigmask(SIGKILL));
	sigprocmask(SIG_BLOCK, &mask, oldset);
}

static void ceph_restore_sigs(sigset_t *oldset)
{
	sigprocmask(SIG_SETMASK, oldset, NULL);
}
S
Sage Weil 已提交
1429 1430 1431 1432

/*
 * vm ops
 */
1433
static vm_fault_t ceph_filemap_fault(struct vm_fault *vmf)
1434
{
1435
	struct vm_area_struct *vma = vmf->vma;
1436 1437 1438
	struct inode *inode = file_inode(vma->vm_file);
	struct ceph_inode_info *ci = ceph_inode(inode);
	struct ceph_file_info *fi = vma->vm_file->private_data;
1439
	struct page *pinned_page = NULL;
1440
	loff_t off = vmf->pgoff << PAGE_SHIFT;
1441
	int want, got, err;
1442
	sigset_t oldset;
1443
	vm_fault_t ret = VM_FAULT_SIGBUS;
1444 1445

	ceph_block_sigs(&oldset);
1446 1447

	dout("filemap_fault %p %llx.%llx %llu~%zd trying to get caps\n",
1448
	     inode, ceph_vinop(inode), off, (size_t)PAGE_SIZE);
1449 1450 1451 1452
	if (fi->fmode & CEPH_FILE_MODE_LAZY)
		want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
	else
		want = CEPH_CAP_FILE_CACHE;
1453 1454

	got = 0;
1455 1456
	err = ceph_get_caps(ci, CEPH_CAP_FILE_RD, want, -1, &got, &pinned_page);
	if (err < 0)
1457
		goto out_restore;
1458

1459
	dout("filemap_fault %p %llu~%zd got cap refs on %s\n",
1460
	     inode, off, (size_t)PAGE_SIZE, ceph_cap_string(got));
1461

Y
Yan, Zheng 已提交
1462
	if ((got & (CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO)) ||
1463
	    ci->i_inline_version == CEPH_INLINE_NONE) {
1464 1465
		CEPH_DEFINE_RW_CONTEXT(rw_ctx, got);
		ceph_add_rw_context(fi, &rw_ctx);
1466
		ret = filemap_fault(vmf);
1467
		ceph_del_rw_context(fi, &rw_ctx);
1468 1469 1470
		dout("filemap_fault %p %llu~%zd drop cap refs %s ret %x\n",
			inode, off, (size_t)PAGE_SIZE,
				ceph_cap_string(got), ret);
1471
	} else
1472
		err = -EAGAIN;
1473

1474
	if (pinned_page)
1475
		put_page(pinned_page);
1476 1477
	ceph_put_cap_refs(ci, got);

1478
	if (err != -EAGAIN)
1479
		goto out_restore;
Y
Yan, Zheng 已提交
1480 1481

	/* read inline data */
1482
	if (off >= PAGE_SIZE) {
Y
Yan, Zheng 已提交
1483 1484 1485 1486 1487
		/* does not support inline data > PAGE_SIZE */
		ret = VM_FAULT_SIGBUS;
	} else {
		struct address_space *mapping = inode->i_mapping;
		struct page *page = find_or_create_page(mapping, 0,
1488 1489
						mapping_gfp_constraint(mapping,
						~__GFP_FS));
Y
Yan, Zheng 已提交
1490 1491
		if (!page) {
			ret = VM_FAULT_OOM;
1492
			goto out_inline;
Y
Yan, Zheng 已提交
1493
		}
1494
		err = __ceph_do_getattr(inode, page,
Y
Yan, Zheng 已提交
1495
					 CEPH_STAT_CAP_INLINE_DATA, true);
1496
		if (err < 0 || off >= i_size_read(inode)) {
Y
Yan, Zheng 已提交
1497
			unlock_page(page);
1498
			put_page(page);
1499 1500
			if (err == -ENOMEM)
				ret = VM_FAULT_OOM;
1501 1502
			else
				ret = VM_FAULT_SIGBUS;
1503
			goto out_inline;
Y
Yan, Zheng 已提交
1504
		}
1505 1506
		if (err < PAGE_SIZE)
			zero_user_segment(page, err, PAGE_SIZE);
Y
Yan, Zheng 已提交
1507 1508 1509 1510 1511
		else
			flush_dcache_page(page);
		SetPageUptodate(page);
		vmf->page = page;
		ret = VM_FAULT_MAJOR | VM_FAULT_LOCKED;
1512
out_inline:
1513
		dout("filemap_fault %p %llu~%zd read inline data ret %x\n",
1514
		     inode, off, (size_t)PAGE_SIZE, ret);
Y
Yan, Zheng 已提交
1515
	}
1516 1517
out_restore:
	ceph_restore_sigs(&oldset);
1518 1519
	if (err < 0)
		ret = vmf_error(err);
1520

1521 1522
	return ret;
}
S
Sage Weil 已提交
1523 1524 1525 1526

/*
 * Reuse write_begin here for simplicity.
 */
1527
static vm_fault_t ceph_page_mkwrite(struct vm_fault *vmf)
S
Sage Weil 已提交
1528
{
1529
	struct vm_area_struct *vma = vmf->vma;
A
Al Viro 已提交
1530
	struct inode *inode = file_inode(vma->vm_file);
1531 1532
	struct ceph_inode_info *ci = ceph_inode(inode);
	struct ceph_file_info *fi = vma->vm_file->private_data;
1533
	struct ceph_cap_flush *prealloc_cf;
1534
	struct page *page = vmf->page;
1535
	loff_t off = page_offset(page);
1536 1537
	loff_t size = i_size_read(inode);
	size_t len;
1538
	int want, got, err;
1539
	sigset_t oldset;
1540
	vm_fault_t ret = VM_FAULT_SIGBUS;
1541

1542 1543
	prealloc_cf = ceph_alloc_cap_flush();
	if (!prealloc_cf)
1544
		return VM_FAULT_OOM;
1545

1546
	ceph_block_sigs(&oldset);
1547

1548 1549 1550 1551 1552 1553
	if (ci->i_inline_version != CEPH_INLINE_NONE) {
		struct page *locked_page = NULL;
		if (off == 0) {
			lock_page(page);
			locked_page = page;
		}
1554
		err = ceph_uninline_data(vma->vm_file, locked_page);
1555 1556
		if (locked_page)
			unlock_page(locked_page);
1557
		if (err < 0)
1558
			goto out_free;
1559 1560
	}

1561 1562
	if (off + PAGE_SIZE <= size)
		len = PAGE_SIZE;
S
Sage Weil 已提交
1563
	else
1564
		len = size & ~PAGE_MASK;
S
Sage Weil 已提交
1565

1566 1567 1568 1569 1570 1571
	dout("page_mkwrite %p %llx.%llx %llu~%zd getting caps i_size %llu\n",
	     inode, ceph_vinop(inode), off, len, size);
	if (fi->fmode & CEPH_FILE_MODE_LAZY)
		want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
	else
		want = CEPH_CAP_FILE_BUFFER;
1572 1573

	got = 0;
1574
	err = ceph_get_caps(ci, CEPH_CAP_FILE_WR, want, off + len,
1575
			    &got, NULL);
1576
	if (err < 0)
1577
		goto out_free;
1578

1579 1580 1581 1582 1583
	dout("page_mkwrite %p %llu~%zd got cap refs on %s\n",
	     inode, off, len, ceph_cap_string(got));

	/* Update time before taking page lock */
	file_update_time(vma->vm_file);
1584

1585 1586
	do {
		lock_page(page);
1587

1588 1589 1590 1591 1592 1593
		if ((off > size) || (page->mapping != inode->i_mapping)) {
			unlock_page(page);
			ret = VM_FAULT_NOPAGE;
			break;
		}

1594 1595
		err = ceph_update_writeable_page(vma->vm_file, off, len, page);
		if (err >= 0) {
1596 1597 1598 1599
			/* success.  we'll keep the page locked. */
			set_page_dirty(page);
			ret = VM_FAULT_LOCKED;
		}
1600
	} while (err == -EAGAIN);
1601

1602 1603
	if (ret == VM_FAULT_LOCKED ||
	    ci->i_inline_version != CEPH_INLINE_NONE) {
1604 1605
		int dirty;
		spin_lock(&ci->i_ceph_lock);
1606
		ci->i_inline_version = CEPH_INLINE_NONE;
1607 1608
		dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
					       &prealloc_cf);
1609 1610 1611 1612 1613
		spin_unlock(&ci->i_ceph_lock);
		if (dirty)
			__mark_inode_dirty(inode, dirty);
	}

1614
	dout("page_mkwrite %p %llu~%zd dropping cap refs on %s ret %x\n",
1615 1616
	     inode, off, len, ceph_cap_string(got), ret);
	ceph_put_cap_refs(ci, got);
1617
out_free:
1618
	ceph_restore_sigs(&oldset);
1619
	ceph_free_cap_flush(prealloc_cf);
1620 1621
	if (err < 0)
		ret = vmf_error(err);
S
Sage Weil 已提交
1622 1623 1624
	return ret;
}

Y
Yan, Zheng 已提交
1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636
void ceph_fill_inline_data(struct inode *inode, struct page *locked_page,
			   char	*data, size_t len)
{
	struct address_space *mapping = inode->i_mapping;
	struct page *page;

	if (locked_page) {
		page = locked_page;
	} else {
		if (i_size_read(inode) == 0)
			return;
		page = find_or_create_page(mapping, 0,
1637 1638
					   mapping_gfp_constraint(mapping,
					   ~__GFP_FS));
Y
Yan, Zheng 已提交
1639 1640 1641 1642
		if (!page)
			return;
		if (PageUptodate(page)) {
			unlock_page(page);
1643
			put_page(page);
Y
Yan, Zheng 已提交
1644 1645 1646 1647
			return;
		}
	}

1648
	dout("fill_inline_data %p %llx.%llx len %zu locked_page %p\n",
Y
Yan, Zheng 已提交
1649 1650 1651 1652 1653 1654 1655 1656 1657
	     inode, ceph_vinop(inode), len, locked_page);

	if (len > 0) {
		void *kaddr = kmap_atomic(page);
		memcpy(kaddr, data, len);
		kunmap_atomic(kaddr);
	}

	if (page != locked_page) {
1658 1659
		if (len < PAGE_SIZE)
			zero_user_segment(page, len, PAGE_SIZE);
Y
Yan, Zheng 已提交
1660 1661 1662 1663 1664
		else
			flush_dcache_page(page);

		SetPageUptodate(page);
		unlock_page(page);
1665
		put_page(page);
Y
Yan, Zheng 已提交
1666 1667 1668
	}
}

1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701
int ceph_uninline_data(struct file *filp, struct page *locked_page)
{
	struct inode *inode = file_inode(filp);
	struct ceph_inode_info *ci = ceph_inode(inode);
	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
	struct ceph_osd_request *req;
	struct page *page = NULL;
	u64 len, inline_version;
	int err = 0;
	bool from_pagecache = false;

	spin_lock(&ci->i_ceph_lock);
	inline_version = ci->i_inline_version;
	spin_unlock(&ci->i_ceph_lock);

	dout("uninline_data %p %llx.%llx inline_version %llu\n",
	     inode, ceph_vinop(inode), inline_version);

	if (inline_version == 1 || /* initial version, no data */
	    inline_version == CEPH_INLINE_NONE)
		goto out;

	if (locked_page) {
		page = locked_page;
		WARN_ON(!PageUptodate(page));
	} else if (ceph_caps_issued(ci) &
		   (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) {
		page = find_get_page(inode->i_mapping, 0);
		if (page) {
			if (PageUptodate(page)) {
				from_pagecache = true;
				lock_page(page);
			} else {
1702
				put_page(page);
1703 1704 1705 1706 1707 1708 1709
				page = NULL;
			}
		}
	}

	if (page) {
		len = i_size_read(inode);
1710 1711
		if (len > PAGE_SIZE)
			len = PAGE_SIZE;
1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730
	} else {
		page = __page_cache_alloc(GFP_NOFS);
		if (!page) {
			err = -ENOMEM;
			goto out;
		}
		err = __ceph_do_getattr(inode, page,
					CEPH_STAT_CAP_INLINE_DATA, true);
		if (err < 0) {
			/* no inline data */
			if (err == -ENODATA)
				err = 0;
			goto out;
		}
		len = err;
	}

	req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
				    ceph_vino(inode), 0, &len, 0, 1,
1731
				    CEPH_OSD_OP_CREATE, CEPH_OSD_FLAG_WRITE,
I
Ilya Dryomov 已提交
1732
				    NULL, 0, 0, false);
1733 1734 1735 1736 1737
	if (IS_ERR(req)) {
		err = PTR_ERR(req);
		goto out;
	}

1738
	req->r_mtime = inode->i_mtime;
1739 1740 1741 1742 1743 1744 1745 1746 1747
	err = ceph_osdc_start_request(&fsc->client->osdc, req, false);
	if (!err)
		err = ceph_osdc_wait_request(&fsc->client->osdc, req);
	ceph_osdc_put_request(req);
	if (err < 0)
		goto out;

	req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
				    ceph_vino(inode), 0, &len, 1, 3,
1748
				    CEPH_OSD_OP_WRITE, CEPH_OSD_FLAG_WRITE,
I
Ilya Dryomov 已提交
1749 1750
				    NULL, ci->i_truncate_seq,
				    ci->i_truncate_size, false);
1751 1752 1753 1754 1755 1756 1757
	if (IS_ERR(req)) {
		err = PTR_ERR(req);
		goto out;
	}

	osd_req_op_extent_osd_data_pages(req, 1, &page, len, 0, false, false);

Y
Yan, Zheng 已提交
1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771 1772 1773 1774 1775 1776 1777 1778
	{
		__le64 xattr_buf = cpu_to_le64(inline_version);
		err = osd_req_op_xattr_init(req, 0, CEPH_OSD_OP_CMPXATTR,
					    "inline_version", &xattr_buf,
					    sizeof(xattr_buf),
					    CEPH_OSD_CMPXATTR_OP_GT,
					    CEPH_OSD_CMPXATTR_MODE_U64);
		if (err)
			goto out_put;
	}

	{
		char xattr_buf[32];
		int xattr_len = snprintf(xattr_buf, sizeof(xattr_buf),
					 "%llu", inline_version);
		err = osd_req_op_xattr_init(req, 2, CEPH_OSD_OP_SETXATTR,
					    "inline_version",
					    xattr_buf, xattr_len, 0, 0);
		if (err)
			goto out_put;
	}
1779

1780
	req->r_mtime = inode->i_mtime;
1781 1782 1783 1784 1785 1786 1787 1788 1789 1790 1791
	err = ceph_osdc_start_request(&fsc->client->osdc, req, false);
	if (!err)
		err = ceph_osdc_wait_request(&fsc->client->osdc, req);
out_put:
	ceph_osdc_put_request(req);
	if (err == -ECANCELED)
		err = 0;
out:
	if (page && page != locked_page) {
		if (from_pagecache) {
			unlock_page(page);
1792
			put_page(page);
1793 1794 1795 1796 1797 1798 1799 1800 1801
		} else
			__free_pages(page, 0);
	}

	dout("uninline_data %p %llx.%llx inline_version %llu = %d\n",
	     inode, ceph_vinop(inode), inline_version, err);
	return err;
}

1802
static const struct vm_operations_struct ceph_vmops = {
1803
	.fault		= ceph_filemap_fault,
S
Sage Weil 已提交
1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815 1816
	.page_mkwrite	= ceph_page_mkwrite,
};

int ceph_mmap(struct file *file, struct vm_area_struct *vma)
{
	struct address_space *mapping = file->f_mapping;

	if (!mapping->a_ops->readpage)
		return -ENOEXEC;
	file_accessed(file);
	vma->vm_ops = &ceph_vmops;
	return 0;
}
1817 1818 1819 1820 1821 1822

enum {
	POOL_READ	= 1,
	POOL_WRITE	= 2,
};

Y
Yan, Zheng 已提交
1823 1824
static int __ceph_pool_perm_get(struct ceph_inode_info *ci,
				s64 pool, struct ceph_string *pool_ns)
1825 1826 1827 1828 1829 1830 1831
{
	struct ceph_fs_client *fsc = ceph_inode_to_client(&ci->vfs_inode);
	struct ceph_mds_client *mdsc = fsc->mdsc;
	struct ceph_osd_request *rd_req = NULL, *wr_req = NULL;
	struct rb_node **p, *parent;
	struct ceph_pool_perm *perm;
	struct page **pages;
Y
Yan, Zheng 已提交
1832
	size_t pool_ns_len;
1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843
	int err = 0, err2 = 0, have = 0;

	down_read(&mdsc->pool_perm_rwsem);
	p = &mdsc->pool_perm_tree.rb_node;
	while (*p) {
		perm = rb_entry(*p, struct ceph_pool_perm, node);
		if (pool < perm->pool)
			p = &(*p)->rb_left;
		else if (pool > perm->pool)
			p = &(*p)->rb_right;
		else {
Y
Yan, Zheng 已提交
1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854
			int ret = ceph_compare_string(pool_ns,
						perm->pool_ns,
						perm->pool_ns_len);
			if (ret < 0)
				p = &(*p)->rb_left;
			else if (ret > 0)
				p = &(*p)->rb_right;
			else {
				have = perm->perm;
				break;
			}
1855 1856 1857 1858 1859 1860
		}
	}
	up_read(&mdsc->pool_perm_rwsem);
	if (*p)
		goto out;

Y
Yan, Zheng 已提交
1861 1862 1863 1864 1865
	if (pool_ns)
		dout("__ceph_pool_perm_get pool %lld ns %.*s no perm cached\n",
		     pool, (int)pool_ns->len, pool_ns->str);
	else
		dout("__ceph_pool_perm_get pool %lld no perm cached\n", pool);
1866 1867

	down_write(&mdsc->pool_perm_rwsem);
Y
Yan, Zheng 已提交
1868
	p = &mdsc->pool_perm_tree.rb_node;
1869 1870 1871 1872 1873 1874 1875 1876 1877
	parent = NULL;
	while (*p) {
		parent = *p;
		perm = rb_entry(parent, struct ceph_pool_perm, node);
		if (pool < perm->pool)
			p = &(*p)->rb_left;
		else if (pool > perm->pool)
			p = &(*p)->rb_right;
		else {
Y
Yan, Zheng 已提交
1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888
			int ret = ceph_compare_string(pool_ns,
						perm->pool_ns,
						perm->pool_ns_len);
			if (ret < 0)
				p = &(*p)->rb_left;
			else if (ret > 0)
				p = &(*p)->rb_right;
			else {
				have = perm->perm;
				break;
			}
1889 1890 1891 1892 1893 1894 1895
		}
	}
	if (*p) {
		up_write(&mdsc->pool_perm_rwsem);
		goto out;
	}

I
Ilya Dryomov 已提交
1896
	rd_req = ceph_osdc_alloc_request(&fsc->client->osdc, NULL,
1897 1898 1899 1900 1901 1902 1903 1904 1905
					 1, false, GFP_NOFS);
	if (!rd_req) {
		err = -ENOMEM;
		goto out_unlock;
	}

	rd_req->r_flags = CEPH_OSD_FLAG_READ;
	osd_req_op_init(rd_req, 0, CEPH_OSD_OP_STAT, 0);
	rd_req->r_base_oloc.pool = pool;
Y
Yan, Zheng 已提交
1906 1907
	if (pool_ns)
		rd_req->r_base_oloc.pool_ns = ceph_get_string(pool_ns);
1908
	ceph_oid_printf(&rd_req->r_base_oid, "%llx.00000000", ci->i_vino.ino);
1909

1910 1911 1912
	err = ceph_osdc_alloc_messages(rd_req, GFP_NOFS);
	if (err)
		goto out_unlock;
1913

I
Ilya Dryomov 已提交
1914
	wr_req = ceph_osdc_alloc_request(&fsc->client->osdc, NULL,
1915 1916 1917 1918 1919 1920
					 1, false, GFP_NOFS);
	if (!wr_req) {
		err = -ENOMEM;
		goto out_unlock;
	}

1921
	wr_req->r_flags = CEPH_OSD_FLAG_WRITE;
1922
	osd_req_op_init(wr_req, 0, CEPH_OSD_OP_CREATE, CEPH_OSD_OP_FLAG_EXCL);
1923
	ceph_oloc_copy(&wr_req->r_base_oloc, &rd_req->r_base_oloc);
1924
	ceph_oid_copy(&wr_req->r_base_oid, &rd_req->r_base_oid);
1925

1926 1927 1928
	err = ceph_osdc_alloc_messages(wr_req, GFP_NOFS);
	if (err)
		goto out_unlock;
1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940

	/* one page should be large enough for STAT data */
	pages = ceph_alloc_page_vector(1, GFP_KERNEL);
	if (IS_ERR(pages)) {
		err = PTR_ERR(pages);
		goto out_unlock;
	}

	osd_req_op_raw_data_in_pages(rd_req, 0, pages, PAGE_SIZE,
				     0, false, true);
	err = ceph_osdc_start_request(&fsc->client->osdc, rd_req, false);

1941
	wr_req->r_mtime = ci->vfs_inode.i_mtime;
1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960
	err2 = ceph_osdc_start_request(&fsc->client->osdc, wr_req, false);

	if (!err)
		err = ceph_osdc_wait_request(&fsc->client->osdc, rd_req);
	if (!err2)
		err2 = ceph_osdc_wait_request(&fsc->client->osdc, wr_req);

	if (err >= 0 || err == -ENOENT)
		have |= POOL_READ;
	else if (err != -EPERM)
		goto out_unlock;

	if (err2 == 0 || err2 == -EEXIST)
		have |= POOL_WRITE;
	else if (err2 != -EPERM) {
		err = err2;
		goto out_unlock;
	}

Y
Yan, Zheng 已提交
1961 1962
	pool_ns_len = pool_ns ? pool_ns->len : 0;
	perm = kmalloc(sizeof(*perm) + pool_ns_len + 1, GFP_NOFS);
1963 1964 1965 1966 1967 1968 1969
	if (!perm) {
		err = -ENOMEM;
		goto out_unlock;
	}

	perm->pool = pool;
	perm->perm = have;
Y
Yan, Zheng 已提交
1970 1971 1972 1973 1974
	perm->pool_ns_len = pool_ns_len;
	if (pool_ns_len > 0)
		memcpy(perm->pool_ns, pool_ns->str, pool_ns_len);
	perm->pool_ns[pool_ns_len] = 0;

1975 1976 1977 1978 1979 1980
	rb_link_node(&perm->node, parent, p);
	rb_insert_color(&perm->node, &mdsc->pool_perm_tree);
	err = 0;
out_unlock:
	up_write(&mdsc->pool_perm_rwsem);

1981 1982
	ceph_osdc_put_request(rd_req);
	ceph_osdc_put_request(wr_req);
1983 1984 1985
out:
	if (!err)
		err = have;
Y
Yan, Zheng 已提交
1986 1987 1988 1989 1990
	if (pool_ns)
		dout("__ceph_pool_perm_get pool %lld ns %.*s result = %d\n",
		     pool, (int)pool_ns->len, pool_ns->str, err);
	else
		dout("__ceph_pool_perm_get pool %lld result = %d\n", pool, err);
1991 1992 1993 1994 1995
	return err;
}

int ceph_pool_perm_check(struct ceph_inode_info *ci, int need)
{
1996
	s64 pool;
Y
Yan, Zheng 已提交
1997
	struct ceph_string *pool_ns;
1998 1999
	int ret, flags;

2000 2001 2002 2003 2004 2005 2006 2007 2008
	if (ci->i_vino.snap != CEPH_NOSNAP) {
		/*
		 * Pool permission check needs to write to the first object.
		 * But for snapshot, head of the first object may have alread
		 * been deleted. Skip check to avoid creating orphan object.
		 */
		return 0;
	}

2009 2010 2011 2012 2013 2014
	if (ceph_test_mount_opt(ceph_inode_to_client(&ci->vfs_inode),
				NOPOOLPERM))
		return 0;

	spin_lock(&ci->i_ceph_lock);
	flags = ci->i_ceph_flags;
2015
	pool = ci->i_layout.pool_id;
2016 2017 2018 2019
	spin_unlock(&ci->i_ceph_lock);
check:
	if (flags & CEPH_I_POOL_PERM) {
		if ((need & CEPH_CAP_FILE_RD) && !(flags & CEPH_I_POOL_RD)) {
2020
			dout("ceph_pool_perm_check pool %lld no read perm\n",
2021 2022 2023 2024
			     pool);
			return -EPERM;
		}
		if ((need & CEPH_CAP_FILE_WR) && !(flags & CEPH_I_POOL_WR)) {
2025
			dout("ceph_pool_perm_check pool %lld no write perm\n",
2026 2027 2028 2029 2030 2031
			     pool);
			return -EPERM;
		}
		return 0;
	}

Y
Yan, Zheng 已提交
2032 2033 2034
	pool_ns = ceph_try_get_string(ci->i_layout.pool_ns);
	ret = __ceph_pool_perm_get(ci, pool, pool_ns);
	ceph_put_string(pool_ns);
2035 2036 2037 2038 2039 2040 2041 2042 2043 2044
	if (ret < 0)
		return ret;

	flags = CEPH_I_POOL_PERM;
	if (ret & POOL_READ)
		flags |= CEPH_I_POOL_RD;
	if (ret & POOL_WRITE)
		flags |= CEPH_I_POOL_WR;

	spin_lock(&ci->i_ceph_lock);
Y
Yan, Zheng 已提交
2045 2046 2047
	if (pool == ci->i_layout.pool_id &&
	    pool_ns == rcu_dereference_raw(ci->i_layout.pool_ns)) {
		ci->i_ceph_flags |= flags;
2048
        } else {
2049
		pool = ci->i_layout.pool_id;
2050 2051 2052 2053 2054 2055 2056 2057 2058 2059 2060 2061 2062 2063 2064 2065 2066 2067
		flags = ci->i_ceph_flags;
	}
	spin_unlock(&ci->i_ceph_lock);
	goto check;
}

void ceph_pool_perm_destroy(struct ceph_mds_client *mdsc)
{
	struct ceph_pool_perm *perm;
	struct rb_node *n;

	while (!RB_EMPTY_ROOT(&mdsc->pool_perm_tree)) {
		n = rb_first(&mdsc->pool_perm_tree);
		perm = rb_entry(n, struct ceph_pool_perm, node);
		rb_erase(n, &mdsc->pool_perm_tree);
		kfree(perm);
	}
}