addr.c 52.8 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>
S
Sage Weil 已提交
18 19 20 21 22 23 24 25 26 27 28 29

/*
 * 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 已提交
30
 * count dirty pages on the inode.  In the absence of snapshots,
S
Sage Weil 已提交
31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54
 * 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 已提交
55 56 57 58 59
#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))

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

/*
 * 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;
77
	int ret;
S
Sage Weil 已提交
78 79 80 81

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

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

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

	/* dirty the head */
93
	spin_lock(&ci->i_ceph_lock);
94 95 96 97 98 99 100 101 102 103 104 105 106
	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 已提交
107
	if (ci->i_wrbuffer_ref == 0)
108
		ihold(inode);
S
Sage Weil 已提交
109 110 111 112 113 114 115
	++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);
116
	spin_unlock(&ci->i_ceph_lock);
S
Sage Weil 已提交
117

118 119 120 121 122 123 124
	/*
	 * 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 已提交
125

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

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

/*
 * 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.
 */
138 139
static void ceph_invalidatepage(struct page *page, unsigned int offset,
				unsigned int length)
S
Sage Weil 已提交
140
{
141
	struct inode *inode;
S
Sage Weil 已提交
142
	struct ceph_inode_info *ci;
143
	struct ceph_snap_context *snapc = page_snap_context(page);
S
Sage Weil 已提交
144

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

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

154 155
	ceph_invalidate_fscache_page(inode, page);

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

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

162 163 164 165 166 167 168
	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 已提交
169 170 171 172
}

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

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

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

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

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

202 203 204 205 206 207 208
	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;
209
		zero_user_segment(page, 0, PAGE_SIZE);
210 211 212
		SetPageUptodate(page);
		return 0;
	}
Y
Yan, Zheng 已提交
213 214

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

S
Sage Weil 已提交
218 219 220
	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 已提交
221
				  off, &len,
S
Sage Weil 已提交
222
				  ci->i_truncate_seq, ci->i_truncate_size,
223
				  &page, 1, 0);
S
Sage Weil 已提交
224 225 226 227
	if (err == -ENOENT)
		err = 0;
	if (err < 0) {
		SetPageError(page);
228
		ceph_fscache_readpage_cancel(inode, page);
S
Sage Weil 已提交
229 230
		goto out;
	}
231
	if (err < PAGE_SIZE)
232
		/* zero fill remainder of page */
233
		zero_user_segment(page, err, PAGE_SIZE);
234 235
	else
		flush_dcache_page(page);
S
Sage Weil 已提交
236

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

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

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

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

S
Sage Weil 已提交
266 267 268
	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 */
269
	osd_data = osd_req_op_extent_osd_data(req, 0);
270 271 272
	BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGES);
	num_pages = calc_pages_for((u64)osd_data->alignment,
					(u64)osd_data->length);
273
	for (i = 0; i < num_pages; i++) {
274
		struct page *page = osd_data->pages[i];
S
Sage Weil 已提交
275

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

/*
S
Sage Weil 已提交
299 300
 * start an async read(ahead) operation.  return nr_pages we submitted
 * a read for on success, or negative error code.
S
Sage Weil 已提交
301
 */
302
static int start_read(struct inode *inode, struct list_head *page_list, int max)
S
Sage Weil 已提交
303
{
304 305
	struct ceph_osd_client *osdc =
		&ceph_inode_to_client(inode)->client->osdc;
S
Sage Weil 已提交
306 307
	struct ceph_inode_info *ci = ceph_inode(inode);
	struct page *page = list_entry(page_list->prev, struct page, lru);
308
	struct ceph_vino vino;
S
Sage Weil 已提交
309 310
	struct ceph_osd_request *req;
	u64 off;
S
Sage Weil 已提交
311
	u64 len;
S
Sage Weil 已提交
312 313 314 315
	int i;
	struct page **pages;
	pgoff_t next_index;
	int nr_pages = 0;
316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341
	int got = 0;
	int ret = 0;

	if (!current->journal_info) {
		/* 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 已提交
342

343
	off = (u64) page_offset(page);
S
Sage Weil 已提交
344

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

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

S
Sage Weil 已提交
381 382 383
 		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,
384
					  GFP_KERNEL)) {
385
			ceph_fscache_uncache_page(inode, page);
386
			put_page(page);
S
Sage Weil 已提交
387
			dout("start_read %p add_to_page_cache failed %p\n",
S
Sage Weil 已提交
388
			     inode, page);
S
Sage Weil 已提交
389
			nr_pages = i;
390 391
			if (nr_pages > 0) {
				len = nr_pages << PAGE_SHIFT;
392
				osd_req_op_extent_update(req, 0, len);
393 394
				break;
			}
S
Sage Weil 已提交
395
			goto out_pages;
S
Sage Weil 已提交
396
		}
S
Sage Weil 已提交
397
		pages[i] = page;
S
Sage Weil 已提交
398
	}
399
	osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0, false, false);
S
Sage Weil 已提交
400 401 402 403 404 405 406 407
	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);
408 409 410 411 412 413

	/* 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 已提交
414 415 416
	return nr_pages;

out_pages:
417 418 419 420 421
	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);
422
out_put:
S
Sage Weil 已提交
423
	ceph_osdc_put_request(req);
424 425 426
out:
	if (got)
		ceph_put_cap_refs(ci, got);
S
Sage Weil 已提交
427 428
	return ret;
}
S
Sage Weil 已提交
429

S
Sage Weil 已提交
430 431 432 433 434 435 436 437

/*
 * 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 已提交
438
	struct inode *inode = file_inode(file);
439
	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
S
Sage Weil 已提交
440
	int rc = 0;
441 442
	int max = 0;

Y
Yan, Zheng 已提交
443 444 445
	if (ceph_inode(inode)->i_inline_version != CEPH_INLINE_NONE)
		return -EINVAL;

446 447 448 449 450 451
	rc = ceph_readpages_from_fscache(mapping->host, mapping, page_list,
					 &nr_pages);

	if (rc == 0)
		goto out;

452 453 454
	max = fsc->mount_options->rsize >> PAGE_SHIFT;
	dout("readpages %p file %p nr_pages %d max %d\n",
	     inode, file, nr_pages, max);
S
Sage Weil 已提交
455
	while (!list_empty(page_list)) {
456
		rc = start_read(inode, page_list, max);
S
Sage Weil 已提交
457 458 459
		if (rc < 0)
			goto out;
	}
S
Sage Weil 已提交
460
out:
461 462
	ceph_fscache_readpages_cancel(inode, page_list);

S
Sage Weil 已提交
463
	dout("readpages %p file %p ret %d\n", inode, file, rc);
S
Sage Weil 已提交
464 465 466
	return rc;
}

467 468 469 470 471 472
struct ceph_writeback_ctl
{
	loff_t i_size;
	u64 truncate_size;
	u32 truncate_seq;
	bool size_stable;
473
	bool head_snapc;
474 475
};

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

488
	spin_lock(&ci->i_ceph_lock);
S
Sage Weil 已提交
489 490 491
	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);
492 493 494 495 496 497 498 499 500 501 502 503 504 505
		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;
506
			}
507 508
			ctl->truncate_size = capsnap->truncate_size;
			ctl->truncate_seq = capsnap->truncate_seq;
509
			ctl->head_snapc = false;
S
Sage Weil 已提交
510
		}
511 512 513 514 515 516 517 518 519

		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 已提交
520
	}
521
	if (!snapc && ci->i_wrbuffer_ref_head) {
522
		snapc = ceph_get_snap_context(ci->i_head_snapc);
S
Sage Weil 已提交
523 524
		dout(" head snapc %p has %d dirty pages\n",
		     snapc, ci->i_wrbuffer_ref_head);
525 526 527 528 529
		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;
530
			ctl->head_snapc = true;
531
		}
S
Sage Weil 已提交
532
	}
533
	spin_unlock(&ci->i_ceph_lock);
S
Sage Weil 已提交
534 535 536
	return snapc;
}

537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563
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 已提交
564 565 566 567 568 569 570 571 572 573
/*
 * 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;
574
	struct ceph_fs_client *fsc;
575
	struct ceph_snap_context *snapc, *oldest;
576
	loff_t page_off = page_offset(page);
Y
Yehuda Sadeh 已提交
577
	long writeback_stat;
Y
Yan, Zheng 已提交
578
	int err, len = PAGE_SIZE;
579
	struct ceph_writeback_ctl ceph_wbc;
S
Sage Weil 已提交
580 581 582 583 584

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

	inode = page->mapping->host;
	ci = ceph_inode(inode);
585
	fsc = ceph_inode_to_client(inode);
S
Sage Weil 已提交
586 587

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

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

612 613
	if (ceph_wbc.i_size < page_off + len)
		len = ceph_wbc.i_size - page_off;
S
Sage Weil 已提交
614

615 616
	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 已提交
617

618
	writeback_stat = atomic_long_inc_return(&fsc->writeback_count);
Y
Yehuda Sadeh 已提交
619
	if (writeback_stat >
620
	    CONGESTION_ON_THRESH(fsc->mount_options->congestion_kb))
621
		set_bdi_congested(inode_to_bdi(inode), BLK_RW_ASYNC);
Y
Yehuda Sadeh 已提交
622

S
Sage Weil 已提交
623
	set_page_writeback(page);
624 625 626 627
	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,
628
				   &inode->i_mtime, &page, 1);
S
Sage Weil 已提交
629
	if (err < 0) {
630 631 632 633 634 635 636 637
		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 已提交
638
			return err;
639 640 641
		}
		dout("writepage setting page/mapping error %d %p\n",
		     err, page);
S
Sage Weil 已提交
642 643
		SetPageError(page);
		mapping_set_error(&inode->i_data, err);
644
		wbc->pages_skipped++;
S
Sage Weil 已提交
645 646 647 648 649 650 651 652
	} 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);
653
	ceph_put_snap_context(snapc);  /* page's reference */
S
Sage Weil 已提交
654 655 656 657 658
	return err;
}

static int ceph_writepage(struct page *page, struct writeback_control *wbc)
{
659 660 661
	int err;
	struct inode *inode = page->mapping->host;
	BUG_ON(!inode);
662
	ihold(inode);
663
	err = writepage_nounlock(page, wbc);
664 665 666 667 668
	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 已提交
669
	unlock_page(page);
670
	iput(inode);
S
Sage Weil 已提交
671 672 673 674 675 676 677 678 679 680 681 682
	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;

683
	pagevec_init(&pvec);
S
Sage Weil 已提交
684 685 686 687 688 689 690 691 692 693 694 695 696
	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.
 */
697
static void writepages_finish(struct ceph_osd_request *req)
S
Sage Weil 已提交
698 699 700
{
	struct inode *inode = req->r_inode;
	struct ceph_inode_info *ci = ceph_inode(inode);
701
	struct ceph_osd_data *osd_data;
S
Sage Weil 已提交
702
	struct page *page;
Y
Yan, Zheng 已提交
703 704 705
	int num_pages, total_pages = 0;
	int i, j;
	int rc = req->r_result;
S
Sage Weil 已提交
706 707
	struct ceph_snap_context *snapc = req->r_snapc;
	struct address_space *mapping = inode->i_mapping;
708
	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
Y
Yan, Zheng 已提交
709
	bool remove_page;
S
Sage Weil 已提交
710

Y
Yan, Zheng 已提交
711
	dout("writepages_finish %p rc %d\n", inode, rc);
712
	if (rc < 0) {
S
Sage Weil 已提交
713
		mapping_set_error(mapping, rc);
714 715 716 717
		ceph_set_error_write(ci);
	} else {
		ceph_clear_error_write(ci);
	}
Y
Yan, Zheng 已提交
718 719 720 721 722 723 724 725 726

	/*
	 * 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 已提交
727 728

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

Y
Yan, Zheng 已提交
733 734 735 736 737 738 739 740 741 742 743 744 745
		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))
746
				clear_bdi_congested(inode_to_bdi(inode),
Y
Yan, Zheng 已提交
747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762
						    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);
763

Y
Yan, Zheng 已提交
764
		ceph_release_pages(osd_data->pages, num_pages);
S
Sage Weil 已提交
765 766
	}

Y
Yan, Zheng 已提交
767 768 769
	ceph_put_wrbuffer_cap_refs(ci, total_pages, snapc);

	osd_data = osd_req_op_extent_osd_data(req, 0);
770 771
	if (osd_data->pages_from_pool)
		mempool_free(osd_data->pages,
772
			     ceph_sb_to_client(inode->i_sb)->wb_pagevec_pool);
S
Sage Weil 已提交
773
	else
774
		kfree(osd_data->pages);
S
Sage Weil 已提交
775 776 777 778 779 780 781 782 783 784 785
	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);
786 787
	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
	struct ceph_vino vino = ceph_vino(inode);
788
	pgoff_t index, start_index, end = -1;
789
	struct ceph_snap_context *snapc = NULL, *last_snapc = NULL, *pgsnapc;
S
Sage Weil 已提交
790 791
	struct pagevec pvec;
	int rc = 0;
F
Fabian Frederick 已提交
792
	unsigned int wsize = i_blocksize(inode);
S
Sage Weil 已提交
793
	struct ceph_osd_request *req = NULL;
794
	struct ceph_writeback_ctl ceph_wbc;
795 796
	bool should_loop, range_whole = false;
	bool stop, done = false;
S
Sage Weil 已提交
797

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

802
	if (READ_ONCE(fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
803 804 805 806 807
		if (ci->i_wrbuffer_ref > 0) {
			pr_warn_ratelimited(
				"writepage_start %p %lld forced umount\n",
				inode, ceph_ino(inode));
		}
808
		mapping_set_error(mapping, -EIO);
S
Sage Weil 已提交
809 810
		return -EIO; /* we're in a forced umount, don't write! */
	}
Y
Yan, Zheng 已提交
811
	if (fsc->mount_options->wsize < wsize)
812
		wsize = fsc->mount_options->wsize;
S
Sage Weil 已提交
813

814
	pagevec_init(&pvec);
S
Sage Weil 已提交
815

816
	start_index = wbc->range_cyclic ? mapping->writeback_index : 0;
817
	index = start_index;
S
Sage Weil 已提交
818 819 820

retry:
	/* find oldest snap context with dirty data */
821
	snapc = get_oldest_context(inode, &ceph_wbc, NULL);
S
Sage Weil 已提交
822 823 824 825 826 827 828 829
	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);
830

831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854
	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 */
		if (index > 0 || wbc->sync_mode != WB_SYNC_NONE)
			should_loop = true;
		dout(" non-head snapc, range whole\n");
S
Sage Weil 已提交
855
	}
856 857

	ceph_put_snap_context(last_snapc);
S
Sage Weil 已提交
858 859
	last_snapc = snapc;

860 861
	stop = false;
	while (!stop && index <= end) {
Y
Yan, Zheng 已提交
862
		int num_ops = 0, op_idx;
863
		unsigned i, pvec_pages, max_pages, locked_pages = 0;
Y
Yan, Zheng 已提交
864
		struct page **pages = NULL, **data_pages;
865
		mempool_t *pool = NULL;	/* Becomes non-null if mempool used */
S
Sage Weil 已提交
866
		struct page *page;
867
		pgoff_t strip_unit_end = 0;
Y
Yan, Zheng 已提交
868
		u64 offset = 0, len = 0;
S
Sage Weil 已提交
869

870
		max_pages = wsize >> PAGE_SHIFT;
S
Sage Weil 已提交
871 872

get_more_pages:
J
Jan Kara 已提交
873
		pvec_pages = pagevec_lookup_range_nr_tag(&pvec, mapping, &index,
J
Jan Kara 已提交
874
						end, PAGECACHE_TAG_DIRTY,
J
Jan Kara 已提交
875
						max_pages - locked_pages);
J
Jan Kara 已提交
876
		dout("pagevec_lookup_range_tag got %d\n", pvec_pages);
S
Sage Weil 已提交
877 878 879 880 881 882 883 884 885 886 887 888 889 890 891
		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);
892
				continue;
S
Sage Weil 已提交
893
			}
Y
Yan, Zheng 已提交
894 895
			if (strip_unit_end && (page->index > strip_unit_end)) {
				dout("end of strip unit %p\n", page);
S
Sage Weil 已提交
896 897 898
				unlock_page(page);
				break;
			}
899 900 901
			if (page_offset(page) >= ceph_wbc.i_size) {
				dout("%p page eof %llu\n",
				     page, ceph_wbc.i_size);
902 903
				/* not done if range_cyclic */
				stop = true;
S
Sage Weil 已提交
904 905 906 907
				unlock_page(page);
				break;
			}
			if (PageWriteback(page)) {
908 909 910 911 912 913 914
				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 已提交
915 916 917
			}

			/* only if matching snap context */
918
			pgsnapc = page_snap_context(page);
919 920
			if (pgsnapc != snapc) {
				dout("page snapc %p %lld != oldest %p %lld\n",
921
				     pgsnapc, pgsnapc->seq, snapc, snapc->seq);
S
Sage Weil 已提交
922
				unlock_page(page);
923
				continue;
S
Sage Weil 已提交
924 925 926 927 928
			}

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

932 933 934
			/*
			 * We have something to write.  If this is
			 * the first locked page this time through,
Y
Yan, Zheng 已提交
935 936
			 * calculate max possinle write size and
			 * allocate a page array
937
			 */
S
Sage Weil 已提交
938
			if (locked_pages == 0) {
Y
Yan, Zheng 已提交
939 940 941
				u64 objnum;
				u64 objoff;

S
Sage Weil 已提交
942
				/* prepare async write request */
943
				offset = (u64)page_offset(page);
S
Sage Weil 已提交
944
				len = wsize;
Y
Yan, Zheng 已提交
945 946 947 948 949 950

				rc = ceph_calc_file_object_mapping(&ci->i_layout,
								offset, len,
								&objnum, &objoff,
								&len);
				if (rc < 0) {
951 952 953 954
					unlock_page(page);
					break;
				}

Y
Yanhu Cao 已提交
955
				num_ops = 1;
Y
Yan, Zheng 已提交
956
				strip_unit_end = page->index +
957
					((len - 1) >> PAGE_SHIFT);
A
Alex Elder 已提交
958

Y
Yan, Zheng 已提交
959
				BUG_ON(pages);
A
Alex Elder 已提交
960
				max_pages = calc_pages_for(0, (u64)len);
961 962
				pages = kmalloc(max_pages * sizeof (*pages),
						GFP_NOFS);
A
Alex Elder 已提交
963 964 965
				if (!pages) {
					pool = fsc->wb_pagevec_pool;
					pages = mempool_alloc(pool, GFP_NOFS);
966
					BUG_ON(!pages);
A
Alex Elder 已提交
967
				}
Y
Yan, Zheng 已提交
968 969 970

				len = 0;
			} else if (page->index !=
971
				   (offset + len) >> PAGE_SHIFT) {
Y
Yan, Zheng 已提交
972 973 974 975 976 977 978 979 980 981
				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 已提交
982 983 984 985 986
			}

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

Y
Yan, Zheng 已提交
988 989
			if (atomic_long_inc_return(&fsc->writeback_count) >
			    CONGESTION_ON_THRESH(
990
				    fsc->mount_options->congestion_kb)) {
991
				set_bdi_congested(inode_to_bdi(inode),
S
Sage Weil 已提交
992
						  BLK_RW_ASYNC);
Y
Yehuda Sadeh 已提交
993 994
			}

995 996 997 998

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

999
			len += PAGE_SIZE;
S
Sage Weil 已提交
1000 1001 1002 1003 1004 1005
		}

		/* did we get anything? */
		if (!locked_pages)
			goto release_pvec_pages;
		if (i) {
1006 1007 1008 1009 1010 1011 1012 1013 1014 1015
			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 已提交
1016 1017 1018 1019

			if (pvec_pages && i == pvec_pages &&
			    locked_pages < max_pages) {
				dout("reached end pvec, trying for more\n");
1020
				pagevec_release(&pvec);
S
Sage Weil 已提交
1021 1022 1023 1024
				goto get_more_pages;
			}
		}

Y
Yan, Zheng 已提交
1025
new_request:
1026
		offset = page_offset(pages[0]);
Y
Yan, Zheng 已提交
1027 1028 1029 1030 1031
		len = wsize;

		req = ceph_osdc_new_request(&fsc->client->osdc,
					&ci->i_layout, vino,
					offset, &len, 0, num_ops,
1032 1033 1034
					CEPH_OSD_OP_WRITE, CEPH_OSD_FLAG_WRITE,
					snapc, ceph_wbc.truncate_seq,
					ceph_wbc.truncate_size, false);
Y
Yan, Zheng 已提交
1035 1036 1037 1038 1039 1040 1041
		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,
1042
						CEPH_OSD_FLAG_WRITE,
1043 1044
						snapc, ceph_wbc.truncate_seq,
						ceph_wbc.truncate_size, true);
Y
Yan, Zheng 已提交
1045
			BUG_ON(IS_ERR(req));
Y
Yan, Zheng 已提交
1046
		}
Y
Yan, Zheng 已提交
1047
		BUG_ON(len < page_offset(pages[locked_pages - 1]) +
1048
			     PAGE_SIZE - offset);
Y
Yan, Zheng 已提交
1049 1050 1051

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

Y
Yan, Zheng 已提交
1053 1054 1055 1056 1057 1058 1059
		/* 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 已提交
1060
				if (op_idx + 1 == req->r_num_ops)
Y
Yan, Zheng 已提交
1061 1062 1063 1064 1065 1066 1067
					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,
1068
							!!pool, false);
Y
Yan, Zheng 已提交
1069
				osd_req_op_extent_update(req, op_idx, len);
1070

Y
Yan, Zheng 已提交
1071 1072 1073 1074 1075 1076 1077
				len = 0;
				offset = cur_offset; 
				data_pages = pages + i;
				op_idx++;
			}

			set_page_writeback(pages[i]);
1078
			len += PAGE_SIZE;
Y
Yan, Zheng 已提交
1079 1080
		}

1081 1082
		if (ceph_wbc.size_stable) {
			len = min(len, ceph_wbc.i_size - offset);
Y
Yan, Zheng 已提交
1083 1084 1085 1086
		} else if (i == locked_pages) {
			/* writepages_finish() clears writeback pages
			 * according to the data length, so make sure
			 * data length covers all locked pages */
1087
			u64 min_len = len + 1 - PAGE_SIZE;
1088 1089
			len = get_writepages_data_length(inode, pages[i - 1],
							 offset);
Y
Yan, Zheng 已提交
1090 1091 1092
			len = max(len, min_len);
		}
		dout("writepages got pages at %llu~%llu\n", offset, len);
1093

Y
Yan, Zheng 已提交
1094 1095 1096
		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);
1097

Y
Yan, Zheng 已提交
1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124
		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;
			pages = kmalloc(locked_pages * sizeof (*pages),
					GFP_NOFS);
			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;
		}
1125

1126
		req->r_mtime = inode->i_mtime;
1127 1128
		rc = ceph_osdc_start_request(&fsc->client->osdc, req, true);
		BUG_ON(rc);
S
Sage Weil 已提交
1129 1130
		req = NULL;

Y
Yan, Zheng 已提交
1131 1132 1133 1134
		wbc->nr_to_write -= i;
		if (pages)
			goto new_request;

1135 1136 1137 1138 1139 1140 1141 1142
		/*
		 * 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)
			done = stop = true;
S
Sage Weil 已提交
1143 1144 1145 1146 1147 1148 1149 1150 1151 1152

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");
1153
		end = start_index - 1; /* OK even when start_index == 0 */
1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164

		/* 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,
1165
						PAGECACHE_TAG_WRITEBACK))) {
1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176
				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();
			}
		}

1177
		start_index = 0;
S
Sage Weil 已提交
1178 1179 1180 1181 1182 1183 1184 1185
		index = 0;
		goto retry;
	}

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

out:
1186
	ceph_osdc_put_request(req);
1187 1188
	ceph_put_snap_context(last_snapc);
	dout("writepages dend - startone, rc = %d\n", rc);
S
Sage Weil 已提交
1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199
	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)
{
1200
	struct ceph_snap_context *oldest = get_oldest_context(inode, NULL, NULL);
1201 1202 1203 1204
	int ret = !oldest || snapc->seq <= oldest->seq;

	ceph_put_snap_context(oldest);
	return ret;
S
Sage Weil 已提交
1205 1206 1207 1208 1209
}

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

1229
	if (READ_ONCE(fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
1230 1231 1232 1233 1234
		dout(" page %p forced umount\n", page);
		unlock_page(page);
		return -EIO;
	}

S
Sage Weil 已提交
1235 1236 1237 1238
retry_locked:
	/* writepages currently holds page lock, but if we change that later, */
	wait_on_page_writeback(page);

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

		/* 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 已提交
1273
			goto fail_unlock;
S
Sage Weil 已提交
1274 1275 1276 1277 1278 1279 1280 1281 1282
		goto retry_locked;
	}

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

	/* full page? */
1283
	if (pos_in_page == 0 && len == PAGE_SIZE)
S
Sage Weil 已提交
1284 1285 1286
		return 0;

	/* past end of file? */
1287
	i_size = i_size_read(inode);
S
Sage Weil 已提交
1288 1289 1290

	if (page_off >= i_size ||
	    (pos_in_page == 0 && (pos+len) >= i_size &&
1291
	     end_in_page - pos_in_page != PAGE_SIZE)) {
S
Sage Weil 已提交
1292
		dout(" zeroing %p 0 - %d and %d - %d\n",
1293
		     page, pos_in_page, end_in_page, (int)PAGE_SIZE);
S
Sage Weil 已提交
1294 1295
		zero_user_segments(page,
				   0, pos_in_page,
1296
				   end_in_page, PAGE_SIZE);
S
Sage Weil 已提交
1297 1298 1299 1300
		return 0;
	}

	/* we need to read it. */
Y
Yan, Zheng 已提交
1301 1302 1303 1304 1305 1306
	r = ceph_do_readpage(file, page);
	if (r < 0) {
		if (r == -EINPROGRESS)
			return -EAGAIN;
		goto fail_unlock;
	}
S
Sage Weil 已提交
1307
	goto retry_locked;
Y
Yan, Zheng 已提交
1308
fail_unlock:
S
Sage Weil 已提交
1309 1310 1311 1312
	unlock_page(page);
	return r;
}

1313 1314 1315 1316 1317 1318 1319 1320
/*
 * 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 已提交
1321
	struct inode *inode = file_inode(file);
1322
	struct page *page;
1323
	pgoff_t index = pos >> PAGE_SHIFT;
S
Sage Weil 已提交
1324
	int r;
1325 1326

	do {
1327
		/* get a page */
1328
		page = grab_cache_page_write_begin(mapping, index, 0);
S
Sage Weil 已提交
1329 1330
		if (!page)
			return -ENOMEM;
1331 1332

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

		r = ceph_update_writeable_page(file, pos, len, page);
1336
		if (r < 0)
1337
			put_page(page);
1338 1339
		else
			*pagep = page;
1340 1341 1342 1343 1344
	} while (r == -EAGAIN);

	return r;
}

S
Sage Weil 已提交
1345 1346
/*
 * we don't do anything in here that simple_write_end doesn't do
1347
 * except adjust dirty page accounting
S
Sage Weil 已提交
1348 1349 1350 1351 1352
 */
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 已提交
1353
	struct inode *inode = file_inode(file);
1354
	bool check_cap = false;
S
Sage Weil 已提交
1355 1356 1357 1358 1359

	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 已提交
1360 1361 1362 1363 1364 1365 1366
	if (!PageUptodate(page)) {
		if (copied < len) {
			copied = 0;
			goto out;
		}
		SetPageUptodate(page);
	}
S
Sage Weil 已提交
1367 1368

	/* did file size increase? */
1369
	if (pos+copied > i_size_read(inode))
S
Sage Weil 已提交
1370 1371 1372 1373
		check_cap = ceph_inode_set_size(inode, pos+copied);

	set_page_dirty(page);

A
Al Viro 已提交
1374
out:
S
Sage Weil 已提交
1375
	unlock_page(page);
1376
	put_page(page);
S
Sage Weil 已提交
1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388

	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.
 */
1389
static ssize_t ceph_direct_io(struct kiocb *iocb, struct iov_iter *iter)
S
Sage Weil 已提交
1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407
{
	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,
};

1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418
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 已提交
1419 1420 1421 1422

/*
 * vm ops
 */
1423
static int ceph_filemap_fault(struct vm_fault *vmf)
1424
{
1425
	struct vm_area_struct *vma = vmf->vma;
1426 1427 1428
	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;
1429
	struct page *pinned_page = NULL;
1430
	loff_t off = vmf->pgoff << PAGE_SHIFT;
1431
	int want, got, ret;
1432 1433 1434
	sigset_t oldset;

	ceph_block_sigs(&oldset);
1435 1436

	dout("filemap_fault %p %llx.%llx %llu~%zd trying to get caps\n",
1437
	     inode, ceph_vinop(inode), off, (size_t)PAGE_SIZE);
1438 1439 1440 1441
	if (fi->fmode & CEPH_FILE_MODE_LAZY)
		want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
	else
		want = CEPH_CAP_FILE_CACHE;
1442 1443 1444

	got = 0;
	ret = ceph_get_caps(ci, CEPH_CAP_FILE_RD, want, -1, &got, &pinned_page);
1445
	if (ret < 0)
1446
		goto out_restore;
1447

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

Y
Yan, Zheng 已提交
1451
	if ((got & (CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO)) ||
1452 1453
	    ci->i_inline_version == CEPH_INLINE_NONE) {
		current->journal_info = vma->vm_file;
1454
		ret = filemap_fault(vmf);
1455 1456
		current->journal_info = NULL;
	} else
Y
Yan, Zheng 已提交
1457
		ret = -EAGAIN;
1458 1459

	dout("filemap_fault %p %llu~%zd dropping cap refs on %s ret %d\n",
1460
	     inode, off, (size_t)PAGE_SIZE, ceph_cap_string(got), ret);
1461
	if (pinned_page)
1462
		put_page(pinned_page);
1463 1464
	ceph_put_cap_refs(ci, got);

Y
Yan, Zheng 已提交
1465
	if (ret != -EAGAIN)
1466
		goto out_restore;
Y
Yan, Zheng 已提交
1467 1468

	/* read inline data */
1469
	if (off >= PAGE_SIZE) {
Y
Yan, Zheng 已提交
1470 1471 1472 1473 1474 1475
		/* does not support inline data > PAGE_SIZE */
		ret = VM_FAULT_SIGBUS;
	} else {
		int ret1;
		struct address_space *mapping = inode->i_mapping;
		struct page *page = find_or_create_page(mapping, 0,
1476 1477
						mapping_gfp_constraint(mapping,
						~__GFP_FS));
Y
Yan, Zheng 已提交
1478 1479
		if (!page) {
			ret = VM_FAULT_OOM;
1480
			goto out_inline;
Y
Yan, Zheng 已提交
1481 1482 1483 1484 1485
		}
		ret1 = __ceph_do_getattr(inode, page,
					 CEPH_STAT_CAP_INLINE_DATA, true);
		if (ret1 < 0 || off >= i_size_read(inode)) {
			unlock_page(page);
1486
			put_page(page);
1487 1488 1489 1490
			if (ret1 < 0)
				ret = ret1;
			else
				ret = VM_FAULT_SIGBUS;
1491
			goto out_inline;
Y
Yan, Zheng 已提交
1492
		}
1493 1494
		if (ret1 < PAGE_SIZE)
			zero_user_segment(page, ret1, PAGE_SIZE);
Y
Yan, Zheng 已提交
1495 1496 1497 1498 1499
		else
			flush_dcache_page(page);
		SetPageUptodate(page);
		vmf->page = page;
		ret = VM_FAULT_MAJOR | VM_FAULT_LOCKED;
1500 1501 1502
out_inline:
		dout("filemap_fault %p %llu~%zd read inline data ret %d\n",
		     inode, off, (size_t)PAGE_SIZE, ret);
Y
Yan, Zheng 已提交
1503
	}
1504 1505
out_restore:
	ceph_restore_sigs(&oldset);
1506 1507 1508
	if (ret < 0)
		ret = (ret == -ENOMEM) ? VM_FAULT_OOM : VM_FAULT_SIGBUS;

1509 1510
	return ret;
}
S
Sage Weil 已提交
1511 1512 1513 1514

/*
 * Reuse write_begin here for simplicity.
 */
1515
static int ceph_page_mkwrite(struct vm_fault *vmf)
S
Sage Weil 已提交
1516
{
1517
	struct vm_area_struct *vma = vmf->vma;
A
Al Viro 已提交
1518
	struct inode *inode = file_inode(vma->vm_file);
1519 1520
	struct ceph_inode_info *ci = ceph_inode(inode);
	struct ceph_file_info *fi = vma->vm_file->private_data;
1521
	struct ceph_cap_flush *prealloc_cf;
1522
	struct page *page = vmf->page;
1523
	loff_t off = page_offset(page);
1524 1525 1526
	loff_t size = i_size_read(inode);
	size_t len;
	int want, got, ret;
1527
	sigset_t oldset;
1528

1529 1530
	prealloc_cf = ceph_alloc_cap_flush();
	if (!prealloc_cf)
1531
		return VM_FAULT_OOM;
1532

1533
	ceph_block_sigs(&oldset);
1534

1535 1536 1537 1538 1539 1540 1541 1542 1543
	if (ci->i_inline_version != CEPH_INLINE_NONE) {
		struct page *locked_page = NULL;
		if (off == 0) {
			lock_page(page);
			locked_page = page;
		}
		ret = ceph_uninline_data(vma->vm_file, locked_page);
		if (locked_page)
			unlock_page(locked_page);
1544
		if (ret < 0)
1545
			goto out_free;
1546 1547
	}

1548 1549
	if (off + PAGE_SIZE <= size)
		len = PAGE_SIZE;
S
Sage Weil 已提交
1550
	else
1551
		len = size & ~PAGE_MASK;
S
Sage Weil 已提交
1552

1553 1554 1555 1556 1557 1558
	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;
1559 1560 1561 1562

	got = 0;
	ret = ceph_get_caps(ci, CEPH_CAP_FILE_WR, want, off + len,
			    &got, NULL);
1563
	if (ret < 0)
1564
		goto out_free;
1565

1566 1567 1568 1569 1570
	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);
1571

1572 1573
	do {
		lock_page(page);
1574

1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587
		if ((off > size) || (page->mapping != inode->i_mapping)) {
			unlock_page(page);
			ret = VM_FAULT_NOPAGE;
			break;
		}

		ret = ceph_update_writeable_page(vma->vm_file, off, len, page);
		if (ret >= 0) {
			/* success.  we'll keep the page locked. */
			set_page_dirty(page);
			ret = VM_FAULT_LOCKED;
		}
	} while (ret == -EAGAIN);
1588

1589 1590
	if (ret == VM_FAULT_LOCKED ||
	    ci->i_inline_version != CEPH_INLINE_NONE) {
1591 1592
		int dirty;
		spin_lock(&ci->i_ceph_lock);
1593
		ci->i_inline_version = CEPH_INLINE_NONE;
1594 1595
		dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
					       &prealloc_cf);
1596 1597 1598 1599 1600 1601 1602 1603
		spin_unlock(&ci->i_ceph_lock);
		if (dirty)
			__mark_inode_dirty(inode, dirty);
	}

	dout("page_mkwrite %p %llu~%zd dropping cap refs on %s ret %d\n",
	     inode, off, len, ceph_cap_string(got), ret);
	ceph_put_cap_refs(ci, got);
1604
out_free:
1605
	ceph_restore_sigs(&oldset);
1606
	ceph_free_cap_flush(prealloc_cf);
1607 1608
	if (ret < 0)
		ret = (ret == -ENOMEM) ? VM_FAULT_OOM : VM_FAULT_SIGBUS;
S
Sage Weil 已提交
1609 1610 1611
	return ret;
}

Y
Yan, Zheng 已提交
1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623
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,
1624 1625
					   mapping_gfp_constraint(mapping,
					   ~__GFP_FS));
Y
Yan, Zheng 已提交
1626 1627 1628 1629
		if (!page)
			return;
		if (PageUptodate(page)) {
			unlock_page(page);
1630
			put_page(page);
Y
Yan, Zheng 已提交
1631 1632 1633 1634
			return;
		}
	}

1635
	dout("fill_inline_data %p %llx.%llx len %zu locked_page %p\n",
Y
Yan, Zheng 已提交
1636 1637 1638 1639 1640 1641 1642 1643 1644
	     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) {
1645 1646
		if (len < PAGE_SIZE)
			zero_user_segment(page, len, PAGE_SIZE);
Y
Yan, Zheng 已提交
1647 1648 1649 1650 1651
		else
			flush_dcache_page(page);

		SetPageUptodate(page);
		unlock_page(page);
1652
		put_page(page);
Y
Yan, Zheng 已提交
1653 1654 1655
	}
}

1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688
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 {
1689
				put_page(page);
1690 1691 1692 1693 1694 1695 1696
				page = NULL;
			}
		}
	}

	if (page) {
		len = i_size_read(inode);
1697 1698
		if (len > PAGE_SIZE)
			len = PAGE_SIZE;
1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717
	} 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,
1718
				    CEPH_OSD_OP_CREATE, CEPH_OSD_FLAG_WRITE,
I
Ilya Dryomov 已提交
1719
				    NULL, 0, 0, false);
1720 1721 1722 1723 1724
	if (IS_ERR(req)) {
		err = PTR_ERR(req);
		goto out;
	}

1725
	req->r_mtime = inode->i_mtime;
1726 1727 1728 1729 1730 1731 1732 1733 1734
	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,
1735
				    CEPH_OSD_OP_WRITE, CEPH_OSD_FLAG_WRITE,
I
Ilya Dryomov 已提交
1736 1737
				    NULL, ci->i_truncate_seq,
				    ci->i_truncate_size, false);
1738 1739 1740 1741 1742 1743 1744
	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 已提交
1745 1746 1747 1748 1749 1750 1751 1752 1753 1754 1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765
	{
		__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;
	}
1766

1767
	req->r_mtime = inode->i_mtime;
1768 1769 1770 1771 1772 1773 1774 1775 1776 1777 1778
	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);
1779
			put_page(page);
1780 1781 1782 1783 1784 1785 1786 1787 1788
		} 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;
}

1789
static const struct vm_operations_struct ceph_vmops = {
1790
	.fault		= ceph_filemap_fault,
S
Sage Weil 已提交
1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802 1803
	.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;
}
1804 1805 1806 1807 1808 1809

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

Y
Yan, Zheng 已提交
1810 1811
static int __ceph_pool_perm_get(struct ceph_inode_info *ci,
				s64 pool, struct ceph_string *pool_ns)
1812 1813 1814 1815 1816 1817 1818
{
	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 已提交
1819
	size_t pool_ns_len;
1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830
	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 已提交
1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841
			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;
			}
1842 1843 1844 1845 1846 1847
		}
	}
	up_read(&mdsc->pool_perm_rwsem);
	if (*p)
		goto out;

Y
Yan, Zheng 已提交
1848 1849 1850 1851 1852
	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);
1853 1854

	down_write(&mdsc->pool_perm_rwsem);
Y
Yan, Zheng 已提交
1855
	p = &mdsc->pool_perm_tree.rb_node;
1856 1857 1858 1859 1860 1861 1862 1863 1864
	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 已提交
1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875
			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;
			}
1876 1877 1878 1879 1880 1881 1882
		}
	}
	if (*p) {
		up_write(&mdsc->pool_perm_rwsem);
		goto out;
	}

I
Ilya Dryomov 已提交
1883
	rd_req = ceph_osdc_alloc_request(&fsc->client->osdc, NULL,
1884 1885 1886 1887 1888 1889 1890 1891 1892
					 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 已提交
1893 1894
	if (pool_ns)
		rd_req->r_base_oloc.pool_ns = ceph_get_string(pool_ns);
1895
	ceph_oid_printf(&rd_req->r_base_oid, "%llx.00000000", ci->i_vino.ino);
1896

1897 1898 1899
	err = ceph_osdc_alloc_messages(rd_req, GFP_NOFS);
	if (err)
		goto out_unlock;
1900

I
Ilya Dryomov 已提交
1901
	wr_req = ceph_osdc_alloc_request(&fsc->client->osdc, NULL,
1902 1903 1904 1905 1906 1907
					 1, false, GFP_NOFS);
	if (!wr_req) {
		err = -ENOMEM;
		goto out_unlock;
	}

1908
	wr_req->r_flags = CEPH_OSD_FLAG_WRITE;
1909
	osd_req_op_init(wr_req, 0, CEPH_OSD_OP_CREATE, CEPH_OSD_OP_FLAG_EXCL);
1910
	ceph_oloc_copy(&wr_req->r_base_oloc, &rd_req->r_base_oloc);
1911
	ceph_oid_copy(&wr_req->r_base_oid, &rd_req->r_base_oid);
1912

1913 1914 1915
	err = ceph_osdc_alloc_messages(wr_req, GFP_NOFS);
	if (err)
		goto out_unlock;
1916 1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927

	/* 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);

1928
	wr_req->r_mtime = ci->vfs_inode.i_mtime;
1929
	wr_req->r_abort_on_full = true;
1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948
	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 已提交
1949 1950
	pool_ns_len = pool_ns ? pool_ns->len : 0;
	perm = kmalloc(sizeof(*perm) + pool_ns_len + 1, GFP_NOFS);
1951 1952 1953 1954 1955 1956 1957
	if (!perm) {
		err = -ENOMEM;
		goto out_unlock;
	}

	perm->pool = pool;
	perm->perm = have;
Y
Yan, Zheng 已提交
1958 1959 1960 1961 1962
	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;

1963 1964 1965 1966 1967 1968
	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);

1969 1970
	ceph_osdc_put_request(rd_req);
	ceph_osdc_put_request(wr_req);
1971 1972 1973
out:
	if (!err)
		err = have;
Y
Yan, Zheng 已提交
1974 1975 1976 1977 1978
	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);
1979 1980 1981 1982 1983
	return err;
}

int ceph_pool_perm_check(struct ceph_inode_info *ci, int need)
{
1984
	s64 pool;
Y
Yan, Zheng 已提交
1985
	struct ceph_string *pool_ns;
1986 1987
	int ret, flags;

1988 1989 1990 1991 1992 1993 1994 1995 1996
	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;
	}

1997 1998 1999 2000 2001 2002
	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;
2003
	pool = ci->i_layout.pool_id;
2004 2005 2006 2007
	spin_unlock(&ci->i_ceph_lock);
check:
	if (flags & CEPH_I_POOL_PERM) {
		if ((need & CEPH_CAP_FILE_RD) && !(flags & CEPH_I_POOL_RD)) {
2008
			dout("ceph_pool_perm_check pool %lld no read perm\n",
2009 2010 2011 2012
			     pool);
			return -EPERM;
		}
		if ((need & CEPH_CAP_FILE_WR) && !(flags & CEPH_I_POOL_WR)) {
2013
			dout("ceph_pool_perm_check pool %lld no write perm\n",
2014 2015 2016 2017 2018 2019
			     pool);
			return -EPERM;
		}
		return 0;
	}

Y
Yan, Zheng 已提交
2020 2021 2022
	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);
2023 2024 2025 2026 2027 2028 2029 2030 2031 2032
	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 已提交
2033 2034 2035
	if (pool == ci->i_layout.pool_id &&
	    pool_ns == rcu_dereference_raw(ci->i_layout.pool_ns)) {
		ci->i_ceph_flags |= flags;
2036
        } else {
2037
		pool = ci->i_layout.pool_id;
2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055
		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);
	}
}