xref: /linux/fs/ceph/file.c (revision a23e1966932464e1c5226cb9ac4ce1d5fc10ba22)
1 // SPDX-License-Identifier: GPL-2.0
2 #include <linux/ceph/ceph_debug.h>
3 #include <linux/ceph/striper.h>
4 
5 #include <linux/module.h>
6 #include <linux/sched.h>
7 #include <linux/slab.h>
8 #include <linux/file.h>
9 #include <linux/mount.h>
10 #include <linux/namei.h>
11 #include <linux/writeback.h>
12 #include <linux/falloc.h>
13 #include <linux/iversion.h>
14 #include <linux/ktime.h>
15 #include <linux/splice.h>
16 
17 #include "super.h"
18 #include "mds_client.h"
19 #include "cache.h"
20 #include "io.h"
21 #include "metric.h"
22 
23 static __le32 ceph_flags_sys2wire(struct ceph_mds_client *mdsc, u32 flags)
24 {
25 	struct ceph_client *cl = mdsc->fsc->client;
26 	u32 wire_flags = 0;
27 
28 	switch (flags & O_ACCMODE) {
29 	case O_RDONLY:
30 		wire_flags |= CEPH_O_RDONLY;
31 		break;
32 	case O_WRONLY:
33 		wire_flags |= CEPH_O_WRONLY;
34 		break;
35 	case O_RDWR:
36 		wire_flags |= CEPH_O_RDWR;
37 		break;
38 	}
39 
40 	flags &= ~O_ACCMODE;
41 
42 #define ceph_sys2wire(a) if (flags & a) { wire_flags |= CEPH_##a; flags &= ~a; }
43 
44 	ceph_sys2wire(O_CREAT);
45 	ceph_sys2wire(O_EXCL);
46 	ceph_sys2wire(O_TRUNC);
47 	ceph_sys2wire(O_DIRECTORY);
48 	ceph_sys2wire(O_NOFOLLOW);
49 
50 #undef ceph_sys2wire
51 
52 	if (flags)
53 		doutc(cl, "unused open flags: %x\n", flags);
54 
55 	return cpu_to_le32(wire_flags);
56 }
57 
58 /*
59  * Ceph file operations
60  *
61  * Implement basic open/close functionality, and implement
62  * read/write.
63  *
64  * We implement three modes of file I/O:
65  *  - buffered uses the generic_file_aio_{read,write} helpers
66  *
67  *  - synchronous is used when there is multi-client read/write
68  *    sharing, avoids the page cache, and synchronously waits for an
69  *    ack from the OSD.
70  *
71  *  - direct io takes the variant of the sync path that references
72  *    user pages directly.
73  *
74  * fsync() flushes and waits on dirty pages, but just queues metadata
75  * for writeback: since the MDS can recover size and mtime there is no
76  * need to wait for MDS acknowledgement.
77  */
78 
79 /*
80  * How many pages to get in one call to iov_iter_get_pages().  This
81  * determines the size of the on-stack array used as a buffer.
82  */
83 #define ITER_GET_BVECS_PAGES	64
84 
85 static ssize_t __iter_get_bvecs(struct iov_iter *iter, size_t maxsize,
86 				struct bio_vec *bvecs)
87 {
88 	size_t size = 0;
89 	int bvec_idx = 0;
90 
91 	if (maxsize > iov_iter_count(iter))
92 		maxsize = iov_iter_count(iter);
93 
94 	while (size < maxsize) {
95 		struct page *pages[ITER_GET_BVECS_PAGES];
96 		ssize_t bytes;
97 		size_t start;
98 		int idx = 0;
99 
100 		bytes = iov_iter_get_pages2(iter, pages, maxsize - size,
101 					   ITER_GET_BVECS_PAGES, &start);
102 		if (bytes < 0)
103 			return size ?: bytes;
104 
105 		size += bytes;
106 
107 		for ( ; bytes; idx++, bvec_idx++) {
108 			int len = min_t(int, bytes, PAGE_SIZE - start);
109 
110 			bvec_set_page(&bvecs[bvec_idx], pages[idx], len, start);
111 			bytes -= len;
112 			start = 0;
113 		}
114 	}
115 
116 	return size;
117 }
118 
119 /*
120  * iov_iter_get_pages() only considers one iov_iter segment, no matter
121  * what maxsize or maxpages are given.  For ITER_BVEC that is a single
122  * page.
123  *
124  * Attempt to get up to @maxsize bytes worth of pages from @iter.
125  * Return the number of bytes in the created bio_vec array, or an error.
126  */
127 static ssize_t iter_get_bvecs_alloc(struct iov_iter *iter, size_t maxsize,
128 				    struct bio_vec **bvecs, int *num_bvecs)
129 {
130 	struct bio_vec *bv;
131 	size_t orig_count = iov_iter_count(iter);
132 	ssize_t bytes;
133 	int npages;
134 
135 	iov_iter_truncate(iter, maxsize);
136 	npages = iov_iter_npages(iter, INT_MAX);
137 	iov_iter_reexpand(iter, orig_count);
138 
139 	/*
140 	 * __iter_get_bvecs() may populate only part of the array -- zero it
141 	 * out.
142 	 */
143 	bv = kvmalloc_array(npages, sizeof(*bv), GFP_KERNEL | __GFP_ZERO);
144 	if (!bv)
145 		return -ENOMEM;
146 
147 	bytes = __iter_get_bvecs(iter, maxsize, bv);
148 	if (bytes < 0) {
149 		/*
150 		 * No pages were pinned -- just free the array.
151 		 */
152 		kvfree(bv);
153 		return bytes;
154 	}
155 
156 	*bvecs = bv;
157 	*num_bvecs = npages;
158 	return bytes;
159 }
160 
161 static void put_bvecs(struct bio_vec *bvecs, int num_bvecs, bool should_dirty)
162 {
163 	int i;
164 
165 	for (i = 0; i < num_bvecs; i++) {
166 		if (bvecs[i].bv_page) {
167 			if (should_dirty)
168 				set_page_dirty_lock(bvecs[i].bv_page);
169 			put_page(bvecs[i].bv_page);
170 		}
171 	}
172 	kvfree(bvecs);
173 }
174 
175 /*
176  * Prepare an open request.  Preallocate ceph_cap to avoid an
177  * inopportune ENOMEM later.
178  */
179 static struct ceph_mds_request *
180 prepare_open_request(struct super_block *sb, int flags, int create_mode)
181 {
182 	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(sb);
183 	struct ceph_mds_request *req;
184 	int want_auth = USE_ANY_MDS;
185 	int op = (flags & O_CREAT) ? CEPH_MDS_OP_CREATE : CEPH_MDS_OP_OPEN;
186 
187 	if (flags & (O_WRONLY|O_RDWR|O_CREAT|O_TRUNC))
188 		want_auth = USE_AUTH_MDS;
189 
190 	req = ceph_mdsc_create_request(mdsc, op, want_auth);
191 	if (IS_ERR(req))
192 		goto out;
193 	req->r_fmode = ceph_flags_to_mode(flags);
194 	req->r_args.open.flags = ceph_flags_sys2wire(mdsc, flags);
195 	req->r_args.open.mode = cpu_to_le32(create_mode);
196 out:
197 	return req;
198 }
199 
200 static int ceph_init_file_info(struct inode *inode, struct file *file,
201 					int fmode, bool isdir)
202 {
203 	struct ceph_inode_info *ci = ceph_inode(inode);
204 	struct ceph_mount_options *opt =
205 		ceph_inode_to_fs_client(&ci->netfs.inode)->mount_options;
206 	struct ceph_client *cl = ceph_inode_to_client(inode);
207 	struct ceph_file_info *fi;
208 	int ret;
209 
210 	doutc(cl, "%p %llx.%llx %p 0%o (%s)\n", inode, ceph_vinop(inode),
211 	      file, inode->i_mode, isdir ? "dir" : "regular");
212 	BUG_ON(inode->i_fop->release != ceph_release);
213 
214 	if (isdir) {
215 		struct ceph_dir_file_info *dfi =
216 			kmem_cache_zalloc(ceph_dir_file_cachep, GFP_KERNEL);
217 		if (!dfi)
218 			return -ENOMEM;
219 
220 		file->private_data = dfi;
221 		fi = &dfi->file_info;
222 		dfi->next_offset = 2;
223 		dfi->readdir_cache_idx = -1;
224 	} else {
225 		fi = kmem_cache_zalloc(ceph_file_cachep, GFP_KERNEL);
226 		if (!fi)
227 			return -ENOMEM;
228 
229 		if (opt->flags & CEPH_MOUNT_OPT_NOPAGECACHE)
230 			fi->flags |= CEPH_F_SYNC;
231 
232 		file->private_data = fi;
233 	}
234 
235 	ceph_get_fmode(ci, fmode, 1);
236 	fi->fmode = fmode;
237 
238 	spin_lock_init(&fi->rw_contexts_lock);
239 	INIT_LIST_HEAD(&fi->rw_contexts);
240 	fi->filp_gen = READ_ONCE(ceph_inode_to_fs_client(inode)->filp_gen);
241 
242 	if ((file->f_mode & FMODE_WRITE) && ceph_has_inline_data(ci)) {
243 		ret = ceph_uninline_data(file);
244 		if (ret < 0)
245 			goto error;
246 	}
247 
248 	return 0;
249 
250 error:
251 	ceph_fscache_unuse_cookie(inode, file->f_mode & FMODE_WRITE);
252 	ceph_put_fmode(ci, fi->fmode, 1);
253 	kmem_cache_free(ceph_file_cachep, fi);
254 	/* wake up anyone waiting for caps on this inode */
255 	wake_up_all(&ci->i_cap_wq);
256 	return ret;
257 }
258 
259 /*
260  * initialize private struct file data.
261  * if we fail, clean up by dropping fmode reference on the ceph_inode
262  */
263 static int ceph_init_file(struct inode *inode, struct file *file, int fmode)
264 {
265 	struct ceph_client *cl = ceph_inode_to_client(inode);
266 	int ret = 0;
267 
268 	switch (inode->i_mode & S_IFMT) {
269 	case S_IFREG:
270 		ceph_fscache_use_cookie(inode, file->f_mode & FMODE_WRITE);
271 		fallthrough;
272 	case S_IFDIR:
273 		ret = ceph_init_file_info(inode, file, fmode,
274 						S_ISDIR(inode->i_mode));
275 		break;
276 
277 	case S_IFLNK:
278 		doutc(cl, "%p %llx.%llx %p 0%o (symlink)\n", inode,
279 		      ceph_vinop(inode), file, inode->i_mode);
280 		break;
281 
282 	default:
283 		doutc(cl, "%p %llx.%llx %p 0%o (special)\n", inode,
284 		      ceph_vinop(inode), file, inode->i_mode);
285 		/*
286 		 * we need to drop the open ref now, since we don't
287 		 * have .release set to ceph_release.
288 		 */
289 		BUG_ON(inode->i_fop->release == ceph_release);
290 
291 		/* call the proper open fop */
292 		ret = inode->i_fop->open(inode, file);
293 	}
294 	return ret;
295 }
296 
297 /*
298  * try renew caps after session gets killed.
299  */
300 int ceph_renew_caps(struct inode *inode, int fmode)
301 {
302 	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
303 	struct ceph_client *cl = mdsc->fsc->client;
304 	struct ceph_inode_info *ci = ceph_inode(inode);
305 	struct ceph_mds_request *req;
306 	int err, flags, wanted;
307 
308 	spin_lock(&ci->i_ceph_lock);
309 	__ceph_touch_fmode(ci, mdsc, fmode);
310 	wanted = __ceph_caps_file_wanted(ci);
311 	if (__ceph_is_any_real_caps(ci) &&
312 	    (!(wanted & CEPH_CAP_ANY_WR) || ci->i_auth_cap)) {
313 		int issued = __ceph_caps_issued(ci, NULL);
314 		spin_unlock(&ci->i_ceph_lock);
315 		doutc(cl, "%p %llx.%llx want %s issued %s updating mds_wanted\n",
316 		      inode, ceph_vinop(inode), ceph_cap_string(wanted),
317 		      ceph_cap_string(issued));
318 		ceph_check_caps(ci, 0);
319 		return 0;
320 	}
321 	spin_unlock(&ci->i_ceph_lock);
322 
323 	flags = 0;
324 	if ((wanted & CEPH_CAP_FILE_RD) && (wanted & CEPH_CAP_FILE_WR))
325 		flags = O_RDWR;
326 	else if (wanted & CEPH_CAP_FILE_RD)
327 		flags = O_RDONLY;
328 	else if (wanted & CEPH_CAP_FILE_WR)
329 		flags = O_WRONLY;
330 #ifdef O_LAZY
331 	if (wanted & CEPH_CAP_FILE_LAZYIO)
332 		flags |= O_LAZY;
333 #endif
334 
335 	req = prepare_open_request(inode->i_sb, flags, 0);
336 	if (IS_ERR(req)) {
337 		err = PTR_ERR(req);
338 		goto out;
339 	}
340 
341 	req->r_inode = inode;
342 	ihold(inode);
343 	req->r_num_caps = 1;
344 
345 	err = ceph_mdsc_do_request(mdsc, NULL, req);
346 	ceph_mdsc_put_request(req);
347 out:
348 	doutc(cl, "%p %llx.%llx open result=%d\n", inode, ceph_vinop(inode),
349 	      err);
350 	return err < 0 ? err : 0;
351 }
352 
353 /*
354  * If we already have the requisite capabilities, we can satisfy
355  * the open request locally (no need to request new caps from the
356  * MDS).  We do, however, need to inform the MDS (asynchronously)
357  * if our wanted caps set expands.
358  */
359 int ceph_open(struct inode *inode, struct file *file)
360 {
361 	struct ceph_inode_info *ci = ceph_inode(inode);
362 	struct ceph_fs_client *fsc = ceph_sb_to_fs_client(inode->i_sb);
363 	struct ceph_client *cl = fsc->client;
364 	struct ceph_mds_client *mdsc = fsc->mdsc;
365 	struct ceph_mds_request *req;
366 	struct ceph_file_info *fi = file->private_data;
367 	int err;
368 	int flags, fmode, wanted;
369 
370 	if (fi) {
371 		doutc(cl, "file %p is already opened\n", file);
372 		return 0;
373 	}
374 
375 	/* filter out O_CREAT|O_EXCL; vfs did that already.  yuck. */
376 	flags = file->f_flags & ~(O_CREAT|O_EXCL);
377 	if (S_ISDIR(inode->i_mode)) {
378 		flags = O_DIRECTORY;  /* mds likes to know */
379 	} else if (S_ISREG(inode->i_mode)) {
380 		err = fscrypt_file_open(inode, file);
381 		if (err)
382 			return err;
383 	}
384 
385 	doutc(cl, "%p %llx.%llx file %p flags %d (%d)\n", inode,
386 	      ceph_vinop(inode), file, flags, file->f_flags);
387 	fmode = ceph_flags_to_mode(flags);
388 	wanted = ceph_caps_for_mode(fmode);
389 
390 	/* snapped files are read-only */
391 	if (ceph_snap(inode) != CEPH_NOSNAP && (file->f_mode & FMODE_WRITE))
392 		return -EROFS;
393 
394 	/* trivially open snapdir */
395 	if (ceph_snap(inode) == CEPH_SNAPDIR) {
396 		return ceph_init_file(inode, file, fmode);
397 	}
398 
399 	/*
400 	 * No need to block if we have caps on the auth MDS (for
401 	 * write) or any MDS (for read).  Update wanted set
402 	 * asynchronously.
403 	 */
404 	spin_lock(&ci->i_ceph_lock);
405 	if (__ceph_is_any_real_caps(ci) &&
406 	    (((fmode & CEPH_FILE_MODE_WR) == 0) || ci->i_auth_cap)) {
407 		int mds_wanted = __ceph_caps_mds_wanted(ci, true);
408 		int issued = __ceph_caps_issued(ci, NULL);
409 
410 		doutc(cl, "open %p fmode %d want %s issued %s using existing\n",
411 		      inode, fmode, ceph_cap_string(wanted),
412 		      ceph_cap_string(issued));
413 		__ceph_touch_fmode(ci, mdsc, fmode);
414 		spin_unlock(&ci->i_ceph_lock);
415 
416 		/* adjust wanted? */
417 		if ((issued & wanted) != wanted &&
418 		    (mds_wanted & wanted) != wanted &&
419 		    ceph_snap(inode) != CEPH_SNAPDIR)
420 			ceph_check_caps(ci, 0);
421 
422 		return ceph_init_file(inode, file, fmode);
423 	} else if (ceph_snap(inode) != CEPH_NOSNAP &&
424 		   (ci->i_snap_caps & wanted) == wanted) {
425 		__ceph_touch_fmode(ci, mdsc, fmode);
426 		spin_unlock(&ci->i_ceph_lock);
427 		return ceph_init_file(inode, file, fmode);
428 	}
429 
430 	spin_unlock(&ci->i_ceph_lock);
431 
432 	doutc(cl, "open fmode %d wants %s\n", fmode, ceph_cap_string(wanted));
433 	req = prepare_open_request(inode->i_sb, flags, 0);
434 	if (IS_ERR(req)) {
435 		err = PTR_ERR(req);
436 		goto out;
437 	}
438 	req->r_inode = inode;
439 	ihold(inode);
440 
441 	req->r_num_caps = 1;
442 	err = ceph_mdsc_do_request(mdsc, NULL, req);
443 	if (!err)
444 		err = ceph_init_file(inode, file, req->r_fmode);
445 	ceph_mdsc_put_request(req);
446 	doutc(cl, "open result=%d on %llx.%llx\n", err, ceph_vinop(inode));
447 out:
448 	return err;
449 }
450 
451 /* Clone the layout from a synchronous create, if the dir now has Dc caps */
452 static void
453 cache_file_layout(struct inode *dst, struct inode *src)
454 {
455 	struct ceph_inode_info *cdst = ceph_inode(dst);
456 	struct ceph_inode_info *csrc = ceph_inode(src);
457 
458 	spin_lock(&cdst->i_ceph_lock);
459 	if ((__ceph_caps_issued(cdst, NULL) & CEPH_CAP_DIR_CREATE) &&
460 	    !ceph_file_layout_is_valid(&cdst->i_cached_layout)) {
461 		memcpy(&cdst->i_cached_layout, &csrc->i_layout,
462 			sizeof(cdst->i_cached_layout));
463 		rcu_assign_pointer(cdst->i_cached_layout.pool_ns,
464 				   ceph_try_get_string(csrc->i_layout.pool_ns));
465 	}
466 	spin_unlock(&cdst->i_ceph_lock);
467 }
468 
469 /*
470  * Try to set up an async create. We need caps, a file layout, and inode number,
471  * and either a lease on the dentry or complete dir info. If any of those
472  * criteria are not satisfied, then return false and the caller can go
473  * synchronous.
474  */
475 static int try_prep_async_create(struct inode *dir, struct dentry *dentry,
476 				 struct ceph_file_layout *lo, u64 *pino)
477 {
478 	struct ceph_inode_info *ci = ceph_inode(dir);
479 	struct ceph_dentry_info *di = ceph_dentry(dentry);
480 	int got = 0, want = CEPH_CAP_FILE_EXCL | CEPH_CAP_DIR_CREATE;
481 	u64 ino;
482 
483 	spin_lock(&ci->i_ceph_lock);
484 	/* No auth cap means no chance for Dc caps */
485 	if (!ci->i_auth_cap)
486 		goto no_async;
487 
488 	/* Any delegated inos? */
489 	if (xa_empty(&ci->i_auth_cap->session->s_delegated_inos))
490 		goto no_async;
491 
492 	if (!ceph_file_layout_is_valid(&ci->i_cached_layout))
493 		goto no_async;
494 
495 	if ((__ceph_caps_issued(ci, NULL) & want) != want)
496 		goto no_async;
497 
498 	if (d_in_lookup(dentry)) {
499 		if (!__ceph_dir_is_complete(ci))
500 			goto no_async;
501 		spin_lock(&dentry->d_lock);
502 		di->lease_shared_gen = atomic_read(&ci->i_shared_gen);
503 		spin_unlock(&dentry->d_lock);
504 	} else if (atomic_read(&ci->i_shared_gen) !=
505 		   READ_ONCE(di->lease_shared_gen)) {
506 		goto no_async;
507 	}
508 
509 	ino = ceph_get_deleg_ino(ci->i_auth_cap->session);
510 	if (!ino)
511 		goto no_async;
512 
513 	*pino = ino;
514 	ceph_take_cap_refs(ci, want, false);
515 	memcpy(lo, &ci->i_cached_layout, sizeof(*lo));
516 	rcu_assign_pointer(lo->pool_ns,
517 			   ceph_try_get_string(ci->i_cached_layout.pool_ns));
518 	got = want;
519 no_async:
520 	spin_unlock(&ci->i_ceph_lock);
521 	return got;
522 }
523 
524 static void restore_deleg_ino(struct inode *dir, u64 ino)
525 {
526 	struct ceph_client *cl = ceph_inode_to_client(dir);
527 	struct ceph_inode_info *ci = ceph_inode(dir);
528 	struct ceph_mds_session *s = NULL;
529 
530 	spin_lock(&ci->i_ceph_lock);
531 	if (ci->i_auth_cap)
532 		s = ceph_get_mds_session(ci->i_auth_cap->session);
533 	spin_unlock(&ci->i_ceph_lock);
534 	if (s) {
535 		int err = ceph_restore_deleg_ino(s, ino);
536 		if (err)
537 			pr_warn_client(cl,
538 				"unable to restore delegated ino 0x%llx to session: %d\n",
539 				ino, err);
540 		ceph_put_mds_session(s);
541 	}
542 }
543 
544 static void wake_async_create_waiters(struct inode *inode,
545 				      struct ceph_mds_session *session)
546 {
547 	struct ceph_inode_info *ci = ceph_inode(inode);
548 	bool check_cap = false;
549 
550 	spin_lock(&ci->i_ceph_lock);
551 	if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE) {
552 		ci->i_ceph_flags &= ~CEPH_I_ASYNC_CREATE;
553 		wake_up_bit(&ci->i_ceph_flags, CEPH_ASYNC_CREATE_BIT);
554 
555 		if (ci->i_ceph_flags & CEPH_I_ASYNC_CHECK_CAPS) {
556 			ci->i_ceph_flags &= ~CEPH_I_ASYNC_CHECK_CAPS;
557 			check_cap = true;
558 		}
559 	}
560 	ceph_kick_flushing_inode_caps(session, ci);
561 	spin_unlock(&ci->i_ceph_lock);
562 
563 	if (check_cap)
564 		ceph_check_caps(ci, CHECK_CAPS_FLUSH);
565 }
566 
567 static void ceph_async_create_cb(struct ceph_mds_client *mdsc,
568                                  struct ceph_mds_request *req)
569 {
570 	struct ceph_client *cl = mdsc->fsc->client;
571 	struct dentry *dentry = req->r_dentry;
572 	struct inode *dinode = d_inode(dentry);
573 	struct inode *tinode = req->r_target_inode;
574 	int result = req->r_err ? req->r_err :
575 			le32_to_cpu(req->r_reply_info.head->result);
576 
577 	WARN_ON_ONCE(dinode && tinode && dinode != tinode);
578 
579 	/* MDS changed -- caller must resubmit */
580 	if (result == -EJUKEBOX)
581 		goto out;
582 
583 	mapping_set_error(req->r_parent->i_mapping, result);
584 
585 	if (result) {
586 		int pathlen = 0;
587 		u64 base = 0;
588 		char *path = ceph_mdsc_build_path(mdsc, req->r_dentry, &pathlen,
589 						  &base, 0);
590 
591 		pr_warn_client(cl,
592 			"async create failure path=(%llx)%s result=%d!\n",
593 			base, IS_ERR(path) ? "<<bad>>" : path, result);
594 		ceph_mdsc_free_path(path, pathlen);
595 
596 		ceph_dir_clear_complete(req->r_parent);
597 		if (!d_unhashed(dentry))
598 			d_drop(dentry);
599 
600 		if (dinode) {
601 			mapping_set_error(dinode->i_mapping, result);
602 			ceph_inode_shutdown(dinode);
603 			wake_async_create_waiters(dinode, req->r_session);
604 		}
605 	}
606 
607 	if (tinode) {
608 		u64 ino = ceph_vino(tinode).ino;
609 
610 		if (req->r_deleg_ino != ino)
611 			pr_warn_client(cl,
612 				"inode number mismatch! err=%d deleg_ino=0x%llx target=0x%llx\n",
613 				req->r_err, req->r_deleg_ino, ino);
614 
615 		mapping_set_error(tinode->i_mapping, result);
616 		wake_async_create_waiters(tinode, req->r_session);
617 	} else if (!result) {
618 		pr_warn_client(cl, "no req->r_target_inode for 0x%llx\n",
619 			       req->r_deleg_ino);
620 	}
621 out:
622 	ceph_mdsc_release_dir_caps(req);
623 }
624 
625 static int ceph_finish_async_create(struct inode *dir, struct inode *inode,
626 				    struct dentry *dentry,
627 				    struct file *file, umode_t mode,
628 				    struct ceph_mds_request *req,
629 				    struct ceph_acl_sec_ctx *as_ctx,
630 				    struct ceph_file_layout *lo)
631 {
632 	int ret;
633 	char xattr_buf[4];
634 	struct ceph_mds_reply_inode in = { };
635 	struct ceph_mds_reply_info_in iinfo = { .in = &in };
636 	struct ceph_inode_info *ci = ceph_inode(dir);
637 	struct ceph_dentry_info *di = ceph_dentry(dentry);
638 	struct timespec64 now;
639 	struct ceph_string *pool_ns;
640 	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(dir->i_sb);
641 	struct ceph_client *cl = mdsc->fsc->client;
642 	struct ceph_vino vino = { .ino = req->r_deleg_ino,
643 				  .snap = CEPH_NOSNAP };
644 
645 	ktime_get_real_ts64(&now);
646 
647 	iinfo.inline_version = CEPH_INLINE_NONE;
648 	iinfo.change_attr = 1;
649 	ceph_encode_timespec64(&iinfo.btime, &now);
650 
651 	if (req->r_pagelist) {
652 		iinfo.xattr_len = req->r_pagelist->length;
653 		iinfo.xattr_data = req->r_pagelist->mapped_tail;
654 	} else {
655 		/* fake it */
656 		iinfo.xattr_len = ARRAY_SIZE(xattr_buf);
657 		iinfo.xattr_data = xattr_buf;
658 		memset(iinfo.xattr_data, 0, iinfo.xattr_len);
659 	}
660 
661 	in.ino = cpu_to_le64(vino.ino);
662 	in.snapid = cpu_to_le64(CEPH_NOSNAP);
663 	in.version = cpu_to_le64(1);	// ???
664 	in.cap.caps = in.cap.wanted = cpu_to_le32(CEPH_CAP_ALL_FILE);
665 	in.cap.cap_id = cpu_to_le64(1);
666 	in.cap.realm = cpu_to_le64(ci->i_snap_realm->ino);
667 	in.cap.flags = CEPH_CAP_FLAG_AUTH;
668 	in.ctime = in.mtime = in.atime = iinfo.btime;
669 	in.truncate_seq = cpu_to_le32(1);
670 	in.truncate_size = cpu_to_le64(-1ULL);
671 	in.xattr_version = cpu_to_le64(1);
672 	in.uid = cpu_to_le32(from_kuid(&init_user_ns,
673 				       mapped_fsuid(req->r_mnt_idmap,
674 						    &init_user_ns)));
675 	if (dir->i_mode & S_ISGID) {
676 		in.gid = cpu_to_le32(from_kgid(&init_user_ns, dir->i_gid));
677 
678 		/* Directories always inherit the setgid bit. */
679 		if (S_ISDIR(mode))
680 			mode |= S_ISGID;
681 	} else {
682 		in.gid = cpu_to_le32(from_kgid(&init_user_ns,
683 				     mapped_fsgid(req->r_mnt_idmap,
684 						  &init_user_ns)));
685 	}
686 	in.mode = cpu_to_le32((u32)mode);
687 
688 	in.nlink = cpu_to_le32(1);
689 	in.max_size = cpu_to_le64(lo->stripe_unit);
690 
691 	ceph_file_layout_to_legacy(lo, &in.layout);
692 	/* lo is private, so pool_ns can't change */
693 	pool_ns = rcu_dereference_raw(lo->pool_ns);
694 	if (pool_ns) {
695 		iinfo.pool_ns_len = pool_ns->len;
696 		iinfo.pool_ns_data = pool_ns->str;
697 	}
698 
699 	down_read(&mdsc->snap_rwsem);
700 	ret = ceph_fill_inode(inode, NULL, &iinfo, NULL, req->r_session,
701 			      req->r_fmode, NULL);
702 	up_read(&mdsc->snap_rwsem);
703 	if (ret) {
704 		doutc(cl, "failed to fill inode: %d\n", ret);
705 		ceph_dir_clear_complete(dir);
706 		if (!d_unhashed(dentry))
707 			d_drop(dentry);
708 		discard_new_inode(inode);
709 	} else {
710 		struct dentry *dn;
711 
712 		doutc(cl, "d_adding new inode 0x%llx to 0x%llx/%s\n",
713 		      vino.ino, ceph_ino(dir), dentry->d_name.name);
714 		ceph_dir_clear_ordered(dir);
715 		ceph_init_inode_acls(inode, as_ctx);
716 		if (inode->i_state & I_NEW) {
717 			/*
718 			 * If it's not I_NEW, then someone created this before
719 			 * we got here. Assume the server is aware of it at
720 			 * that point and don't worry about setting
721 			 * CEPH_I_ASYNC_CREATE.
722 			 */
723 			ceph_inode(inode)->i_ceph_flags = CEPH_I_ASYNC_CREATE;
724 			unlock_new_inode(inode);
725 		}
726 		if (d_in_lookup(dentry) || d_really_is_negative(dentry)) {
727 			if (!d_unhashed(dentry))
728 				d_drop(dentry);
729 			dn = d_splice_alias(inode, dentry);
730 			WARN_ON_ONCE(dn && dn != dentry);
731 		}
732 		file->f_mode |= FMODE_CREATED;
733 		ret = finish_open(file, dentry, ceph_open);
734 	}
735 
736 	spin_lock(&dentry->d_lock);
737 	di->flags &= ~CEPH_DENTRY_ASYNC_CREATE;
738 	wake_up_bit(&di->flags, CEPH_DENTRY_ASYNC_CREATE_BIT);
739 	spin_unlock(&dentry->d_lock);
740 
741 	return ret;
742 }
743 
744 /*
745  * Do a lookup + open with a single request.  If we get a non-existent
746  * file or symlink, return 1 so the VFS can retry.
747  */
748 int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
749 		     struct file *file, unsigned flags, umode_t mode)
750 {
751 	struct mnt_idmap *idmap = file_mnt_idmap(file);
752 	struct ceph_fs_client *fsc = ceph_sb_to_fs_client(dir->i_sb);
753 	struct ceph_client *cl = fsc->client;
754 	struct ceph_mds_client *mdsc = fsc->mdsc;
755 	struct ceph_mds_request *req;
756 	struct inode *new_inode = NULL;
757 	struct dentry *dn;
758 	struct ceph_acl_sec_ctx as_ctx = {};
759 	bool try_async = ceph_test_mount_opt(fsc, ASYNC_DIROPS);
760 	int mask;
761 	int err;
762 
763 	doutc(cl, "%p %llx.%llx dentry %p '%pd' %s flags %d mode 0%o\n",
764 	      dir, ceph_vinop(dir), dentry, dentry,
765 	      d_unhashed(dentry) ? "unhashed" : "hashed", flags, mode);
766 
767 	if (dentry->d_name.len > NAME_MAX)
768 		return -ENAMETOOLONG;
769 
770 	err = ceph_wait_on_conflict_unlink(dentry);
771 	if (err)
772 		return err;
773 	/*
774 	 * Do not truncate the file, since atomic_open is called before the
775 	 * permission check. The caller will do the truncation afterward.
776 	 */
777 	flags &= ~O_TRUNC;
778 
779 retry:
780 	if (flags & O_CREAT) {
781 		if (ceph_quota_is_max_files_exceeded(dir))
782 			return -EDQUOT;
783 
784 		new_inode = ceph_new_inode(dir, dentry, &mode, &as_ctx);
785 		if (IS_ERR(new_inode)) {
786 			err = PTR_ERR(new_inode);
787 			goto out_ctx;
788 		}
789 		/* Async create can't handle more than a page of xattrs */
790 		if (as_ctx.pagelist &&
791 		    !list_is_singular(&as_ctx.pagelist->head))
792 			try_async = false;
793 	} else if (!d_in_lookup(dentry)) {
794 		/* If it's not being looked up, it's negative */
795 		return -ENOENT;
796 	}
797 
798 	/* do the open */
799 	req = prepare_open_request(dir->i_sb, flags, mode);
800 	if (IS_ERR(req)) {
801 		err = PTR_ERR(req);
802 		goto out_ctx;
803 	}
804 	req->r_dentry = dget(dentry);
805 	req->r_num_caps = 2;
806 	mask = CEPH_STAT_CAP_INODE | CEPH_CAP_AUTH_SHARED;
807 	if (ceph_security_xattr_wanted(dir))
808 		mask |= CEPH_CAP_XATTR_SHARED;
809 	req->r_args.open.mask = cpu_to_le32(mask);
810 	req->r_parent = dir;
811 	if (req->r_op == CEPH_MDS_OP_CREATE)
812 		req->r_mnt_idmap = mnt_idmap_get(idmap);
813 	ihold(dir);
814 	if (IS_ENCRYPTED(dir)) {
815 		set_bit(CEPH_MDS_R_FSCRYPT_FILE, &req->r_req_flags);
816 		err = fscrypt_prepare_lookup_partial(dir, dentry);
817 		if (err < 0)
818 			goto out_req;
819 	}
820 
821 	if (flags & O_CREAT) {
822 		struct ceph_file_layout lo;
823 
824 		req->r_dentry_drop = CEPH_CAP_FILE_SHARED | CEPH_CAP_AUTH_EXCL |
825 				     CEPH_CAP_XATTR_EXCL;
826 		req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
827 
828 		ceph_as_ctx_to_req(req, &as_ctx);
829 
830 		if (try_async && (req->r_dir_caps =
831 				  try_prep_async_create(dir, dentry, &lo,
832 							&req->r_deleg_ino))) {
833 			struct ceph_vino vino = { .ino = req->r_deleg_ino,
834 						  .snap = CEPH_NOSNAP };
835 			struct ceph_dentry_info *di = ceph_dentry(dentry);
836 
837 			set_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags);
838 			req->r_args.open.flags |= cpu_to_le32(CEPH_O_EXCL);
839 			req->r_callback = ceph_async_create_cb;
840 
841 			/* Hash inode before RPC */
842 			new_inode = ceph_get_inode(dir->i_sb, vino, new_inode);
843 			if (IS_ERR(new_inode)) {
844 				err = PTR_ERR(new_inode);
845 				new_inode = NULL;
846 				goto out_req;
847 			}
848 			WARN_ON_ONCE(!(new_inode->i_state & I_NEW));
849 
850 			spin_lock(&dentry->d_lock);
851 			di->flags |= CEPH_DENTRY_ASYNC_CREATE;
852 			spin_unlock(&dentry->d_lock);
853 
854 			err = ceph_mdsc_submit_request(mdsc, dir, req);
855 			if (!err) {
856 				err = ceph_finish_async_create(dir, new_inode,
857 							       dentry, file,
858 							       mode, req,
859 							       &as_ctx, &lo);
860 				new_inode = NULL;
861 			} else if (err == -EJUKEBOX) {
862 				restore_deleg_ino(dir, req->r_deleg_ino);
863 				ceph_mdsc_put_request(req);
864 				discard_new_inode(new_inode);
865 				ceph_release_acl_sec_ctx(&as_ctx);
866 				memset(&as_ctx, 0, sizeof(as_ctx));
867 				new_inode = NULL;
868 				try_async = false;
869 				ceph_put_string(rcu_dereference_raw(lo.pool_ns));
870 				goto retry;
871 			}
872 			ceph_put_string(rcu_dereference_raw(lo.pool_ns));
873 			goto out_req;
874 		}
875 	}
876 
877 	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
878 	req->r_new_inode = new_inode;
879 	new_inode = NULL;
880 	err = ceph_mdsc_do_request(mdsc, (flags & O_CREAT) ? dir : NULL, req);
881 	if (err == -ENOENT) {
882 		dentry = ceph_handle_snapdir(req, dentry);
883 		if (IS_ERR(dentry)) {
884 			err = PTR_ERR(dentry);
885 			goto out_req;
886 		}
887 		err = 0;
888 	}
889 
890 	if (!err && (flags & O_CREAT) && !req->r_reply_info.head->is_dentry)
891 		err = ceph_handle_notrace_create(dir, dentry);
892 
893 	if (d_in_lookup(dentry)) {
894 		dn = ceph_finish_lookup(req, dentry, err);
895 		if (IS_ERR(dn))
896 			err = PTR_ERR(dn);
897 	} else {
898 		/* we were given a hashed negative dentry */
899 		dn = NULL;
900 	}
901 	if (err)
902 		goto out_req;
903 	if (dn || d_really_is_negative(dentry) || d_is_symlink(dentry)) {
904 		/* make vfs retry on splice, ENOENT, or symlink */
905 		doutc(cl, "finish_no_open on dn %p\n", dn);
906 		err = finish_no_open(file, dn);
907 	} else {
908 		if (IS_ENCRYPTED(dir) &&
909 		    !fscrypt_has_permitted_context(dir, d_inode(dentry))) {
910 			pr_warn_client(cl,
911 				"Inconsistent encryption context (parent %llx:%llx child %llx:%llx)\n",
912 				ceph_vinop(dir), ceph_vinop(d_inode(dentry)));
913 			goto out_req;
914 		}
915 
916 		doutc(cl, "finish_open on dn %p\n", dn);
917 		if (req->r_op == CEPH_MDS_OP_CREATE && req->r_reply_info.has_create_ino) {
918 			struct inode *newino = d_inode(dentry);
919 
920 			cache_file_layout(dir, newino);
921 			ceph_init_inode_acls(newino, &as_ctx);
922 			file->f_mode |= FMODE_CREATED;
923 		}
924 		err = finish_open(file, dentry, ceph_open);
925 	}
926 out_req:
927 	ceph_mdsc_put_request(req);
928 	iput(new_inode);
929 out_ctx:
930 	ceph_release_acl_sec_ctx(&as_ctx);
931 	doutc(cl, "result=%d\n", err);
932 	return err;
933 }
934 
935 int ceph_release(struct inode *inode, struct file *file)
936 {
937 	struct ceph_client *cl = ceph_inode_to_client(inode);
938 	struct ceph_inode_info *ci = ceph_inode(inode);
939 
940 	if (S_ISDIR(inode->i_mode)) {
941 		struct ceph_dir_file_info *dfi = file->private_data;
942 		doutc(cl, "%p %llx.%llx dir file %p\n", inode,
943 		      ceph_vinop(inode), file);
944 		WARN_ON(!list_empty(&dfi->file_info.rw_contexts));
945 
946 		ceph_put_fmode(ci, dfi->file_info.fmode, 1);
947 
948 		if (dfi->last_readdir)
949 			ceph_mdsc_put_request(dfi->last_readdir);
950 		kfree(dfi->last_name);
951 		kfree(dfi->dir_info);
952 		kmem_cache_free(ceph_dir_file_cachep, dfi);
953 	} else {
954 		struct ceph_file_info *fi = file->private_data;
955 		doutc(cl, "%p %llx.%llx regular file %p\n", inode,
956 		      ceph_vinop(inode), file);
957 		WARN_ON(!list_empty(&fi->rw_contexts));
958 
959 		ceph_fscache_unuse_cookie(inode, file->f_mode & FMODE_WRITE);
960 		ceph_put_fmode(ci, fi->fmode, 1);
961 
962 		kmem_cache_free(ceph_file_cachep, fi);
963 	}
964 
965 	/* wake up anyone waiting for caps on this inode */
966 	wake_up_all(&ci->i_cap_wq);
967 	return 0;
968 }
969 
970 enum {
971 	HAVE_RETRIED = 1,
972 	CHECK_EOF =    2,
973 	READ_INLINE =  3,
974 };
975 
976 /*
977  * Completely synchronous read and write methods.  Direct from __user
978  * buffer to osd, or directly to user pages (if O_DIRECT).
979  *
980  * If the read spans object boundary, just do multiple reads.  (That's not
981  * atomic, but good enough for now.)
982  *
983  * If we get a short result from the OSD, check against i_size; we need to
984  * only return a short read to the caller if we hit EOF.
985  */
986 ssize_t __ceph_sync_read(struct inode *inode, loff_t *ki_pos,
987 			 struct iov_iter *to, int *retry_op,
988 			 u64 *last_objver)
989 {
990 	struct ceph_inode_info *ci = ceph_inode(inode);
991 	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
992 	struct ceph_client *cl = fsc->client;
993 	struct ceph_osd_client *osdc = &fsc->client->osdc;
994 	ssize_t ret;
995 	u64 off = *ki_pos;
996 	u64 len = iov_iter_count(to);
997 	u64 i_size = i_size_read(inode);
998 	bool sparse = IS_ENCRYPTED(inode) || ceph_test_mount_opt(fsc, SPARSEREAD);
999 	u64 objver = 0;
1000 
1001 	doutc(cl, "on inode %p %llx.%llx %llx~%llx\n", inode,
1002 	      ceph_vinop(inode), *ki_pos, len);
1003 
1004 	if (ceph_inode_is_shutdown(inode))
1005 		return -EIO;
1006 
1007 	if (!len)
1008 		return 0;
1009 	/*
1010 	 * flush any page cache pages in this range.  this
1011 	 * will make concurrent normal and sync io slow,
1012 	 * but it will at least behave sensibly when they are
1013 	 * in sequence.
1014 	 */
1015 	ret = filemap_write_and_wait_range(inode->i_mapping,
1016 					   off, off + len - 1);
1017 	if (ret < 0)
1018 		return ret;
1019 
1020 	ret = 0;
1021 	while ((len = iov_iter_count(to)) > 0) {
1022 		struct ceph_osd_request *req;
1023 		struct page **pages;
1024 		int num_pages;
1025 		size_t page_off;
1026 		bool more;
1027 		int idx;
1028 		size_t left;
1029 		struct ceph_osd_req_op *op;
1030 		u64 read_off = off;
1031 		u64 read_len = len;
1032 		int extent_cnt;
1033 
1034 		/* determine new offset/length if encrypted */
1035 		ceph_fscrypt_adjust_off_and_len(inode, &read_off, &read_len);
1036 
1037 		doutc(cl, "orig %llu~%llu reading %llu~%llu", off, len,
1038 		      read_off, read_len);
1039 
1040 		req = ceph_osdc_new_request(osdc, &ci->i_layout,
1041 					ci->i_vino, read_off, &read_len, 0, 1,
1042 					sparse ? CEPH_OSD_OP_SPARSE_READ :
1043 						 CEPH_OSD_OP_READ,
1044 					CEPH_OSD_FLAG_READ,
1045 					NULL, ci->i_truncate_seq,
1046 					ci->i_truncate_size, false);
1047 		if (IS_ERR(req)) {
1048 			ret = PTR_ERR(req);
1049 			break;
1050 		}
1051 
1052 		/* adjust len downward if the request truncated the len */
1053 		if (off + len > read_off + read_len)
1054 			len = read_off + read_len - off;
1055 		more = len < iov_iter_count(to);
1056 
1057 		num_pages = calc_pages_for(read_off, read_len);
1058 		page_off = offset_in_page(off);
1059 		pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1060 		if (IS_ERR(pages)) {
1061 			ceph_osdc_put_request(req);
1062 			ret = PTR_ERR(pages);
1063 			break;
1064 		}
1065 
1066 		osd_req_op_extent_osd_data_pages(req, 0, pages, read_len,
1067 						 offset_in_page(read_off),
1068 						 false, false);
1069 
1070 		op = &req->r_ops[0];
1071 		if (sparse) {
1072 			extent_cnt = __ceph_sparse_read_ext_count(inode, read_len);
1073 			ret = ceph_alloc_sparse_ext_map(op, extent_cnt);
1074 			if (ret) {
1075 				ceph_osdc_put_request(req);
1076 				break;
1077 			}
1078 		}
1079 
1080 		ceph_osdc_start_request(osdc, req);
1081 		ret = ceph_osdc_wait_request(osdc, req);
1082 
1083 		ceph_update_read_metrics(&fsc->mdsc->metric,
1084 					 req->r_start_latency,
1085 					 req->r_end_latency,
1086 					 read_len, ret);
1087 
1088 		if (ret > 0)
1089 			objver = req->r_version;
1090 
1091 		i_size = i_size_read(inode);
1092 		doutc(cl, "%llu~%llu got %zd i_size %llu%s\n", off, len,
1093 		      ret, i_size, (more ? " MORE" : ""));
1094 
1095 		/* Fix it to go to end of extent map */
1096 		if (sparse && ret >= 0)
1097 			ret = ceph_sparse_ext_map_end(op);
1098 		else if (ret == -ENOENT)
1099 			ret = 0;
1100 
1101 		if (ret > 0 && IS_ENCRYPTED(inode)) {
1102 			int fret;
1103 
1104 			fret = ceph_fscrypt_decrypt_extents(inode, pages,
1105 					read_off, op->extent.sparse_ext,
1106 					op->extent.sparse_ext_cnt);
1107 			if (fret < 0) {
1108 				ret = fret;
1109 				ceph_osdc_put_request(req);
1110 				break;
1111 			}
1112 
1113 			/* account for any partial block at the beginning */
1114 			fret -= (off - read_off);
1115 
1116 			/*
1117 			 * Short read after big offset adjustment?
1118 			 * Nothing is usable, just call it a zero
1119 			 * len read.
1120 			 */
1121 			fret = max(fret, 0);
1122 
1123 			/* account for partial block at the end */
1124 			ret = min_t(ssize_t, fret, len);
1125 		}
1126 
1127 		ceph_osdc_put_request(req);
1128 
1129 		/* Short read but not EOF? Zero out the remainder. */
1130 		if (ret >= 0 && ret < len && (off + ret < i_size)) {
1131 			int zlen = min(len - ret, i_size - off - ret);
1132 			int zoff = page_off + ret;
1133 
1134 			doutc(cl, "zero gap %llu~%llu\n", off + ret,
1135 			      off + ret + zlen);
1136 			ceph_zero_page_vector_range(zoff, zlen, pages);
1137 			ret += zlen;
1138 		}
1139 
1140 		idx = 0;
1141 		if (ret <= 0)
1142 			left = 0;
1143 		else if (off + ret > i_size)
1144 			left = i_size - off;
1145 		else
1146 			left = ret;
1147 		while (left > 0) {
1148 			size_t plen, copied;
1149 
1150 			plen = min_t(size_t, left, PAGE_SIZE - page_off);
1151 			SetPageUptodate(pages[idx]);
1152 			copied = copy_page_to_iter(pages[idx++],
1153 						   page_off, plen, to);
1154 			off += copied;
1155 			left -= copied;
1156 			page_off = 0;
1157 			if (copied < plen) {
1158 				ret = -EFAULT;
1159 				break;
1160 			}
1161 		}
1162 		ceph_release_page_vector(pages, num_pages);
1163 
1164 		if (ret < 0) {
1165 			if (ret == -EBLOCKLISTED)
1166 				fsc->blocklisted = true;
1167 			break;
1168 		}
1169 
1170 		if (off >= i_size || !more)
1171 			break;
1172 	}
1173 
1174 	if (ret > 0) {
1175 		if (off >= i_size) {
1176 			*retry_op = CHECK_EOF;
1177 			ret = i_size - *ki_pos;
1178 			*ki_pos = i_size;
1179 		} else {
1180 			ret = off - *ki_pos;
1181 			*ki_pos = off;
1182 		}
1183 
1184 		if (last_objver)
1185 			*last_objver = objver;
1186 	}
1187 	doutc(cl, "result %zd retry_op %d\n", ret, *retry_op);
1188 	return ret;
1189 }
1190 
1191 static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *to,
1192 			      int *retry_op)
1193 {
1194 	struct file *file = iocb->ki_filp;
1195 	struct inode *inode = file_inode(file);
1196 	struct ceph_client *cl = ceph_inode_to_client(inode);
1197 
1198 	doutc(cl, "on file %p %llx~%zx %s\n", file, iocb->ki_pos,
1199 	      iov_iter_count(to),
1200 	      (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
1201 
1202 	return __ceph_sync_read(inode, &iocb->ki_pos, to, retry_op, NULL);
1203 }
1204 
1205 struct ceph_aio_request {
1206 	struct kiocb *iocb;
1207 	size_t total_len;
1208 	bool write;
1209 	bool should_dirty;
1210 	int error;
1211 	struct list_head osd_reqs;
1212 	unsigned num_reqs;
1213 	atomic_t pending_reqs;
1214 	struct timespec64 mtime;
1215 	struct ceph_cap_flush *prealloc_cf;
1216 };
1217 
1218 struct ceph_aio_work {
1219 	struct work_struct work;
1220 	struct ceph_osd_request *req;
1221 };
1222 
1223 static void ceph_aio_retry_work(struct work_struct *work);
1224 
1225 static void ceph_aio_complete(struct inode *inode,
1226 			      struct ceph_aio_request *aio_req)
1227 {
1228 	struct ceph_client *cl = ceph_inode_to_client(inode);
1229 	struct ceph_inode_info *ci = ceph_inode(inode);
1230 	int ret;
1231 
1232 	if (!atomic_dec_and_test(&aio_req->pending_reqs))
1233 		return;
1234 
1235 	if (aio_req->iocb->ki_flags & IOCB_DIRECT)
1236 		inode_dio_end(inode);
1237 
1238 	ret = aio_req->error;
1239 	if (!ret)
1240 		ret = aio_req->total_len;
1241 
1242 	doutc(cl, "%p %llx.%llx rc %d\n", inode, ceph_vinop(inode), ret);
1243 
1244 	if (ret >= 0 && aio_req->write) {
1245 		int dirty;
1246 
1247 		loff_t endoff = aio_req->iocb->ki_pos + aio_req->total_len;
1248 		if (endoff > i_size_read(inode)) {
1249 			if (ceph_inode_set_size(inode, endoff))
1250 				ceph_check_caps(ci, CHECK_CAPS_AUTHONLY);
1251 		}
1252 
1253 		spin_lock(&ci->i_ceph_lock);
1254 		dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
1255 					       &aio_req->prealloc_cf);
1256 		spin_unlock(&ci->i_ceph_lock);
1257 		if (dirty)
1258 			__mark_inode_dirty(inode, dirty);
1259 
1260 	}
1261 
1262 	ceph_put_cap_refs(ci, (aio_req->write ? CEPH_CAP_FILE_WR :
1263 						CEPH_CAP_FILE_RD));
1264 
1265 	aio_req->iocb->ki_complete(aio_req->iocb, ret);
1266 
1267 	ceph_free_cap_flush(aio_req->prealloc_cf);
1268 	kfree(aio_req);
1269 }
1270 
1271 static void ceph_aio_complete_req(struct ceph_osd_request *req)
1272 {
1273 	int rc = req->r_result;
1274 	struct inode *inode = req->r_inode;
1275 	struct ceph_aio_request *aio_req = req->r_priv;
1276 	struct ceph_osd_data *osd_data = osd_req_op_extent_osd_data(req, 0);
1277 	struct ceph_osd_req_op *op = &req->r_ops[0];
1278 	struct ceph_client_metric *metric = &ceph_sb_to_mdsc(inode->i_sb)->metric;
1279 	unsigned int len = osd_data->bvec_pos.iter.bi_size;
1280 	bool sparse = (op->op == CEPH_OSD_OP_SPARSE_READ);
1281 	struct ceph_client *cl = ceph_inode_to_client(inode);
1282 
1283 	BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_BVECS);
1284 	BUG_ON(!osd_data->num_bvecs);
1285 
1286 	doutc(cl, "req %p inode %p %llx.%llx, rc %d bytes %u\n", req,
1287 	      inode, ceph_vinop(inode), rc, len);
1288 
1289 	if (rc == -EOLDSNAPC) {
1290 		struct ceph_aio_work *aio_work;
1291 		BUG_ON(!aio_req->write);
1292 
1293 		aio_work = kmalloc(sizeof(*aio_work), GFP_NOFS);
1294 		if (aio_work) {
1295 			INIT_WORK(&aio_work->work, ceph_aio_retry_work);
1296 			aio_work->req = req;
1297 			queue_work(ceph_inode_to_fs_client(inode)->inode_wq,
1298 				   &aio_work->work);
1299 			return;
1300 		}
1301 		rc = -ENOMEM;
1302 	} else if (!aio_req->write) {
1303 		if (sparse && rc >= 0)
1304 			rc = ceph_sparse_ext_map_end(op);
1305 		if (rc == -ENOENT)
1306 			rc = 0;
1307 		if (rc >= 0 && len > rc) {
1308 			struct iov_iter i;
1309 			int zlen = len - rc;
1310 
1311 			/*
1312 			 * If read is satisfied by single OSD request,
1313 			 * it can pass EOF. Otherwise read is within
1314 			 * i_size.
1315 			 */
1316 			if (aio_req->num_reqs == 1) {
1317 				loff_t i_size = i_size_read(inode);
1318 				loff_t endoff = aio_req->iocb->ki_pos + rc;
1319 				if (endoff < i_size)
1320 					zlen = min_t(size_t, zlen,
1321 						     i_size - endoff);
1322 				aio_req->total_len = rc + zlen;
1323 			}
1324 
1325 			iov_iter_bvec(&i, ITER_DEST, osd_data->bvec_pos.bvecs,
1326 				      osd_data->num_bvecs, len);
1327 			iov_iter_advance(&i, rc);
1328 			iov_iter_zero(zlen, &i);
1329 		}
1330 	}
1331 
1332 	/* r_start_latency == 0 means the request was not submitted */
1333 	if (req->r_start_latency) {
1334 		if (aio_req->write)
1335 			ceph_update_write_metrics(metric, req->r_start_latency,
1336 						  req->r_end_latency, len, rc);
1337 		else
1338 			ceph_update_read_metrics(metric, req->r_start_latency,
1339 						 req->r_end_latency, len, rc);
1340 	}
1341 
1342 	put_bvecs(osd_data->bvec_pos.bvecs, osd_data->num_bvecs,
1343 		  aio_req->should_dirty);
1344 	ceph_osdc_put_request(req);
1345 
1346 	if (rc < 0)
1347 		cmpxchg(&aio_req->error, 0, rc);
1348 
1349 	ceph_aio_complete(inode, aio_req);
1350 	return;
1351 }
1352 
1353 static void ceph_aio_retry_work(struct work_struct *work)
1354 {
1355 	struct ceph_aio_work *aio_work =
1356 		container_of(work, struct ceph_aio_work, work);
1357 	struct ceph_osd_request *orig_req = aio_work->req;
1358 	struct ceph_aio_request *aio_req = orig_req->r_priv;
1359 	struct inode *inode = orig_req->r_inode;
1360 	struct ceph_inode_info *ci = ceph_inode(inode);
1361 	struct ceph_snap_context *snapc;
1362 	struct ceph_osd_request *req;
1363 	int ret;
1364 
1365 	spin_lock(&ci->i_ceph_lock);
1366 	if (__ceph_have_pending_cap_snap(ci)) {
1367 		struct ceph_cap_snap *capsnap =
1368 			list_last_entry(&ci->i_cap_snaps,
1369 					struct ceph_cap_snap,
1370 					ci_item);
1371 		snapc = ceph_get_snap_context(capsnap->context);
1372 	} else {
1373 		BUG_ON(!ci->i_head_snapc);
1374 		snapc = ceph_get_snap_context(ci->i_head_snapc);
1375 	}
1376 	spin_unlock(&ci->i_ceph_lock);
1377 
1378 	req = ceph_osdc_alloc_request(orig_req->r_osdc, snapc, 1,
1379 			false, GFP_NOFS);
1380 	if (!req) {
1381 		ret = -ENOMEM;
1382 		req = orig_req;
1383 		goto out;
1384 	}
1385 
1386 	req->r_flags = /* CEPH_OSD_FLAG_ORDERSNAP | */ CEPH_OSD_FLAG_WRITE;
1387 	ceph_oloc_copy(&req->r_base_oloc, &orig_req->r_base_oloc);
1388 	ceph_oid_copy(&req->r_base_oid, &orig_req->r_base_oid);
1389 
1390 	req->r_ops[0] = orig_req->r_ops[0];
1391 
1392 	req->r_mtime = aio_req->mtime;
1393 	req->r_data_offset = req->r_ops[0].extent.offset;
1394 
1395 	ret = ceph_osdc_alloc_messages(req, GFP_NOFS);
1396 	if (ret) {
1397 		ceph_osdc_put_request(req);
1398 		req = orig_req;
1399 		goto out;
1400 	}
1401 
1402 	ceph_osdc_put_request(orig_req);
1403 
1404 	req->r_callback = ceph_aio_complete_req;
1405 	req->r_inode = inode;
1406 	req->r_priv = aio_req;
1407 
1408 	ceph_osdc_start_request(req->r_osdc, req);
1409 out:
1410 	if (ret < 0) {
1411 		req->r_result = ret;
1412 		ceph_aio_complete_req(req);
1413 	}
1414 
1415 	ceph_put_snap_context(snapc);
1416 	kfree(aio_work);
1417 }
1418 
1419 static ssize_t
1420 ceph_direct_read_write(struct kiocb *iocb, struct iov_iter *iter,
1421 		       struct ceph_snap_context *snapc,
1422 		       struct ceph_cap_flush **pcf)
1423 {
1424 	struct file *file = iocb->ki_filp;
1425 	struct inode *inode = file_inode(file);
1426 	struct ceph_inode_info *ci = ceph_inode(inode);
1427 	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1428 	struct ceph_client *cl = fsc->client;
1429 	struct ceph_client_metric *metric = &fsc->mdsc->metric;
1430 	struct ceph_vino vino;
1431 	struct ceph_osd_request *req;
1432 	struct bio_vec *bvecs;
1433 	struct ceph_aio_request *aio_req = NULL;
1434 	int num_pages = 0;
1435 	int flags;
1436 	int ret = 0;
1437 	struct timespec64 mtime = current_time(inode);
1438 	size_t count = iov_iter_count(iter);
1439 	loff_t pos = iocb->ki_pos;
1440 	bool write = iov_iter_rw(iter) == WRITE;
1441 	bool should_dirty = !write && user_backed_iter(iter);
1442 	bool sparse = ceph_test_mount_opt(fsc, SPARSEREAD);
1443 
1444 	if (write && ceph_snap(file_inode(file)) != CEPH_NOSNAP)
1445 		return -EROFS;
1446 
1447 	doutc(cl, "sync_direct_%s on file %p %lld~%u snapc %p seq %lld\n",
1448 	      (write ? "write" : "read"), file, pos, (unsigned)count,
1449 	      snapc, snapc ? snapc->seq : 0);
1450 
1451 	if (write) {
1452 		int ret2;
1453 
1454 		ceph_fscache_invalidate(inode, true);
1455 
1456 		ret2 = invalidate_inode_pages2_range(inode->i_mapping,
1457 					pos >> PAGE_SHIFT,
1458 					(pos + count - 1) >> PAGE_SHIFT);
1459 		if (ret2 < 0)
1460 			doutc(cl, "invalidate_inode_pages2_range returned %d\n",
1461 			      ret2);
1462 
1463 		flags = /* CEPH_OSD_FLAG_ORDERSNAP | */ CEPH_OSD_FLAG_WRITE;
1464 	} else {
1465 		flags = CEPH_OSD_FLAG_READ;
1466 	}
1467 
1468 	while (iov_iter_count(iter) > 0) {
1469 		u64 size = iov_iter_count(iter);
1470 		ssize_t len;
1471 		struct ceph_osd_req_op *op;
1472 		int readop = sparse ? CEPH_OSD_OP_SPARSE_READ : CEPH_OSD_OP_READ;
1473 		int extent_cnt;
1474 
1475 		if (write)
1476 			size = min_t(u64, size, fsc->mount_options->wsize);
1477 		else
1478 			size = min_t(u64, size, fsc->mount_options->rsize);
1479 
1480 		vino = ceph_vino(inode);
1481 		req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
1482 					    vino, pos, &size, 0,
1483 					    1,
1484 					    write ? CEPH_OSD_OP_WRITE : readop,
1485 					    flags, snapc,
1486 					    ci->i_truncate_seq,
1487 					    ci->i_truncate_size,
1488 					    false);
1489 		if (IS_ERR(req)) {
1490 			ret = PTR_ERR(req);
1491 			break;
1492 		}
1493 
1494 		len = iter_get_bvecs_alloc(iter, size, &bvecs, &num_pages);
1495 		if (len < 0) {
1496 			ceph_osdc_put_request(req);
1497 			ret = len;
1498 			break;
1499 		}
1500 		if (len != size)
1501 			osd_req_op_extent_update(req, 0, len);
1502 
1503 		/*
1504 		 * To simplify error handling, allow AIO when IO within i_size
1505 		 * or IO can be satisfied by single OSD request.
1506 		 */
1507 		if (pos == iocb->ki_pos && !is_sync_kiocb(iocb) &&
1508 		    (len == count || pos + count <= i_size_read(inode))) {
1509 			aio_req = kzalloc(sizeof(*aio_req), GFP_KERNEL);
1510 			if (aio_req) {
1511 				aio_req->iocb = iocb;
1512 				aio_req->write = write;
1513 				aio_req->should_dirty = should_dirty;
1514 				INIT_LIST_HEAD(&aio_req->osd_reqs);
1515 				if (write) {
1516 					aio_req->mtime = mtime;
1517 					swap(aio_req->prealloc_cf, *pcf);
1518 				}
1519 			}
1520 			/* ignore error */
1521 		}
1522 
1523 		if (write) {
1524 			/*
1525 			 * throw out any page cache pages in this range. this
1526 			 * may block.
1527 			 */
1528 			truncate_inode_pages_range(inode->i_mapping, pos,
1529 						   PAGE_ALIGN(pos + len) - 1);
1530 
1531 			req->r_mtime = mtime;
1532 		}
1533 
1534 		osd_req_op_extent_osd_data_bvecs(req, 0, bvecs, num_pages, len);
1535 		op = &req->r_ops[0];
1536 		if (sparse) {
1537 			extent_cnt = __ceph_sparse_read_ext_count(inode, size);
1538 			ret = ceph_alloc_sparse_ext_map(op, extent_cnt);
1539 			if (ret) {
1540 				ceph_osdc_put_request(req);
1541 				break;
1542 			}
1543 		}
1544 
1545 		if (aio_req) {
1546 			aio_req->total_len += len;
1547 			aio_req->num_reqs++;
1548 			atomic_inc(&aio_req->pending_reqs);
1549 
1550 			req->r_callback = ceph_aio_complete_req;
1551 			req->r_inode = inode;
1552 			req->r_priv = aio_req;
1553 			list_add_tail(&req->r_private_item, &aio_req->osd_reqs);
1554 
1555 			pos += len;
1556 			continue;
1557 		}
1558 
1559 		ceph_osdc_start_request(req->r_osdc, req);
1560 		ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
1561 
1562 		if (write)
1563 			ceph_update_write_metrics(metric, req->r_start_latency,
1564 						  req->r_end_latency, len, ret);
1565 		else
1566 			ceph_update_read_metrics(metric, req->r_start_latency,
1567 						 req->r_end_latency, len, ret);
1568 
1569 		size = i_size_read(inode);
1570 		if (!write) {
1571 			if (sparse && ret >= 0)
1572 				ret = ceph_sparse_ext_map_end(op);
1573 			else if (ret == -ENOENT)
1574 				ret = 0;
1575 
1576 			if (ret >= 0 && ret < len && pos + ret < size) {
1577 				struct iov_iter i;
1578 				int zlen = min_t(size_t, len - ret,
1579 						 size - pos - ret);
1580 
1581 				iov_iter_bvec(&i, ITER_DEST, bvecs, num_pages, len);
1582 				iov_iter_advance(&i, ret);
1583 				iov_iter_zero(zlen, &i);
1584 				ret += zlen;
1585 			}
1586 			if (ret >= 0)
1587 				len = ret;
1588 		}
1589 
1590 		put_bvecs(bvecs, num_pages, should_dirty);
1591 		ceph_osdc_put_request(req);
1592 		if (ret < 0)
1593 			break;
1594 
1595 		pos += len;
1596 		if (!write && pos >= size)
1597 			break;
1598 
1599 		if (write && pos > size) {
1600 			if (ceph_inode_set_size(inode, pos))
1601 				ceph_check_caps(ceph_inode(inode),
1602 						CHECK_CAPS_AUTHONLY);
1603 		}
1604 	}
1605 
1606 	if (aio_req) {
1607 		LIST_HEAD(osd_reqs);
1608 
1609 		if (aio_req->num_reqs == 0) {
1610 			kfree(aio_req);
1611 			return ret;
1612 		}
1613 
1614 		ceph_get_cap_refs(ci, write ? CEPH_CAP_FILE_WR :
1615 					      CEPH_CAP_FILE_RD);
1616 
1617 		list_splice(&aio_req->osd_reqs, &osd_reqs);
1618 		inode_dio_begin(inode);
1619 		while (!list_empty(&osd_reqs)) {
1620 			req = list_first_entry(&osd_reqs,
1621 					       struct ceph_osd_request,
1622 					       r_private_item);
1623 			list_del_init(&req->r_private_item);
1624 			if (ret >= 0)
1625 				ceph_osdc_start_request(req->r_osdc, req);
1626 			if (ret < 0) {
1627 				req->r_result = ret;
1628 				ceph_aio_complete_req(req);
1629 			}
1630 		}
1631 		return -EIOCBQUEUED;
1632 	}
1633 
1634 	if (ret != -EOLDSNAPC && pos > iocb->ki_pos) {
1635 		ret = pos - iocb->ki_pos;
1636 		iocb->ki_pos = pos;
1637 	}
1638 	return ret;
1639 }
1640 
1641 /*
1642  * Synchronous write, straight from __user pointer or user pages.
1643  *
1644  * If write spans object boundary, just do multiple writes.  (For a
1645  * correct atomic write, we should e.g. take write locks on all
1646  * objects, rollback on failure, etc.)
1647  */
1648 static ssize_t
1649 ceph_sync_write(struct kiocb *iocb, struct iov_iter *from, loff_t pos,
1650 		struct ceph_snap_context *snapc)
1651 {
1652 	struct file *file = iocb->ki_filp;
1653 	struct inode *inode = file_inode(file);
1654 	struct ceph_inode_info *ci = ceph_inode(inode);
1655 	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1656 	struct ceph_client *cl = fsc->client;
1657 	struct ceph_osd_client *osdc = &fsc->client->osdc;
1658 	struct ceph_osd_request *req;
1659 	struct page **pages;
1660 	u64 len;
1661 	int num_pages;
1662 	int written = 0;
1663 	int ret;
1664 	bool check_caps = false;
1665 	struct timespec64 mtime = current_time(inode);
1666 	size_t count = iov_iter_count(from);
1667 
1668 	if (ceph_snap(file_inode(file)) != CEPH_NOSNAP)
1669 		return -EROFS;
1670 
1671 	doutc(cl, "on file %p %lld~%u snapc %p seq %lld\n", file, pos,
1672 	      (unsigned)count, snapc, snapc->seq);
1673 
1674 	ret = filemap_write_and_wait_range(inode->i_mapping,
1675 					   pos, pos + count - 1);
1676 	if (ret < 0)
1677 		return ret;
1678 
1679 	ceph_fscache_invalidate(inode, false);
1680 
1681 	while ((len = iov_iter_count(from)) > 0) {
1682 		size_t left;
1683 		int n;
1684 		u64 write_pos = pos;
1685 		u64 write_len = len;
1686 		u64 objnum, objoff;
1687 		u32 xlen;
1688 		u64 assert_ver = 0;
1689 		bool rmw;
1690 		bool first, last;
1691 		struct iov_iter saved_iter = *from;
1692 		size_t off;
1693 
1694 		ceph_fscrypt_adjust_off_and_len(inode, &write_pos, &write_len);
1695 
1696 		/* clamp the length to the end of first object */
1697 		ceph_calc_file_object_mapping(&ci->i_layout, write_pos,
1698 					      write_len, &objnum, &objoff,
1699 					      &xlen);
1700 		write_len = xlen;
1701 
1702 		/* adjust len downward if it goes beyond current object */
1703 		if (pos + len > write_pos + write_len)
1704 			len = write_pos + write_len - pos;
1705 
1706 		/*
1707 		 * If we had to adjust the length or position to align with a
1708 		 * crypto block, then we must do a read/modify/write cycle. We
1709 		 * use a version assertion to redrive the thing if something
1710 		 * changes in between.
1711 		 */
1712 		first = pos != write_pos;
1713 		last = (pos + len) != (write_pos + write_len);
1714 		rmw = first || last;
1715 
1716 		doutc(cl, "ino %llx %lld~%llu adjusted %lld~%llu -- %srmw\n",
1717 		      ci->i_vino.ino, pos, len, write_pos, write_len,
1718 		      rmw ? "" : "no ");
1719 
1720 		/*
1721 		 * The data is emplaced into the page as it would be if it were
1722 		 * in an array of pagecache pages.
1723 		 */
1724 		num_pages = calc_pages_for(write_pos, write_len);
1725 		pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1726 		if (IS_ERR(pages)) {
1727 			ret = PTR_ERR(pages);
1728 			break;
1729 		}
1730 
1731 		/* Do we need to preload the pages? */
1732 		if (rmw) {
1733 			u64 first_pos = write_pos;
1734 			u64 last_pos = (write_pos + write_len) - CEPH_FSCRYPT_BLOCK_SIZE;
1735 			u64 read_len = CEPH_FSCRYPT_BLOCK_SIZE;
1736 			struct ceph_osd_req_op *op;
1737 
1738 			/* We should only need to do this for encrypted inodes */
1739 			WARN_ON_ONCE(!IS_ENCRYPTED(inode));
1740 
1741 			/* No need to do two reads if first and last blocks are same */
1742 			if (first && last_pos == first_pos)
1743 				last = false;
1744 
1745 			/*
1746 			 * Allocate a read request for one or two extents,
1747 			 * depending on how the request was aligned.
1748 			 */
1749 			req = ceph_osdc_new_request(osdc, &ci->i_layout,
1750 					ci->i_vino, first ? first_pos : last_pos,
1751 					&read_len, 0, (first && last) ? 2 : 1,
1752 					CEPH_OSD_OP_SPARSE_READ, CEPH_OSD_FLAG_READ,
1753 					NULL, ci->i_truncate_seq,
1754 					ci->i_truncate_size, false);
1755 			if (IS_ERR(req)) {
1756 				ceph_release_page_vector(pages, num_pages);
1757 				ret = PTR_ERR(req);
1758 				break;
1759 			}
1760 
1761 			/* Something is misaligned! */
1762 			if (read_len != CEPH_FSCRYPT_BLOCK_SIZE) {
1763 				ceph_osdc_put_request(req);
1764 				ceph_release_page_vector(pages, num_pages);
1765 				ret = -EIO;
1766 				break;
1767 			}
1768 
1769 			/* Add extent for first block? */
1770 			op = &req->r_ops[0];
1771 
1772 			if (first) {
1773 				osd_req_op_extent_osd_data_pages(req, 0, pages,
1774 							 CEPH_FSCRYPT_BLOCK_SIZE,
1775 							 offset_in_page(first_pos),
1776 							 false, false);
1777 				/* We only expect a single extent here */
1778 				ret = __ceph_alloc_sparse_ext_map(op, 1);
1779 				if (ret) {
1780 					ceph_osdc_put_request(req);
1781 					ceph_release_page_vector(pages, num_pages);
1782 					break;
1783 				}
1784 			}
1785 
1786 			/* Add extent for last block */
1787 			if (last) {
1788 				/* Init the other extent if first extent has been used */
1789 				if (first) {
1790 					op = &req->r_ops[1];
1791 					osd_req_op_extent_init(req, 1,
1792 							CEPH_OSD_OP_SPARSE_READ,
1793 							last_pos, CEPH_FSCRYPT_BLOCK_SIZE,
1794 							ci->i_truncate_size,
1795 							ci->i_truncate_seq);
1796 				}
1797 
1798 				ret = __ceph_alloc_sparse_ext_map(op, 1);
1799 				if (ret) {
1800 					ceph_osdc_put_request(req);
1801 					ceph_release_page_vector(pages, num_pages);
1802 					break;
1803 				}
1804 
1805 				osd_req_op_extent_osd_data_pages(req, first ? 1 : 0,
1806 							&pages[num_pages - 1],
1807 							CEPH_FSCRYPT_BLOCK_SIZE,
1808 							offset_in_page(last_pos),
1809 							false, false);
1810 			}
1811 
1812 			ceph_osdc_start_request(osdc, req);
1813 			ret = ceph_osdc_wait_request(osdc, req);
1814 
1815 			/* FIXME: length field is wrong if there are 2 extents */
1816 			ceph_update_read_metrics(&fsc->mdsc->metric,
1817 						 req->r_start_latency,
1818 						 req->r_end_latency,
1819 						 read_len, ret);
1820 
1821 			/* Ok if object is not already present */
1822 			if (ret == -ENOENT) {
1823 				/*
1824 				 * If there is no object, then we can't assert
1825 				 * on its version. Set it to 0, and we'll use an
1826 				 * exclusive create instead.
1827 				 */
1828 				ceph_osdc_put_request(req);
1829 				ret = 0;
1830 
1831 				/*
1832 				 * zero out the soon-to-be uncopied parts of the
1833 				 * first and last pages.
1834 				 */
1835 				if (first)
1836 					zero_user_segment(pages[0], 0,
1837 							  offset_in_page(first_pos));
1838 				if (last)
1839 					zero_user_segment(pages[num_pages - 1],
1840 							  offset_in_page(last_pos),
1841 							  PAGE_SIZE);
1842 			} else {
1843 				if (ret < 0) {
1844 					ceph_osdc_put_request(req);
1845 					ceph_release_page_vector(pages, num_pages);
1846 					break;
1847 				}
1848 
1849 				op = &req->r_ops[0];
1850 				if (op->extent.sparse_ext_cnt == 0) {
1851 					if (first)
1852 						zero_user_segment(pages[0], 0,
1853 								  offset_in_page(first_pos));
1854 					else
1855 						zero_user_segment(pages[num_pages - 1],
1856 								  offset_in_page(last_pos),
1857 								  PAGE_SIZE);
1858 				} else if (op->extent.sparse_ext_cnt != 1 ||
1859 					   ceph_sparse_ext_map_end(op) !=
1860 						CEPH_FSCRYPT_BLOCK_SIZE) {
1861 					ret = -EIO;
1862 					ceph_osdc_put_request(req);
1863 					ceph_release_page_vector(pages, num_pages);
1864 					break;
1865 				}
1866 
1867 				if (first && last) {
1868 					op = &req->r_ops[1];
1869 					if (op->extent.sparse_ext_cnt == 0) {
1870 						zero_user_segment(pages[num_pages - 1],
1871 								  offset_in_page(last_pos),
1872 								  PAGE_SIZE);
1873 					} else if (op->extent.sparse_ext_cnt != 1 ||
1874 						   ceph_sparse_ext_map_end(op) !=
1875 							CEPH_FSCRYPT_BLOCK_SIZE) {
1876 						ret = -EIO;
1877 						ceph_osdc_put_request(req);
1878 						ceph_release_page_vector(pages, num_pages);
1879 						break;
1880 					}
1881 				}
1882 
1883 				/* Grab assert version. It must be non-zero. */
1884 				assert_ver = req->r_version;
1885 				WARN_ON_ONCE(ret > 0 && assert_ver == 0);
1886 
1887 				ceph_osdc_put_request(req);
1888 				if (first) {
1889 					ret = ceph_fscrypt_decrypt_block_inplace(inode,
1890 							pages[0], CEPH_FSCRYPT_BLOCK_SIZE,
1891 							offset_in_page(first_pos),
1892 							first_pos >> CEPH_FSCRYPT_BLOCK_SHIFT);
1893 					if (ret < 0) {
1894 						ceph_release_page_vector(pages, num_pages);
1895 						break;
1896 					}
1897 				}
1898 				if (last) {
1899 					ret = ceph_fscrypt_decrypt_block_inplace(inode,
1900 							pages[num_pages - 1],
1901 							CEPH_FSCRYPT_BLOCK_SIZE,
1902 							offset_in_page(last_pos),
1903 							last_pos >> CEPH_FSCRYPT_BLOCK_SHIFT);
1904 					if (ret < 0) {
1905 						ceph_release_page_vector(pages, num_pages);
1906 						break;
1907 					}
1908 				}
1909 			}
1910 		}
1911 
1912 		left = len;
1913 		off = offset_in_page(pos);
1914 		for (n = 0; n < num_pages; n++) {
1915 			size_t plen = min_t(size_t, left, PAGE_SIZE - off);
1916 
1917 			/* copy the data */
1918 			ret = copy_page_from_iter(pages[n], off, plen, from);
1919 			if (ret != plen) {
1920 				ret = -EFAULT;
1921 				break;
1922 			}
1923 			off = 0;
1924 			left -= ret;
1925 		}
1926 		if (ret < 0) {
1927 			doutc(cl, "write failed with %d\n", ret);
1928 			ceph_release_page_vector(pages, num_pages);
1929 			break;
1930 		}
1931 
1932 		if (IS_ENCRYPTED(inode)) {
1933 			ret = ceph_fscrypt_encrypt_pages(inode, pages,
1934 							 write_pos, write_len,
1935 							 GFP_KERNEL);
1936 			if (ret < 0) {
1937 				doutc(cl, "encryption failed with %d\n", ret);
1938 				ceph_release_page_vector(pages, num_pages);
1939 				break;
1940 			}
1941 		}
1942 
1943 		req = ceph_osdc_new_request(osdc, &ci->i_layout,
1944 					    ci->i_vino, write_pos, &write_len,
1945 					    rmw ? 1 : 0, rmw ? 2 : 1,
1946 					    CEPH_OSD_OP_WRITE,
1947 					    CEPH_OSD_FLAG_WRITE,
1948 					    snapc, ci->i_truncate_seq,
1949 					    ci->i_truncate_size, false);
1950 		if (IS_ERR(req)) {
1951 			ret = PTR_ERR(req);
1952 			ceph_release_page_vector(pages, num_pages);
1953 			break;
1954 		}
1955 
1956 		doutc(cl, "write op %lld~%llu\n", write_pos, write_len);
1957 		osd_req_op_extent_osd_data_pages(req, rmw ? 1 : 0, pages, write_len,
1958 						 offset_in_page(write_pos), false,
1959 						 true);
1960 		req->r_inode = inode;
1961 		req->r_mtime = mtime;
1962 
1963 		/* Set up the assertion */
1964 		if (rmw) {
1965 			/*
1966 			 * Set up the assertion. If we don't have a version
1967 			 * number, then the object doesn't exist yet. Use an
1968 			 * exclusive create instead of a version assertion in
1969 			 * that case.
1970 			 */
1971 			if (assert_ver) {
1972 				osd_req_op_init(req, 0, CEPH_OSD_OP_ASSERT_VER, 0);
1973 				req->r_ops[0].assert_ver.ver = assert_ver;
1974 			} else {
1975 				osd_req_op_init(req, 0, CEPH_OSD_OP_CREATE,
1976 						CEPH_OSD_OP_FLAG_EXCL);
1977 			}
1978 		}
1979 
1980 		ceph_osdc_start_request(osdc, req);
1981 		ret = ceph_osdc_wait_request(osdc, req);
1982 
1983 		ceph_update_write_metrics(&fsc->mdsc->metric, req->r_start_latency,
1984 					  req->r_end_latency, len, ret);
1985 		ceph_osdc_put_request(req);
1986 		if (ret != 0) {
1987 			doutc(cl, "osd write returned %d\n", ret);
1988 			/* Version changed! Must re-do the rmw cycle */
1989 			if ((assert_ver && (ret == -ERANGE || ret == -EOVERFLOW)) ||
1990 			    (!assert_ver && ret == -EEXIST)) {
1991 				/* We should only ever see this on a rmw */
1992 				WARN_ON_ONCE(!rmw);
1993 
1994 				/* The version should never go backward */
1995 				WARN_ON_ONCE(ret == -EOVERFLOW);
1996 
1997 				*from = saved_iter;
1998 
1999 				/* FIXME: limit number of times we loop? */
2000 				continue;
2001 			}
2002 			ceph_set_error_write(ci);
2003 			break;
2004 		}
2005 
2006 		ceph_clear_error_write(ci);
2007 
2008 		/*
2009 		 * We successfully wrote to a range of the file. Declare
2010 		 * that region of the pagecache invalid.
2011 		 */
2012 		ret = invalidate_inode_pages2_range(
2013 				inode->i_mapping,
2014 				pos >> PAGE_SHIFT,
2015 				(pos + len - 1) >> PAGE_SHIFT);
2016 		if (ret < 0) {
2017 			doutc(cl, "invalidate_inode_pages2_range returned %d\n",
2018 			      ret);
2019 			ret = 0;
2020 		}
2021 		pos += len;
2022 		written += len;
2023 		doutc(cl, "written %d\n", written);
2024 		if (pos > i_size_read(inode)) {
2025 			check_caps = ceph_inode_set_size(inode, pos);
2026 			if (check_caps)
2027 				ceph_check_caps(ceph_inode(inode),
2028 						CHECK_CAPS_AUTHONLY);
2029 		}
2030 
2031 	}
2032 
2033 	if (ret != -EOLDSNAPC && written > 0) {
2034 		ret = written;
2035 		iocb->ki_pos = pos;
2036 	}
2037 	doutc(cl, "returning %d\n", ret);
2038 	return ret;
2039 }
2040 
2041 /*
2042  * Wrap generic_file_aio_read with checks for cap bits on the inode.
2043  * Atomically grab references, so that those bits are not released
2044  * back to the MDS mid-read.
2045  *
2046  * Hmm, the sync read case isn't actually async... should it be?
2047  */
2048 static ssize_t ceph_read_iter(struct kiocb *iocb, struct iov_iter *to)
2049 {
2050 	struct file *filp = iocb->ki_filp;
2051 	struct ceph_file_info *fi = filp->private_data;
2052 	size_t len = iov_iter_count(to);
2053 	struct inode *inode = file_inode(filp);
2054 	struct ceph_inode_info *ci = ceph_inode(inode);
2055 	bool direct_lock = iocb->ki_flags & IOCB_DIRECT;
2056 	struct ceph_client *cl = ceph_inode_to_client(inode);
2057 	ssize_t ret;
2058 	int want = 0, got = 0;
2059 	int retry_op = 0, read = 0;
2060 
2061 again:
2062 	doutc(cl, "%llu~%u trying to get caps on %p %llx.%llx\n",
2063 	      iocb->ki_pos, (unsigned)len, inode, ceph_vinop(inode));
2064 
2065 	if (ceph_inode_is_shutdown(inode))
2066 		return -ESTALE;
2067 
2068 	if (direct_lock)
2069 		ceph_start_io_direct(inode);
2070 	else
2071 		ceph_start_io_read(inode);
2072 
2073 	if (!(fi->flags & CEPH_F_SYNC) && !direct_lock)
2074 		want |= CEPH_CAP_FILE_CACHE;
2075 	if (fi->fmode & CEPH_FILE_MODE_LAZY)
2076 		want |= CEPH_CAP_FILE_LAZYIO;
2077 
2078 	ret = ceph_get_caps(filp, CEPH_CAP_FILE_RD, want, -1, &got);
2079 	if (ret < 0) {
2080 		if (direct_lock)
2081 			ceph_end_io_direct(inode);
2082 		else
2083 			ceph_end_io_read(inode);
2084 		return ret;
2085 	}
2086 
2087 	if ((got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0 ||
2088 	    (iocb->ki_flags & IOCB_DIRECT) ||
2089 	    (fi->flags & CEPH_F_SYNC)) {
2090 
2091 		doutc(cl, "sync %p %llx.%llx %llu~%u got cap refs on %s\n",
2092 		      inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len,
2093 		      ceph_cap_string(got));
2094 
2095 		if (!ceph_has_inline_data(ci)) {
2096 			if (!retry_op &&
2097 			    (iocb->ki_flags & IOCB_DIRECT) &&
2098 			    !IS_ENCRYPTED(inode)) {
2099 				ret = ceph_direct_read_write(iocb, to,
2100 							     NULL, NULL);
2101 				if (ret >= 0 && ret < len)
2102 					retry_op = CHECK_EOF;
2103 			} else {
2104 				ret = ceph_sync_read(iocb, to, &retry_op);
2105 			}
2106 		} else {
2107 			retry_op = READ_INLINE;
2108 		}
2109 	} else {
2110 		CEPH_DEFINE_RW_CONTEXT(rw_ctx, got);
2111 		doutc(cl, "async %p %llx.%llx %llu~%u got cap refs on %s\n",
2112 		      inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len,
2113 		      ceph_cap_string(got));
2114 		ceph_add_rw_context(fi, &rw_ctx);
2115 		ret = generic_file_read_iter(iocb, to);
2116 		ceph_del_rw_context(fi, &rw_ctx);
2117 	}
2118 
2119 	doutc(cl, "%p %llx.%llx dropping cap refs on %s = %d\n",
2120 	      inode, ceph_vinop(inode), ceph_cap_string(got), (int)ret);
2121 	ceph_put_cap_refs(ci, got);
2122 
2123 	if (direct_lock)
2124 		ceph_end_io_direct(inode);
2125 	else
2126 		ceph_end_io_read(inode);
2127 
2128 	if (retry_op > HAVE_RETRIED && ret >= 0) {
2129 		int statret;
2130 		struct page *page = NULL;
2131 		loff_t i_size;
2132 		int mask = CEPH_STAT_CAP_SIZE;
2133 		if (retry_op == READ_INLINE) {
2134 			page = __page_cache_alloc(GFP_KERNEL);
2135 			if (!page)
2136 				return -ENOMEM;
2137 
2138 			mask = CEPH_STAT_CAP_INLINE_DATA;
2139 		}
2140 
2141 		statret = __ceph_do_getattr(inode, page, mask, !!page);
2142 		if (statret < 0) {
2143 			if (page)
2144 				__free_page(page);
2145 			if (statret == -ENODATA) {
2146 				BUG_ON(retry_op != READ_INLINE);
2147 				goto again;
2148 			}
2149 			return statret;
2150 		}
2151 
2152 		i_size = i_size_read(inode);
2153 		if (retry_op == READ_INLINE) {
2154 			BUG_ON(ret > 0 || read > 0);
2155 			if (iocb->ki_pos < i_size &&
2156 			    iocb->ki_pos < PAGE_SIZE) {
2157 				loff_t end = min_t(loff_t, i_size,
2158 						   iocb->ki_pos + len);
2159 				end = min_t(loff_t, end, PAGE_SIZE);
2160 				if (statret < end)
2161 					zero_user_segment(page, statret, end);
2162 				ret = copy_page_to_iter(page,
2163 						iocb->ki_pos & ~PAGE_MASK,
2164 						end - iocb->ki_pos, to);
2165 				iocb->ki_pos += ret;
2166 				read += ret;
2167 			}
2168 			if (iocb->ki_pos < i_size && read < len) {
2169 				size_t zlen = min_t(size_t, len - read,
2170 						    i_size - iocb->ki_pos);
2171 				ret = iov_iter_zero(zlen, to);
2172 				iocb->ki_pos += ret;
2173 				read += ret;
2174 			}
2175 			__free_pages(page, 0);
2176 			return read;
2177 		}
2178 
2179 		/* hit EOF or hole? */
2180 		if (retry_op == CHECK_EOF && iocb->ki_pos < i_size &&
2181 		    ret < len) {
2182 			doutc(cl, "may hit hole, ppos %lld < size %lld, reading more\n",
2183 			      iocb->ki_pos, i_size);
2184 
2185 			read += ret;
2186 			len -= ret;
2187 			retry_op = HAVE_RETRIED;
2188 			goto again;
2189 		}
2190 	}
2191 
2192 	if (ret >= 0)
2193 		ret += read;
2194 
2195 	return ret;
2196 }
2197 
2198 /*
2199  * Wrap filemap_splice_read with checks for cap bits on the inode.
2200  * Atomically grab references, so that those bits are not released
2201  * back to the MDS mid-read.
2202  */
2203 static ssize_t ceph_splice_read(struct file *in, loff_t *ppos,
2204 				struct pipe_inode_info *pipe,
2205 				size_t len, unsigned int flags)
2206 {
2207 	struct ceph_file_info *fi = in->private_data;
2208 	struct inode *inode = file_inode(in);
2209 	struct ceph_inode_info *ci = ceph_inode(inode);
2210 	ssize_t ret;
2211 	int want = 0, got = 0;
2212 	CEPH_DEFINE_RW_CONTEXT(rw_ctx, 0);
2213 
2214 	dout("splice_read %p %llx.%llx %llu~%zu trying to get caps on %p\n",
2215 	     inode, ceph_vinop(inode), *ppos, len, inode);
2216 
2217 	if (ceph_inode_is_shutdown(inode))
2218 		return -ESTALE;
2219 
2220 	if (ceph_has_inline_data(ci) ||
2221 	    (fi->flags & CEPH_F_SYNC))
2222 		return copy_splice_read(in, ppos, pipe, len, flags);
2223 
2224 	ceph_start_io_read(inode);
2225 
2226 	want = CEPH_CAP_FILE_CACHE;
2227 	if (fi->fmode & CEPH_FILE_MODE_LAZY)
2228 		want |= CEPH_CAP_FILE_LAZYIO;
2229 
2230 	ret = ceph_get_caps(in, CEPH_CAP_FILE_RD, want, -1, &got);
2231 	if (ret < 0)
2232 		goto out_end;
2233 
2234 	if ((got & (CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO)) == 0) {
2235 		dout("splice_read/sync %p %llx.%llx %llu~%zu got cap refs on %s\n",
2236 		     inode, ceph_vinop(inode), *ppos, len,
2237 		     ceph_cap_string(got));
2238 
2239 		ceph_put_cap_refs(ci, got);
2240 		ceph_end_io_read(inode);
2241 		return copy_splice_read(in, ppos, pipe, len, flags);
2242 	}
2243 
2244 	dout("splice_read %p %llx.%llx %llu~%zu got cap refs on %s\n",
2245 	     inode, ceph_vinop(inode), *ppos, len, ceph_cap_string(got));
2246 
2247 	rw_ctx.caps = got;
2248 	ceph_add_rw_context(fi, &rw_ctx);
2249 	ret = filemap_splice_read(in, ppos, pipe, len, flags);
2250 	ceph_del_rw_context(fi, &rw_ctx);
2251 
2252 	dout("splice_read %p %llx.%llx dropping cap refs on %s = %zd\n",
2253 	     inode, ceph_vinop(inode), ceph_cap_string(got), ret);
2254 
2255 	ceph_put_cap_refs(ci, got);
2256 out_end:
2257 	ceph_end_io_read(inode);
2258 	return ret;
2259 }
2260 
2261 /*
2262  * Take cap references to avoid releasing caps to MDS mid-write.
2263  *
2264  * If we are synchronous, and write with an old snap context, the OSD
2265  * may return EOLDSNAPC.  In that case, retry the write.. _after_
2266  * dropping our cap refs and allowing the pending snap to logically
2267  * complete _before_ this write occurs.
2268  *
2269  * If we are near ENOSPC, write synchronously.
2270  */
2271 static ssize_t ceph_write_iter(struct kiocb *iocb, struct iov_iter *from)
2272 {
2273 	struct file *file = iocb->ki_filp;
2274 	struct ceph_file_info *fi = file->private_data;
2275 	struct inode *inode = file_inode(file);
2276 	struct ceph_inode_info *ci = ceph_inode(inode);
2277 	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
2278 	struct ceph_client *cl = fsc->client;
2279 	struct ceph_osd_client *osdc = &fsc->client->osdc;
2280 	struct ceph_cap_flush *prealloc_cf;
2281 	ssize_t count, written = 0;
2282 	int err, want = 0, got;
2283 	bool direct_lock = false;
2284 	u32 map_flags;
2285 	u64 pool_flags;
2286 	loff_t pos;
2287 	loff_t limit = max(i_size_read(inode), fsc->max_file_size);
2288 
2289 	if (ceph_inode_is_shutdown(inode))
2290 		return -ESTALE;
2291 
2292 	if (ceph_snap(inode) != CEPH_NOSNAP)
2293 		return -EROFS;
2294 
2295 	prealloc_cf = ceph_alloc_cap_flush();
2296 	if (!prealloc_cf)
2297 		return -ENOMEM;
2298 
2299 	if ((iocb->ki_flags & (IOCB_DIRECT | IOCB_APPEND)) == IOCB_DIRECT)
2300 		direct_lock = true;
2301 
2302 retry_snap:
2303 	if (direct_lock)
2304 		ceph_start_io_direct(inode);
2305 	else
2306 		ceph_start_io_write(inode);
2307 
2308 	if (iocb->ki_flags & IOCB_APPEND) {
2309 		err = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE, false);
2310 		if (err < 0)
2311 			goto out;
2312 	}
2313 
2314 	err = generic_write_checks(iocb, from);
2315 	if (err <= 0)
2316 		goto out;
2317 
2318 	pos = iocb->ki_pos;
2319 	if (unlikely(pos >= limit)) {
2320 		err = -EFBIG;
2321 		goto out;
2322 	} else {
2323 		iov_iter_truncate(from, limit - pos);
2324 	}
2325 
2326 	count = iov_iter_count(from);
2327 	if (ceph_quota_is_max_bytes_exceeded(inode, pos + count)) {
2328 		err = -EDQUOT;
2329 		goto out;
2330 	}
2331 
2332 	down_read(&osdc->lock);
2333 	map_flags = osdc->osdmap->flags;
2334 	pool_flags = ceph_pg_pool_flags(osdc->osdmap, ci->i_layout.pool_id);
2335 	up_read(&osdc->lock);
2336 	if ((map_flags & CEPH_OSDMAP_FULL) ||
2337 	    (pool_flags & CEPH_POOL_FLAG_FULL)) {
2338 		err = -ENOSPC;
2339 		goto out;
2340 	}
2341 
2342 	err = file_remove_privs(file);
2343 	if (err)
2344 		goto out;
2345 
2346 	doutc(cl, "%p %llx.%llx %llu~%zd getting caps. i_size %llu\n",
2347 	      inode, ceph_vinop(inode), pos, count,
2348 	      i_size_read(inode));
2349 	if (!(fi->flags & CEPH_F_SYNC) && !direct_lock)
2350 		want |= CEPH_CAP_FILE_BUFFER;
2351 	if (fi->fmode & CEPH_FILE_MODE_LAZY)
2352 		want |= CEPH_CAP_FILE_LAZYIO;
2353 	got = 0;
2354 	err = ceph_get_caps(file, CEPH_CAP_FILE_WR, want, pos + count, &got);
2355 	if (err < 0)
2356 		goto out;
2357 
2358 	err = file_update_time(file);
2359 	if (err)
2360 		goto out_caps;
2361 
2362 	inode_inc_iversion_raw(inode);
2363 
2364 	doutc(cl, "%p %llx.%llx %llu~%zd got cap refs on %s\n",
2365 	      inode, ceph_vinop(inode), pos, count, ceph_cap_string(got));
2366 
2367 	if ((got & (CEPH_CAP_FILE_BUFFER|CEPH_CAP_FILE_LAZYIO)) == 0 ||
2368 	    (iocb->ki_flags & IOCB_DIRECT) || (fi->flags & CEPH_F_SYNC) ||
2369 	    (ci->i_ceph_flags & CEPH_I_ERROR_WRITE)) {
2370 		struct ceph_snap_context *snapc;
2371 		struct iov_iter data;
2372 
2373 		spin_lock(&ci->i_ceph_lock);
2374 		if (__ceph_have_pending_cap_snap(ci)) {
2375 			struct ceph_cap_snap *capsnap =
2376 					list_last_entry(&ci->i_cap_snaps,
2377 							struct ceph_cap_snap,
2378 							ci_item);
2379 			snapc = ceph_get_snap_context(capsnap->context);
2380 		} else {
2381 			BUG_ON(!ci->i_head_snapc);
2382 			snapc = ceph_get_snap_context(ci->i_head_snapc);
2383 		}
2384 		spin_unlock(&ci->i_ceph_lock);
2385 
2386 		/* we might need to revert back to that point */
2387 		data = *from;
2388 		if ((iocb->ki_flags & IOCB_DIRECT) && !IS_ENCRYPTED(inode))
2389 			written = ceph_direct_read_write(iocb, &data, snapc,
2390 							 &prealloc_cf);
2391 		else
2392 			written = ceph_sync_write(iocb, &data, pos, snapc);
2393 		if (direct_lock)
2394 			ceph_end_io_direct(inode);
2395 		else
2396 			ceph_end_io_write(inode);
2397 		if (written > 0)
2398 			iov_iter_advance(from, written);
2399 		ceph_put_snap_context(snapc);
2400 	} else {
2401 		/*
2402 		 * No need to acquire the i_truncate_mutex. Because
2403 		 * the MDS revokes Fwb caps before sending truncate
2404 		 * message to us. We can't get Fwb cap while there
2405 		 * are pending vmtruncate. So write and vmtruncate
2406 		 * can not run at the same time
2407 		 */
2408 		written = generic_perform_write(iocb, from);
2409 		ceph_end_io_write(inode);
2410 	}
2411 
2412 	if (written >= 0) {
2413 		int dirty;
2414 
2415 		spin_lock(&ci->i_ceph_lock);
2416 		dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
2417 					       &prealloc_cf);
2418 		spin_unlock(&ci->i_ceph_lock);
2419 		if (dirty)
2420 			__mark_inode_dirty(inode, dirty);
2421 		if (ceph_quota_is_max_bytes_approaching(inode, iocb->ki_pos))
2422 			ceph_check_caps(ci, CHECK_CAPS_FLUSH);
2423 	}
2424 
2425 	doutc(cl, "%p %llx.%llx %llu~%u  dropping cap refs on %s\n",
2426 	      inode, ceph_vinop(inode), pos, (unsigned)count,
2427 	      ceph_cap_string(got));
2428 	ceph_put_cap_refs(ci, got);
2429 
2430 	if (written == -EOLDSNAPC) {
2431 		doutc(cl, "%p %llx.%llx %llu~%u" "got EOLDSNAPC, retrying\n",
2432 		      inode, ceph_vinop(inode), pos, (unsigned)count);
2433 		goto retry_snap;
2434 	}
2435 
2436 	if (written >= 0) {
2437 		if ((map_flags & CEPH_OSDMAP_NEARFULL) ||
2438 		    (pool_flags & CEPH_POOL_FLAG_NEARFULL))
2439 			iocb->ki_flags |= IOCB_DSYNC;
2440 		written = generic_write_sync(iocb, written);
2441 	}
2442 
2443 	goto out_unlocked;
2444 out_caps:
2445 	ceph_put_cap_refs(ci, got);
2446 out:
2447 	if (direct_lock)
2448 		ceph_end_io_direct(inode);
2449 	else
2450 		ceph_end_io_write(inode);
2451 out_unlocked:
2452 	ceph_free_cap_flush(prealloc_cf);
2453 	return written ? written : err;
2454 }
2455 
2456 /*
2457  * llseek.  be sure to verify file size on SEEK_END.
2458  */
2459 static loff_t ceph_llseek(struct file *file, loff_t offset, int whence)
2460 {
2461 	if (whence == SEEK_END || whence == SEEK_DATA || whence == SEEK_HOLE) {
2462 		struct inode *inode = file_inode(file);
2463 		int ret;
2464 
2465 		ret = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE, false);
2466 		if (ret < 0)
2467 			return ret;
2468 	}
2469 	return generic_file_llseek(file, offset, whence);
2470 }
2471 
2472 static inline void ceph_zero_partial_page(
2473 	struct inode *inode, loff_t offset, unsigned size)
2474 {
2475 	struct page *page;
2476 	pgoff_t index = offset >> PAGE_SHIFT;
2477 
2478 	page = find_lock_page(inode->i_mapping, index);
2479 	if (page) {
2480 		wait_on_page_writeback(page);
2481 		zero_user(page, offset & (PAGE_SIZE - 1), size);
2482 		unlock_page(page);
2483 		put_page(page);
2484 	}
2485 }
2486 
2487 static void ceph_zero_pagecache_range(struct inode *inode, loff_t offset,
2488 				      loff_t length)
2489 {
2490 	loff_t nearly = round_up(offset, PAGE_SIZE);
2491 	if (offset < nearly) {
2492 		loff_t size = nearly - offset;
2493 		if (length < size)
2494 			size = length;
2495 		ceph_zero_partial_page(inode, offset, size);
2496 		offset += size;
2497 		length -= size;
2498 	}
2499 	if (length >= PAGE_SIZE) {
2500 		loff_t size = round_down(length, PAGE_SIZE);
2501 		truncate_pagecache_range(inode, offset, offset + size - 1);
2502 		offset += size;
2503 		length -= size;
2504 	}
2505 	if (length)
2506 		ceph_zero_partial_page(inode, offset, length);
2507 }
2508 
2509 static int ceph_zero_partial_object(struct inode *inode,
2510 				    loff_t offset, loff_t *length)
2511 {
2512 	struct ceph_inode_info *ci = ceph_inode(inode);
2513 	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
2514 	struct ceph_osd_request *req;
2515 	int ret = 0;
2516 	loff_t zero = 0;
2517 	int op;
2518 
2519 	if (ceph_inode_is_shutdown(inode))
2520 		return -EIO;
2521 
2522 	if (!length) {
2523 		op = offset ? CEPH_OSD_OP_DELETE : CEPH_OSD_OP_TRUNCATE;
2524 		length = &zero;
2525 	} else {
2526 		op = CEPH_OSD_OP_ZERO;
2527 	}
2528 
2529 	req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
2530 					ceph_vino(inode),
2531 					offset, length,
2532 					0, 1, op,
2533 					CEPH_OSD_FLAG_WRITE,
2534 					NULL, 0, 0, false);
2535 	if (IS_ERR(req)) {
2536 		ret = PTR_ERR(req);
2537 		goto out;
2538 	}
2539 
2540 	req->r_mtime = inode_get_mtime(inode);
2541 	ceph_osdc_start_request(&fsc->client->osdc, req);
2542 	ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
2543 	if (ret == -ENOENT)
2544 		ret = 0;
2545 	ceph_osdc_put_request(req);
2546 
2547 out:
2548 	return ret;
2549 }
2550 
2551 static int ceph_zero_objects(struct inode *inode, loff_t offset, loff_t length)
2552 {
2553 	int ret = 0;
2554 	struct ceph_inode_info *ci = ceph_inode(inode);
2555 	s32 stripe_unit = ci->i_layout.stripe_unit;
2556 	s32 stripe_count = ci->i_layout.stripe_count;
2557 	s32 object_size = ci->i_layout.object_size;
2558 	u64 object_set_size = object_size * stripe_count;
2559 	u64 nearly, t;
2560 
2561 	/* round offset up to next period boundary */
2562 	nearly = offset + object_set_size - 1;
2563 	t = nearly;
2564 	nearly -= do_div(t, object_set_size);
2565 
2566 	while (length && offset < nearly) {
2567 		loff_t size = length;
2568 		ret = ceph_zero_partial_object(inode, offset, &size);
2569 		if (ret < 0)
2570 			return ret;
2571 		offset += size;
2572 		length -= size;
2573 	}
2574 	while (length >= object_set_size) {
2575 		int i;
2576 		loff_t pos = offset;
2577 		for (i = 0; i < stripe_count; ++i) {
2578 			ret = ceph_zero_partial_object(inode, pos, NULL);
2579 			if (ret < 0)
2580 				return ret;
2581 			pos += stripe_unit;
2582 		}
2583 		offset += object_set_size;
2584 		length -= object_set_size;
2585 	}
2586 	while (length) {
2587 		loff_t size = length;
2588 		ret = ceph_zero_partial_object(inode, offset, &size);
2589 		if (ret < 0)
2590 			return ret;
2591 		offset += size;
2592 		length -= size;
2593 	}
2594 	return ret;
2595 }
2596 
2597 static long ceph_fallocate(struct file *file, int mode,
2598 				loff_t offset, loff_t length)
2599 {
2600 	struct ceph_file_info *fi = file->private_data;
2601 	struct inode *inode = file_inode(file);
2602 	struct ceph_inode_info *ci = ceph_inode(inode);
2603 	struct ceph_cap_flush *prealloc_cf;
2604 	struct ceph_client *cl = ceph_inode_to_client(inode);
2605 	int want, got = 0;
2606 	int dirty;
2607 	int ret = 0;
2608 	loff_t endoff = 0;
2609 	loff_t size;
2610 
2611 	doutc(cl, "%p %llx.%llx mode %x, offset %llu length %llu\n",
2612 	      inode, ceph_vinop(inode), mode, offset, length);
2613 
2614 	if (mode != (FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE))
2615 		return -EOPNOTSUPP;
2616 
2617 	if (!S_ISREG(inode->i_mode))
2618 		return -EOPNOTSUPP;
2619 
2620 	if (IS_ENCRYPTED(inode))
2621 		return -EOPNOTSUPP;
2622 
2623 	prealloc_cf = ceph_alloc_cap_flush();
2624 	if (!prealloc_cf)
2625 		return -ENOMEM;
2626 
2627 	inode_lock(inode);
2628 
2629 	if (ceph_snap(inode) != CEPH_NOSNAP) {
2630 		ret = -EROFS;
2631 		goto unlock;
2632 	}
2633 
2634 	size = i_size_read(inode);
2635 
2636 	/* Are we punching a hole beyond EOF? */
2637 	if (offset >= size)
2638 		goto unlock;
2639 	if ((offset + length) > size)
2640 		length = size - offset;
2641 
2642 	if (fi->fmode & CEPH_FILE_MODE_LAZY)
2643 		want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
2644 	else
2645 		want = CEPH_CAP_FILE_BUFFER;
2646 
2647 	ret = ceph_get_caps(file, CEPH_CAP_FILE_WR, want, endoff, &got);
2648 	if (ret < 0)
2649 		goto unlock;
2650 
2651 	ret = file_modified(file);
2652 	if (ret)
2653 		goto put_caps;
2654 
2655 	filemap_invalidate_lock(inode->i_mapping);
2656 	ceph_fscache_invalidate(inode, false);
2657 	ceph_zero_pagecache_range(inode, offset, length);
2658 	ret = ceph_zero_objects(inode, offset, length);
2659 
2660 	if (!ret) {
2661 		spin_lock(&ci->i_ceph_lock);
2662 		dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
2663 					       &prealloc_cf);
2664 		spin_unlock(&ci->i_ceph_lock);
2665 		if (dirty)
2666 			__mark_inode_dirty(inode, dirty);
2667 	}
2668 	filemap_invalidate_unlock(inode->i_mapping);
2669 
2670 put_caps:
2671 	ceph_put_cap_refs(ci, got);
2672 unlock:
2673 	inode_unlock(inode);
2674 	ceph_free_cap_flush(prealloc_cf);
2675 	return ret;
2676 }
2677 
2678 /*
2679  * This function tries to get FILE_WR capabilities for dst_ci and FILE_RD for
2680  * src_ci.  Two attempts are made to obtain both caps, and an error is return if
2681  * this fails; zero is returned on success.
2682  */
2683 static int get_rd_wr_caps(struct file *src_filp, int *src_got,
2684 			  struct file *dst_filp,
2685 			  loff_t dst_endoff, int *dst_got)
2686 {
2687 	int ret = 0;
2688 	bool retrying = false;
2689 
2690 retry_caps:
2691 	ret = ceph_get_caps(dst_filp, CEPH_CAP_FILE_WR, CEPH_CAP_FILE_BUFFER,
2692 			    dst_endoff, dst_got);
2693 	if (ret < 0)
2694 		return ret;
2695 
2696 	/*
2697 	 * Since we're already holding the FILE_WR capability for the dst file,
2698 	 * we would risk a deadlock by using ceph_get_caps.  Thus, we'll do some
2699 	 * retry dance instead to try to get both capabilities.
2700 	 */
2701 	ret = ceph_try_get_caps(file_inode(src_filp),
2702 				CEPH_CAP_FILE_RD, CEPH_CAP_FILE_SHARED,
2703 				false, src_got);
2704 	if (ret <= 0) {
2705 		/* Start by dropping dst_ci caps and getting src_ci caps */
2706 		ceph_put_cap_refs(ceph_inode(file_inode(dst_filp)), *dst_got);
2707 		if (retrying) {
2708 			if (!ret)
2709 				/* ceph_try_get_caps masks EAGAIN */
2710 				ret = -EAGAIN;
2711 			return ret;
2712 		}
2713 		ret = ceph_get_caps(src_filp, CEPH_CAP_FILE_RD,
2714 				    CEPH_CAP_FILE_SHARED, -1, src_got);
2715 		if (ret < 0)
2716 			return ret;
2717 		/*... drop src_ci caps too, and retry */
2718 		ceph_put_cap_refs(ceph_inode(file_inode(src_filp)), *src_got);
2719 		retrying = true;
2720 		goto retry_caps;
2721 	}
2722 	return ret;
2723 }
2724 
2725 static void put_rd_wr_caps(struct ceph_inode_info *src_ci, int src_got,
2726 			   struct ceph_inode_info *dst_ci, int dst_got)
2727 {
2728 	ceph_put_cap_refs(src_ci, src_got);
2729 	ceph_put_cap_refs(dst_ci, dst_got);
2730 }
2731 
2732 /*
2733  * This function does several size-related checks, returning an error if:
2734  *  - source file is smaller than off+len
2735  *  - destination file size is not OK (inode_newsize_ok())
2736  *  - max bytes quotas is exceeded
2737  */
2738 static int is_file_size_ok(struct inode *src_inode, struct inode *dst_inode,
2739 			   loff_t src_off, loff_t dst_off, size_t len)
2740 {
2741 	struct ceph_client *cl = ceph_inode_to_client(src_inode);
2742 	loff_t size, endoff;
2743 
2744 	size = i_size_read(src_inode);
2745 	/*
2746 	 * Don't copy beyond source file EOF.  Instead of simply setting length
2747 	 * to (size - src_off), just drop to VFS default implementation, as the
2748 	 * local i_size may be stale due to other clients writing to the source
2749 	 * inode.
2750 	 */
2751 	if (src_off + len > size) {
2752 		doutc(cl, "Copy beyond EOF (%llu + %zu > %llu)\n", src_off,
2753 		      len, size);
2754 		return -EOPNOTSUPP;
2755 	}
2756 	size = i_size_read(dst_inode);
2757 
2758 	endoff = dst_off + len;
2759 	if (inode_newsize_ok(dst_inode, endoff))
2760 		return -EOPNOTSUPP;
2761 
2762 	if (ceph_quota_is_max_bytes_exceeded(dst_inode, endoff))
2763 		return -EDQUOT;
2764 
2765 	return 0;
2766 }
2767 
2768 static struct ceph_osd_request *
2769 ceph_alloc_copyfrom_request(struct ceph_osd_client *osdc,
2770 			    u64 src_snapid,
2771 			    struct ceph_object_id *src_oid,
2772 			    struct ceph_object_locator *src_oloc,
2773 			    struct ceph_object_id *dst_oid,
2774 			    struct ceph_object_locator *dst_oloc,
2775 			    u32 truncate_seq, u64 truncate_size)
2776 {
2777 	struct ceph_osd_request *req;
2778 	int ret;
2779 	u32 src_fadvise_flags =
2780 		CEPH_OSD_OP_FLAG_FADVISE_SEQUENTIAL |
2781 		CEPH_OSD_OP_FLAG_FADVISE_NOCACHE;
2782 	u32 dst_fadvise_flags =
2783 		CEPH_OSD_OP_FLAG_FADVISE_SEQUENTIAL |
2784 		CEPH_OSD_OP_FLAG_FADVISE_DONTNEED;
2785 
2786 	req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_KERNEL);
2787 	if (!req)
2788 		return ERR_PTR(-ENOMEM);
2789 
2790 	req->r_flags = CEPH_OSD_FLAG_WRITE;
2791 
2792 	ceph_oloc_copy(&req->r_t.base_oloc, dst_oloc);
2793 	ceph_oid_copy(&req->r_t.base_oid, dst_oid);
2794 
2795 	ret = osd_req_op_copy_from_init(req, src_snapid, 0,
2796 					src_oid, src_oloc,
2797 					src_fadvise_flags,
2798 					dst_fadvise_flags,
2799 					truncate_seq,
2800 					truncate_size,
2801 					CEPH_OSD_COPY_FROM_FLAG_TRUNCATE_SEQ);
2802 	if (ret)
2803 		goto out;
2804 
2805 	ret = ceph_osdc_alloc_messages(req, GFP_KERNEL);
2806 	if (ret)
2807 		goto out;
2808 
2809 	return req;
2810 
2811 out:
2812 	ceph_osdc_put_request(req);
2813 	return ERR_PTR(ret);
2814 }
2815 
2816 static ssize_t ceph_do_objects_copy(struct ceph_inode_info *src_ci, u64 *src_off,
2817 				    struct ceph_inode_info *dst_ci, u64 *dst_off,
2818 				    struct ceph_fs_client *fsc,
2819 				    size_t len, unsigned int flags)
2820 {
2821 	struct ceph_object_locator src_oloc, dst_oloc;
2822 	struct ceph_object_id src_oid, dst_oid;
2823 	struct ceph_osd_client *osdc;
2824 	struct ceph_osd_request *req;
2825 	size_t bytes = 0;
2826 	u64 src_objnum, src_objoff, dst_objnum, dst_objoff;
2827 	u32 src_objlen, dst_objlen;
2828 	u32 object_size = src_ci->i_layout.object_size;
2829 	struct ceph_client *cl = fsc->client;
2830 	int ret;
2831 
2832 	src_oloc.pool = src_ci->i_layout.pool_id;
2833 	src_oloc.pool_ns = ceph_try_get_string(src_ci->i_layout.pool_ns);
2834 	dst_oloc.pool = dst_ci->i_layout.pool_id;
2835 	dst_oloc.pool_ns = ceph_try_get_string(dst_ci->i_layout.pool_ns);
2836 	osdc = &fsc->client->osdc;
2837 
2838 	while (len >= object_size) {
2839 		ceph_calc_file_object_mapping(&src_ci->i_layout, *src_off,
2840 					      object_size, &src_objnum,
2841 					      &src_objoff, &src_objlen);
2842 		ceph_calc_file_object_mapping(&dst_ci->i_layout, *dst_off,
2843 					      object_size, &dst_objnum,
2844 					      &dst_objoff, &dst_objlen);
2845 		ceph_oid_init(&src_oid);
2846 		ceph_oid_printf(&src_oid, "%llx.%08llx",
2847 				src_ci->i_vino.ino, src_objnum);
2848 		ceph_oid_init(&dst_oid);
2849 		ceph_oid_printf(&dst_oid, "%llx.%08llx",
2850 				dst_ci->i_vino.ino, dst_objnum);
2851 		/* Do an object remote copy */
2852 		req = ceph_alloc_copyfrom_request(osdc, src_ci->i_vino.snap,
2853 						  &src_oid, &src_oloc,
2854 						  &dst_oid, &dst_oloc,
2855 						  dst_ci->i_truncate_seq,
2856 						  dst_ci->i_truncate_size);
2857 		if (IS_ERR(req))
2858 			ret = PTR_ERR(req);
2859 		else {
2860 			ceph_osdc_start_request(osdc, req);
2861 			ret = ceph_osdc_wait_request(osdc, req);
2862 			ceph_update_copyfrom_metrics(&fsc->mdsc->metric,
2863 						     req->r_start_latency,
2864 						     req->r_end_latency,
2865 						     object_size, ret);
2866 			ceph_osdc_put_request(req);
2867 		}
2868 		if (ret) {
2869 			if (ret == -EOPNOTSUPP) {
2870 				fsc->have_copy_from2 = false;
2871 				pr_notice_client(cl,
2872 					"OSDs don't support copy-from2; disabling copy offload\n");
2873 			}
2874 			doutc(cl, "returned %d\n", ret);
2875 			if (!bytes)
2876 				bytes = ret;
2877 			goto out;
2878 		}
2879 		len -= object_size;
2880 		bytes += object_size;
2881 		*src_off += object_size;
2882 		*dst_off += object_size;
2883 	}
2884 
2885 out:
2886 	ceph_oloc_destroy(&src_oloc);
2887 	ceph_oloc_destroy(&dst_oloc);
2888 	return bytes;
2889 }
2890 
2891 static ssize_t __ceph_copy_file_range(struct file *src_file, loff_t src_off,
2892 				      struct file *dst_file, loff_t dst_off,
2893 				      size_t len, unsigned int flags)
2894 {
2895 	struct inode *src_inode = file_inode(src_file);
2896 	struct inode *dst_inode = file_inode(dst_file);
2897 	struct ceph_inode_info *src_ci = ceph_inode(src_inode);
2898 	struct ceph_inode_info *dst_ci = ceph_inode(dst_inode);
2899 	struct ceph_cap_flush *prealloc_cf;
2900 	struct ceph_fs_client *src_fsc = ceph_inode_to_fs_client(src_inode);
2901 	struct ceph_client *cl = src_fsc->client;
2902 	loff_t size;
2903 	ssize_t ret = -EIO, bytes;
2904 	u64 src_objnum, dst_objnum, src_objoff, dst_objoff;
2905 	u32 src_objlen, dst_objlen;
2906 	int src_got = 0, dst_got = 0, err, dirty;
2907 
2908 	if (src_inode->i_sb != dst_inode->i_sb) {
2909 		struct ceph_fs_client *dst_fsc = ceph_inode_to_fs_client(dst_inode);
2910 
2911 		if (ceph_fsid_compare(&src_fsc->client->fsid,
2912 				      &dst_fsc->client->fsid)) {
2913 			dout("Copying files across clusters: src: %pU dst: %pU\n",
2914 			     &src_fsc->client->fsid, &dst_fsc->client->fsid);
2915 			return -EXDEV;
2916 		}
2917 	}
2918 	if (ceph_snap(dst_inode) != CEPH_NOSNAP)
2919 		return -EROFS;
2920 
2921 	/*
2922 	 * Some of the checks below will return -EOPNOTSUPP, which will force a
2923 	 * fallback to the default VFS copy_file_range implementation.  This is
2924 	 * desirable in several cases (for ex, the 'len' is smaller than the
2925 	 * size of the objects, or in cases where that would be more
2926 	 * efficient).
2927 	 */
2928 
2929 	if (ceph_test_mount_opt(src_fsc, NOCOPYFROM))
2930 		return -EOPNOTSUPP;
2931 
2932 	if (!src_fsc->have_copy_from2)
2933 		return -EOPNOTSUPP;
2934 
2935 	/*
2936 	 * Striped file layouts require that we copy partial objects, but the
2937 	 * OSD copy-from operation only supports full-object copies.  Limit
2938 	 * this to non-striped file layouts for now.
2939 	 */
2940 	if ((src_ci->i_layout.stripe_unit != dst_ci->i_layout.stripe_unit) ||
2941 	    (src_ci->i_layout.stripe_count != 1) ||
2942 	    (dst_ci->i_layout.stripe_count != 1) ||
2943 	    (src_ci->i_layout.object_size != dst_ci->i_layout.object_size)) {
2944 		doutc(cl, "Invalid src/dst files layout\n");
2945 		return -EOPNOTSUPP;
2946 	}
2947 
2948 	/* Every encrypted inode gets its own key, so we can't offload them */
2949 	if (IS_ENCRYPTED(src_inode) || IS_ENCRYPTED(dst_inode))
2950 		return -EOPNOTSUPP;
2951 
2952 	if (len < src_ci->i_layout.object_size)
2953 		return -EOPNOTSUPP; /* no remote copy will be done */
2954 
2955 	prealloc_cf = ceph_alloc_cap_flush();
2956 	if (!prealloc_cf)
2957 		return -ENOMEM;
2958 
2959 	/* Start by sync'ing the source and destination files */
2960 	ret = file_write_and_wait_range(src_file, src_off, (src_off + len));
2961 	if (ret < 0) {
2962 		doutc(cl, "failed to write src file (%zd)\n", ret);
2963 		goto out;
2964 	}
2965 	ret = file_write_and_wait_range(dst_file, dst_off, (dst_off + len));
2966 	if (ret < 0) {
2967 		doutc(cl, "failed to write dst file (%zd)\n", ret);
2968 		goto out;
2969 	}
2970 
2971 	/*
2972 	 * We need FILE_WR caps for dst_ci and FILE_RD for src_ci as other
2973 	 * clients may have dirty data in their caches.  And OSDs know nothing
2974 	 * about caps, so they can't safely do the remote object copies.
2975 	 */
2976 	err = get_rd_wr_caps(src_file, &src_got,
2977 			     dst_file, (dst_off + len), &dst_got);
2978 	if (err < 0) {
2979 		doutc(cl, "get_rd_wr_caps returned %d\n", err);
2980 		ret = -EOPNOTSUPP;
2981 		goto out;
2982 	}
2983 
2984 	ret = is_file_size_ok(src_inode, dst_inode, src_off, dst_off, len);
2985 	if (ret < 0)
2986 		goto out_caps;
2987 
2988 	/* Drop dst file cached pages */
2989 	ceph_fscache_invalidate(dst_inode, false);
2990 	ret = invalidate_inode_pages2_range(dst_inode->i_mapping,
2991 					    dst_off >> PAGE_SHIFT,
2992 					    (dst_off + len) >> PAGE_SHIFT);
2993 	if (ret < 0) {
2994 		doutc(cl, "Failed to invalidate inode pages (%zd)\n",
2995 			    ret);
2996 		ret = 0; /* XXX */
2997 	}
2998 	ceph_calc_file_object_mapping(&src_ci->i_layout, src_off,
2999 				      src_ci->i_layout.object_size,
3000 				      &src_objnum, &src_objoff, &src_objlen);
3001 	ceph_calc_file_object_mapping(&dst_ci->i_layout, dst_off,
3002 				      dst_ci->i_layout.object_size,
3003 				      &dst_objnum, &dst_objoff, &dst_objlen);
3004 	/* object-level offsets need to the same */
3005 	if (src_objoff != dst_objoff) {
3006 		ret = -EOPNOTSUPP;
3007 		goto out_caps;
3008 	}
3009 
3010 	/*
3011 	 * Do a manual copy if the object offset isn't object aligned.
3012 	 * 'src_objlen' contains the bytes left until the end of the object,
3013 	 * starting at the src_off
3014 	 */
3015 	if (src_objoff) {
3016 		doutc(cl, "Initial partial copy of %u bytes\n", src_objlen);
3017 
3018 		/*
3019 		 * we need to temporarily drop all caps as we'll be calling
3020 		 * {read,write}_iter, which will get caps again.
3021 		 */
3022 		put_rd_wr_caps(src_ci, src_got, dst_ci, dst_got);
3023 		ret = splice_file_range(src_file, &src_off, dst_file, &dst_off,
3024 					src_objlen);
3025 		/* Abort on short copies or on error */
3026 		if (ret < (long)src_objlen) {
3027 			doutc(cl, "Failed partial copy (%zd)\n", ret);
3028 			goto out;
3029 		}
3030 		len -= ret;
3031 		err = get_rd_wr_caps(src_file, &src_got,
3032 				     dst_file, (dst_off + len), &dst_got);
3033 		if (err < 0)
3034 			goto out;
3035 		err = is_file_size_ok(src_inode, dst_inode,
3036 				      src_off, dst_off, len);
3037 		if (err < 0)
3038 			goto out_caps;
3039 	}
3040 
3041 	size = i_size_read(dst_inode);
3042 	bytes = ceph_do_objects_copy(src_ci, &src_off, dst_ci, &dst_off,
3043 				     src_fsc, len, flags);
3044 	if (bytes <= 0) {
3045 		if (!ret)
3046 			ret = bytes;
3047 		goto out_caps;
3048 	}
3049 	doutc(cl, "Copied %zu bytes out of %zu\n", bytes, len);
3050 	len -= bytes;
3051 	ret += bytes;
3052 
3053 	file_update_time(dst_file);
3054 	inode_inc_iversion_raw(dst_inode);
3055 
3056 	if (dst_off > size) {
3057 		/* Let the MDS know about dst file size change */
3058 		if (ceph_inode_set_size(dst_inode, dst_off) ||
3059 		    ceph_quota_is_max_bytes_approaching(dst_inode, dst_off))
3060 			ceph_check_caps(dst_ci, CHECK_CAPS_AUTHONLY | CHECK_CAPS_FLUSH);
3061 	}
3062 	/* Mark Fw dirty */
3063 	spin_lock(&dst_ci->i_ceph_lock);
3064 	dirty = __ceph_mark_dirty_caps(dst_ci, CEPH_CAP_FILE_WR, &prealloc_cf);
3065 	spin_unlock(&dst_ci->i_ceph_lock);
3066 	if (dirty)
3067 		__mark_inode_dirty(dst_inode, dirty);
3068 
3069 out_caps:
3070 	put_rd_wr_caps(src_ci, src_got, dst_ci, dst_got);
3071 
3072 	/*
3073 	 * Do the final manual copy if we still have some bytes left, unless
3074 	 * there were errors in remote object copies (len >= object_size).
3075 	 */
3076 	if (len && (len < src_ci->i_layout.object_size)) {
3077 		doutc(cl, "Final partial copy of %zu bytes\n", len);
3078 		bytes = splice_file_range(src_file, &src_off, dst_file,
3079 					  &dst_off, len);
3080 		if (bytes > 0)
3081 			ret += bytes;
3082 		else
3083 			doutc(cl, "Failed partial copy (%zd)\n", bytes);
3084 	}
3085 
3086 out:
3087 	ceph_free_cap_flush(prealloc_cf);
3088 
3089 	return ret;
3090 }
3091 
3092 static ssize_t ceph_copy_file_range(struct file *src_file, loff_t src_off,
3093 				    struct file *dst_file, loff_t dst_off,
3094 				    size_t len, unsigned int flags)
3095 {
3096 	ssize_t ret;
3097 
3098 	ret = __ceph_copy_file_range(src_file, src_off, dst_file, dst_off,
3099 				     len, flags);
3100 
3101 	if (ret == -EOPNOTSUPP || ret == -EXDEV)
3102 		ret = splice_copy_file_range(src_file, src_off, dst_file,
3103 					     dst_off, len);
3104 	return ret;
3105 }
3106 
3107 const struct file_operations ceph_file_fops = {
3108 	.open = ceph_open,
3109 	.release = ceph_release,
3110 	.llseek = ceph_llseek,
3111 	.read_iter = ceph_read_iter,
3112 	.write_iter = ceph_write_iter,
3113 	.mmap = ceph_mmap,
3114 	.fsync = ceph_fsync,
3115 	.lock = ceph_lock,
3116 	.setlease = simple_nosetlease,
3117 	.flock = ceph_flock,
3118 	.splice_read = ceph_splice_read,
3119 	.splice_write = iter_file_splice_write,
3120 	.unlocked_ioctl = ceph_ioctl,
3121 	.compat_ioctl = compat_ptr_ioctl,
3122 	.fallocate	= ceph_fallocate,
3123 	.copy_file_range = ceph_copy_file_range,
3124 };
3125