xref: /openbmc/linux/fs/ceph/file.c (revision a13f2ef1)
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 
16 #include "super.h"
17 #include "mds_client.h"
18 #include "cache.h"
19 #include "io.h"
20 #include "metric.h"
21 
22 static __le32 ceph_flags_sys2wire(u32 flags)
23 {
24 	u32 wire_flags = 0;
25 
26 	switch (flags & O_ACCMODE) {
27 	case O_RDONLY:
28 		wire_flags |= CEPH_O_RDONLY;
29 		break;
30 	case O_WRONLY:
31 		wire_flags |= CEPH_O_WRONLY;
32 		break;
33 	case O_RDWR:
34 		wire_flags |= CEPH_O_RDWR;
35 		break;
36 	}
37 
38 	flags &= ~O_ACCMODE;
39 
40 #define ceph_sys2wire(a) if (flags & a) { wire_flags |= CEPH_##a; flags &= ~a; }
41 
42 	ceph_sys2wire(O_CREAT);
43 	ceph_sys2wire(O_EXCL);
44 	ceph_sys2wire(O_TRUNC);
45 	ceph_sys2wire(O_DIRECTORY);
46 	ceph_sys2wire(O_NOFOLLOW);
47 
48 #undef ceph_sys2wire
49 
50 	if (flags)
51 		dout("unused open flags: %x\n", flags);
52 
53 	return cpu_to_le32(wire_flags);
54 }
55 
56 /*
57  * Ceph file operations
58  *
59  * Implement basic open/close functionality, and implement
60  * read/write.
61  *
62  * We implement three modes of file I/O:
63  *  - buffered uses the generic_file_aio_{read,write} helpers
64  *
65  *  - synchronous is used when there is multi-client read/write
66  *    sharing, avoids the page cache, and synchronously waits for an
67  *    ack from the OSD.
68  *
69  *  - direct io takes the variant of the sync path that references
70  *    user pages directly.
71  *
72  * fsync() flushes and waits on dirty pages, but just queues metadata
73  * for writeback: since the MDS can recover size and mtime there is no
74  * need to wait for MDS acknowledgement.
75  */
76 
77 /*
78  * How many pages to get in one call to iov_iter_get_pages().  This
79  * determines the size of the on-stack array used as a buffer.
80  */
81 #define ITER_GET_BVECS_PAGES	64
82 
83 static ssize_t __iter_get_bvecs(struct iov_iter *iter, size_t maxsize,
84 				struct bio_vec *bvecs)
85 {
86 	size_t size = 0;
87 	int bvec_idx = 0;
88 
89 	if (maxsize > iov_iter_count(iter))
90 		maxsize = iov_iter_count(iter);
91 
92 	while (size < maxsize) {
93 		struct page *pages[ITER_GET_BVECS_PAGES];
94 		ssize_t bytes;
95 		size_t start;
96 		int idx = 0;
97 
98 		bytes = iov_iter_get_pages(iter, pages, maxsize - size,
99 					   ITER_GET_BVECS_PAGES, &start);
100 		if (bytes < 0)
101 			return size ?: bytes;
102 
103 		iov_iter_advance(iter, bytes);
104 		size += bytes;
105 
106 		for ( ; bytes; idx++, bvec_idx++) {
107 			struct bio_vec bv = {
108 				.bv_page = pages[idx],
109 				.bv_len = min_t(int, bytes, PAGE_SIZE - start),
110 				.bv_offset = start,
111 			};
112 
113 			bvecs[bvec_idx] = bv;
114 			bytes -= bv.bv_len;
115 			start = 0;
116 		}
117 	}
118 
119 	return size;
120 }
121 
122 /*
123  * iov_iter_get_pages() only considers one iov_iter segment, no matter
124  * what maxsize or maxpages are given.  For ITER_BVEC that is a single
125  * page.
126  *
127  * Attempt to get up to @maxsize bytes worth of pages from @iter.
128  * Return the number of bytes in the created bio_vec array, or an error.
129  */
130 static ssize_t iter_get_bvecs_alloc(struct iov_iter *iter, size_t maxsize,
131 				    struct bio_vec **bvecs, int *num_bvecs)
132 {
133 	struct bio_vec *bv;
134 	size_t orig_count = iov_iter_count(iter);
135 	ssize_t bytes;
136 	int npages;
137 
138 	iov_iter_truncate(iter, maxsize);
139 	npages = iov_iter_npages(iter, INT_MAX);
140 	iov_iter_reexpand(iter, orig_count);
141 
142 	/*
143 	 * __iter_get_bvecs() may populate only part of the array -- zero it
144 	 * out.
145 	 */
146 	bv = kvmalloc_array(npages, sizeof(*bv), GFP_KERNEL | __GFP_ZERO);
147 	if (!bv)
148 		return -ENOMEM;
149 
150 	bytes = __iter_get_bvecs(iter, maxsize, bv);
151 	if (bytes < 0) {
152 		/*
153 		 * No pages were pinned -- just free the array.
154 		 */
155 		kvfree(bv);
156 		return bytes;
157 	}
158 
159 	*bvecs = bv;
160 	*num_bvecs = npages;
161 	return bytes;
162 }
163 
164 static void put_bvecs(struct bio_vec *bvecs, int num_bvecs, bool should_dirty)
165 {
166 	int i;
167 
168 	for (i = 0; i < num_bvecs; i++) {
169 		if (bvecs[i].bv_page) {
170 			if (should_dirty)
171 				set_page_dirty_lock(bvecs[i].bv_page);
172 			put_page(bvecs[i].bv_page);
173 		}
174 	}
175 	kvfree(bvecs);
176 }
177 
178 /*
179  * Prepare an open request.  Preallocate ceph_cap to avoid an
180  * inopportune ENOMEM later.
181  */
182 static struct ceph_mds_request *
183 prepare_open_request(struct super_block *sb, int flags, int create_mode)
184 {
185 	struct ceph_fs_client *fsc = ceph_sb_to_client(sb);
186 	struct ceph_mds_client *mdsc = fsc->mdsc;
187 	struct ceph_mds_request *req;
188 	int want_auth = USE_ANY_MDS;
189 	int op = (flags & O_CREAT) ? CEPH_MDS_OP_CREATE : CEPH_MDS_OP_OPEN;
190 
191 	if (flags & (O_WRONLY|O_RDWR|O_CREAT|O_TRUNC))
192 		want_auth = USE_AUTH_MDS;
193 
194 	req = ceph_mdsc_create_request(mdsc, op, want_auth);
195 	if (IS_ERR(req))
196 		goto out;
197 	req->r_fmode = ceph_flags_to_mode(flags);
198 	req->r_args.open.flags = ceph_flags_sys2wire(flags);
199 	req->r_args.open.mode = cpu_to_le32(create_mode);
200 out:
201 	return req;
202 }
203 
204 static int ceph_init_file_info(struct inode *inode, struct file *file,
205 					int fmode, bool isdir)
206 {
207 	struct ceph_inode_info *ci = ceph_inode(inode);
208 	struct ceph_file_info *fi;
209 
210 	dout("%s %p %p 0%o (%s)\n", __func__, inode, file,
211 			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 		file->private_data = fi;
230 	}
231 
232 	ceph_get_fmode(ci, fmode, 1);
233 	fi->fmode = fmode;
234 
235 	spin_lock_init(&fi->rw_contexts_lock);
236 	INIT_LIST_HEAD(&fi->rw_contexts);
237 	fi->meta_err = errseq_sample(&ci->i_meta_err);
238 	fi->filp_gen = READ_ONCE(ceph_inode_to_client(inode)->filp_gen);
239 
240 	return 0;
241 }
242 
243 /*
244  * initialize private struct file data.
245  * if we fail, clean up by dropping fmode reference on the ceph_inode
246  */
247 static int ceph_init_file(struct inode *inode, struct file *file, int fmode)
248 {
249 	int ret = 0;
250 
251 	switch (inode->i_mode & S_IFMT) {
252 	case S_IFREG:
253 		ceph_fscache_register_inode_cookie(inode);
254 		ceph_fscache_file_set_cookie(inode, file);
255 		/* fall through */
256 	case S_IFDIR:
257 		ret = ceph_init_file_info(inode, file, fmode,
258 						S_ISDIR(inode->i_mode));
259 		if (ret)
260 			return ret;
261 		break;
262 
263 	case S_IFLNK:
264 		dout("init_file %p %p 0%o (symlink)\n", inode, file,
265 		     inode->i_mode);
266 		break;
267 
268 	default:
269 		dout("init_file %p %p 0%o (special)\n", inode, file,
270 		     inode->i_mode);
271 		/*
272 		 * we need to drop the open ref now, since we don't
273 		 * have .release set to ceph_release.
274 		 */
275 		BUG_ON(inode->i_fop->release == ceph_release);
276 
277 		/* call the proper open fop */
278 		ret = inode->i_fop->open(inode, file);
279 	}
280 	return ret;
281 }
282 
283 /*
284  * try renew caps after session gets killed.
285  */
286 int ceph_renew_caps(struct inode *inode, int fmode)
287 {
288 	struct ceph_mds_client *mdsc = ceph_inode_to_client(inode)->mdsc;
289 	struct ceph_inode_info *ci = ceph_inode(inode);
290 	struct ceph_mds_request *req;
291 	int err, flags, wanted;
292 
293 	spin_lock(&ci->i_ceph_lock);
294 	__ceph_touch_fmode(ci, mdsc, fmode);
295 	wanted = __ceph_caps_file_wanted(ci);
296 	if (__ceph_is_any_real_caps(ci) &&
297 	    (!(wanted & CEPH_CAP_ANY_WR) || ci->i_auth_cap)) {
298 		int issued = __ceph_caps_issued(ci, NULL);
299 		spin_unlock(&ci->i_ceph_lock);
300 		dout("renew caps %p want %s issued %s updating mds_wanted\n",
301 		     inode, ceph_cap_string(wanted), ceph_cap_string(issued));
302 		ceph_check_caps(ci, 0, NULL);
303 		return 0;
304 	}
305 	spin_unlock(&ci->i_ceph_lock);
306 
307 	flags = 0;
308 	if ((wanted & CEPH_CAP_FILE_RD) && (wanted & CEPH_CAP_FILE_WR))
309 		flags = O_RDWR;
310 	else if (wanted & CEPH_CAP_FILE_RD)
311 		flags = O_RDONLY;
312 	else if (wanted & CEPH_CAP_FILE_WR)
313 		flags = O_WRONLY;
314 #ifdef O_LAZY
315 	if (wanted & CEPH_CAP_FILE_LAZYIO)
316 		flags |= O_LAZY;
317 #endif
318 
319 	req = prepare_open_request(inode->i_sb, flags, 0);
320 	if (IS_ERR(req)) {
321 		err = PTR_ERR(req);
322 		goto out;
323 	}
324 
325 	req->r_inode = inode;
326 	ihold(inode);
327 	req->r_num_caps = 1;
328 
329 	err = ceph_mdsc_do_request(mdsc, NULL, req);
330 	ceph_mdsc_put_request(req);
331 out:
332 	dout("renew caps %p open result=%d\n", inode, err);
333 	return err < 0 ? err : 0;
334 }
335 
336 /*
337  * If we already have the requisite capabilities, we can satisfy
338  * the open request locally (no need to request new caps from the
339  * MDS).  We do, however, need to inform the MDS (asynchronously)
340  * if our wanted caps set expands.
341  */
342 int ceph_open(struct inode *inode, struct file *file)
343 {
344 	struct ceph_inode_info *ci = ceph_inode(inode);
345 	struct ceph_fs_client *fsc = ceph_sb_to_client(inode->i_sb);
346 	struct ceph_mds_client *mdsc = fsc->mdsc;
347 	struct ceph_mds_request *req;
348 	struct ceph_file_info *fi = file->private_data;
349 	int err;
350 	int flags, fmode, wanted;
351 
352 	if (fi) {
353 		dout("open file %p is already opened\n", file);
354 		return 0;
355 	}
356 
357 	/* filter out O_CREAT|O_EXCL; vfs did that already.  yuck. */
358 	flags = file->f_flags & ~(O_CREAT|O_EXCL);
359 	if (S_ISDIR(inode->i_mode))
360 		flags = O_DIRECTORY;  /* mds likes to know */
361 
362 	dout("open inode %p ino %llx.%llx file %p flags %d (%d)\n", inode,
363 	     ceph_vinop(inode), file, flags, file->f_flags);
364 	fmode = ceph_flags_to_mode(flags);
365 	wanted = ceph_caps_for_mode(fmode);
366 
367 	/* snapped files are read-only */
368 	if (ceph_snap(inode) != CEPH_NOSNAP && (file->f_mode & FMODE_WRITE))
369 		return -EROFS;
370 
371 	/* trivially open snapdir */
372 	if (ceph_snap(inode) == CEPH_SNAPDIR) {
373 		return ceph_init_file(inode, file, fmode);
374 	}
375 
376 	/*
377 	 * No need to block if we have caps on the auth MDS (for
378 	 * write) or any MDS (for read).  Update wanted set
379 	 * asynchronously.
380 	 */
381 	spin_lock(&ci->i_ceph_lock);
382 	if (__ceph_is_any_real_caps(ci) &&
383 	    (((fmode & CEPH_FILE_MODE_WR) == 0) || ci->i_auth_cap)) {
384 		int mds_wanted = __ceph_caps_mds_wanted(ci, true);
385 		int issued = __ceph_caps_issued(ci, NULL);
386 
387 		dout("open %p fmode %d want %s issued %s using existing\n",
388 		     inode, fmode, ceph_cap_string(wanted),
389 		     ceph_cap_string(issued));
390 		__ceph_touch_fmode(ci, mdsc, fmode);
391 		spin_unlock(&ci->i_ceph_lock);
392 
393 		/* adjust wanted? */
394 		if ((issued & wanted) != wanted &&
395 		    (mds_wanted & wanted) != wanted &&
396 		    ceph_snap(inode) != CEPH_SNAPDIR)
397 			ceph_check_caps(ci, 0, NULL);
398 
399 		return ceph_init_file(inode, file, fmode);
400 	} else if (ceph_snap(inode) != CEPH_NOSNAP &&
401 		   (ci->i_snap_caps & wanted) == wanted) {
402 		__ceph_touch_fmode(ci, mdsc, fmode);
403 		spin_unlock(&ci->i_ceph_lock);
404 		return ceph_init_file(inode, file, fmode);
405 	}
406 
407 	spin_unlock(&ci->i_ceph_lock);
408 
409 	dout("open fmode %d wants %s\n", fmode, ceph_cap_string(wanted));
410 	req = prepare_open_request(inode->i_sb, flags, 0);
411 	if (IS_ERR(req)) {
412 		err = PTR_ERR(req);
413 		goto out;
414 	}
415 	req->r_inode = inode;
416 	ihold(inode);
417 
418 	req->r_num_caps = 1;
419 	err = ceph_mdsc_do_request(mdsc, NULL, req);
420 	if (!err)
421 		err = ceph_init_file(inode, file, req->r_fmode);
422 	ceph_mdsc_put_request(req);
423 	dout("open result=%d on %llx.%llx\n", err, ceph_vinop(inode));
424 out:
425 	return err;
426 }
427 
428 /* Clone the layout from a synchronous create, if the dir now has Dc caps */
429 static void
430 cache_file_layout(struct inode *dst, struct inode *src)
431 {
432 	struct ceph_inode_info *cdst = ceph_inode(dst);
433 	struct ceph_inode_info *csrc = ceph_inode(src);
434 
435 	spin_lock(&cdst->i_ceph_lock);
436 	if ((__ceph_caps_issued(cdst, NULL) & CEPH_CAP_DIR_CREATE) &&
437 	    !ceph_file_layout_is_valid(&cdst->i_cached_layout)) {
438 		memcpy(&cdst->i_cached_layout, &csrc->i_layout,
439 			sizeof(cdst->i_cached_layout));
440 		rcu_assign_pointer(cdst->i_cached_layout.pool_ns,
441 				   ceph_try_get_string(csrc->i_layout.pool_ns));
442 	}
443 	spin_unlock(&cdst->i_ceph_lock);
444 }
445 
446 /*
447  * Try to set up an async create. We need caps, a file layout, and inode number,
448  * and either a lease on the dentry or complete dir info. If any of those
449  * criteria are not satisfied, then return false and the caller can go
450  * synchronous.
451  */
452 static int try_prep_async_create(struct inode *dir, struct dentry *dentry,
453 				 struct ceph_file_layout *lo, u64 *pino)
454 {
455 	struct ceph_inode_info *ci = ceph_inode(dir);
456 	struct ceph_dentry_info *di = ceph_dentry(dentry);
457 	int got = 0, want = CEPH_CAP_FILE_EXCL | CEPH_CAP_DIR_CREATE;
458 	u64 ino;
459 
460 	spin_lock(&ci->i_ceph_lock);
461 	/* No auth cap means no chance for Dc caps */
462 	if (!ci->i_auth_cap)
463 		goto no_async;
464 
465 	/* Any delegated inos? */
466 	if (xa_empty(&ci->i_auth_cap->session->s_delegated_inos))
467 		goto no_async;
468 
469 	if (!ceph_file_layout_is_valid(&ci->i_cached_layout))
470 		goto no_async;
471 
472 	if ((__ceph_caps_issued(ci, NULL) & want) != want)
473 		goto no_async;
474 
475 	if (d_in_lookup(dentry)) {
476 		if (!__ceph_dir_is_complete(ci))
477 			goto no_async;
478 		spin_lock(&dentry->d_lock);
479 		di->lease_shared_gen = atomic_read(&ci->i_shared_gen);
480 		spin_unlock(&dentry->d_lock);
481 	} else if (atomic_read(&ci->i_shared_gen) !=
482 		   READ_ONCE(di->lease_shared_gen)) {
483 		goto no_async;
484 	}
485 
486 	ino = ceph_get_deleg_ino(ci->i_auth_cap->session);
487 	if (!ino)
488 		goto no_async;
489 
490 	*pino = ino;
491 	ceph_take_cap_refs(ci, want, false);
492 	memcpy(lo, &ci->i_cached_layout, sizeof(*lo));
493 	rcu_assign_pointer(lo->pool_ns,
494 			   ceph_try_get_string(ci->i_cached_layout.pool_ns));
495 	got = want;
496 no_async:
497 	spin_unlock(&ci->i_ceph_lock);
498 	return got;
499 }
500 
501 static void restore_deleg_ino(struct inode *dir, u64 ino)
502 {
503 	struct ceph_inode_info *ci = ceph_inode(dir);
504 	struct ceph_mds_session *s = NULL;
505 
506 	spin_lock(&ci->i_ceph_lock);
507 	if (ci->i_auth_cap)
508 		s = ceph_get_mds_session(ci->i_auth_cap->session);
509 	spin_unlock(&ci->i_ceph_lock);
510 	if (s) {
511 		int err = ceph_restore_deleg_ino(s, ino);
512 		if (err)
513 			pr_warn("ceph: unable to restore delegated ino 0x%llx to session: %d\n",
514 				ino, err);
515 		ceph_put_mds_session(s);
516 	}
517 }
518 
519 static void ceph_async_create_cb(struct ceph_mds_client *mdsc,
520                                  struct ceph_mds_request *req)
521 {
522 	int result = req->r_err ? req->r_err :
523 			le32_to_cpu(req->r_reply_info.head->result);
524 
525 	if (result == -EJUKEBOX)
526 		goto out;
527 
528 	mapping_set_error(req->r_parent->i_mapping, result);
529 
530 	if (result) {
531 		struct dentry *dentry = req->r_dentry;
532 		int pathlen = 0;
533 		u64 base = 0;
534 		char *path = ceph_mdsc_build_path(req->r_dentry, &pathlen,
535 						  &base, 0);
536 
537 		ceph_dir_clear_complete(req->r_parent);
538 		if (!d_unhashed(dentry))
539 			d_drop(dentry);
540 
541 		/* FIXME: start returning I/O errors on all accesses? */
542 		pr_warn("ceph: async create failure path=(%llx)%s result=%d!\n",
543 			base, IS_ERR(path) ? "<<bad>>" : path, result);
544 		ceph_mdsc_free_path(path, pathlen);
545 	}
546 
547 	if (req->r_target_inode) {
548 		struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
549 		u64 ino = ceph_vino(req->r_target_inode).ino;
550 
551 		if (req->r_deleg_ino != ino)
552 			pr_warn("%s: inode number mismatch! err=%d deleg_ino=0x%llx target=0x%llx\n",
553 				__func__, req->r_err, req->r_deleg_ino, ino);
554 		mapping_set_error(req->r_target_inode->i_mapping, result);
555 
556 		spin_lock(&ci->i_ceph_lock);
557 		if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE) {
558 			ci->i_ceph_flags &= ~CEPH_I_ASYNC_CREATE;
559 			wake_up_bit(&ci->i_ceph_flags, CEPH_ASYNC_CREATE_BIT);
560 		}
561 		ceph_kick_flushing_inode_caps(req->r_session, ci);
562 		spin_unlock(&ci->i_ceph_lock);
563 	} else {
564 		pr_warn("%s: no req->r_target_inode for 0x%llx\n", __func__,
565 			req->r_deleg_ino);
566 	}
567 out:
568 	ceph_mdsc_release_dir_caps(req);
569 }
570 
571 static int ceph_finish_async_create(struct inode *dir, struct dentry *dentry,
572 				    struct file *file, umode_t mode,
573 				    struct ceph_mds_request *req,
574 				    struct ceph_acl_sec_ctx *as_ctx,
575 				    struct ceph_file_layout *lo)
576 {
577 	int ret;
578 	char xattr_buf[4];
579 	struct ceph_mds_reply_inode in = { };
580 	struct ceph_mds_reply_info_in iinfo = { .in = &in };
581 	struct ceph_inode_info *ci = ceph_inode(dir);
582 	struct inode *inode;
583 	struct timespec64 now;
584 	struct ceph_vino vino = { .ino = req->r_deleg_ino,
585 				  .snap = CEPH_NOSNAP };
586 
587 	ktime_get_real_ts64(&now);
588 
589 	inode = ceph_get_inode(dentry->d_sb, vino);
590 	if (IS_ERR(inode))
591 		return PTR_ERR(inode);
592 
593 	iinfo.inline_version = CEPH_INLINE_NONE;
594 	iinfo.change_attr = 1;
595 	ceph_encode_timespec64(&iinfo.btime, &now);
596 
597 	iinfo.xattr_len = ARRAY_SIZE(xattr_buf);
598 	iinfo.xattr_data = xattr_buf;
599 	memset(iinfo.xattr_data, 0, iinfo.xattr_len);
600 
601 	in.ino = cpu_to_le64(vino.ino);
602 	in.snapid = cpu_to_le64(CEPH_NOSNAP);
603 	in.version = cpu_to_le64(1);	// ???
604 	in.cap.caps = in.cap.wanted = cpu_to_le32(CEPH_CAP_ALL_FILE);
605 	in.cap.cap_id = cpu_to_le64(1);
606 	in.cap.realm = cpu_to_le64(ci->i_snap_realm->ino);
607 	in.cap.flags = CEPH_CAP_FLAG_AUTH;
608 	in.ctime = in.mtime = in.atime = iinfo.btime;
609 	in.mode = cpu_to_le32((u32)mode);
610 	in.truncate_seq = cpu_to_le32(1);
611 	in.truncate_size = cpu_to_le64(-1ULL);
612 	in.xattr_version = cpu_to_le64(1);
613 	in.uid = cpu_to_le32(from_kuid(&init_user_ns, current_fsuid()));
614 	in.gid = cpu_to_le32(from_kgid(&init_user_ns, dir->i_mode & S_ISGID ?
615 				dir->i_gid : current_fsgid()));
616 	in.nlink = cpu_to_le32(1);
617 	in.max_size = cpu_to_le64(lo->stripe_unit);
618 
619 	ceph_file_layout_to_legacy(lo, &in.layout);
620 
621 	ret = ceph_fill_inode(inode, NULL, &iinfo, NULL, req->r_session,
622 			      req->r_fmode, NULL);
623 	if (ret) {
624 		dout("%s failed to fill inode: %d\n", __func__, ret);
625 		ceph_dir_clear_complete(dir);
626 		if (!d_unhashed(dentry))
627 			d_drop(dentry);
628 		if (inode->i_state & I_NEW)
629 			discard_new_inode(inode);
630 	} else {
631 		struct dentry *dn;
632 
633 		dout("%s d_adding new inode 0x%llx to 0x%lx/%s\n", __func__,
634 			vino.ino, dir->i_ino, dentry->d_name.name);
635 		ceph_dir_clear_ordered(dir);
636 		ceph_init_inode_acls(inode, as_ctx);
637 		if (inode->i_state & I_NEW) {
638 			/*
639 			 * If it's not I_NEW, then someone created this before
640 			 * we got here. Assume the server is aware of it at
641 			 * that point and don't worry about setting
642 			 * CEPH_I_ASYNC_CREATE.
643 			 */
644 			ceph_inode(inode)->i_ceph_flags = CEPH_I_ASYNC_CREATE;
645 			unlock_new_inode(inode);
646 		}
647 		if (d_in_lookup(dentry) || d_really_is_negative(dentry)) {
648 			if (!d_unhashed(dentry))
649 				d_drop(dentry);
650 			dn = d_splice_alias(inode, dentry);
651 			WARN_ON_ONCE(dn && dn != dentry);
652 		}
653 		file->f_mode |= FMODE_CREATED;
654 		ret = finish_open(file, dentry, ceph_open);
655 	}
656 	return ret;
657 }
658 
659 /*
660  * Do a lookup + open with a single request.  If we get a non-existent
661  * file or symlink, return 1 so the VFS can retry.
662  */
663 int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
664 		     struct file *file, unsigned flags, umode_t mode)
665 {
666 	struct ceph_fs_client *fsc = ceph_sb_to_client(dir->i_sb);
667 	struct ceph_mds_client *mdsc = fsc->mdsc;
668 	struct ceph_mds_request *req;
669 	struct dentry *dn;
670 	struct ceph_acl_sec_ctx as_ctx = {};
671 	bool try_async = ceph_test_mount_opt(fsc, ASYNC_DIROPS);
672 	int mask;
673 	int err;
674 
675 	dout("atomic_open %p dentry %p '%pd' %s flags %d mode 0%o\n",
676 	     dir, dentry, dentry,
677 	     d_unhashed(dentry) ? "unhashed" : "hashed", flags, mode);
678 
679 	if (dentry->d_name.len > NAME_MAX)
680 		return -ENAMETOOLONG;
681 
682 	if (flags & O_CREAT) {
683 		if (ceph_quota_is_max_files_exceeded(dir))
684 			return -EDQUOT;
685 		err = ceph_pre_init_acls(dir, &mode, &as_ctx);
686 		if (err < 0)
687 			return err;
688 		err = ceph_security_init_secctx(dentry, mode, &as_ctx);
689 		if (err < 0)
690 			goto out_ctx;
691 	} else if (!d_in_lookup(dentry)) {
692 		/* If it's not being looked up, it's negative */
693 		return -ENOENT;
694 	}
695 retry:
696 	/* do the open */
697 	req = prepare_open_request(dir->i_sb, flags, mode);
698 	if (IS_ERR(req)) {
699 		err = PTR_ERR(req);
700 		goto out_ctx;
701 	}
702 	req->r_dentry = dget(dentry);
703 	req->r_num_caps = 2;
704 	mask = CEPH_STAT_CAP_INODE | CEPH_CAP_AUTH_SHARED;
705 	if (ceph_security_xattr_wanted(dir))
706 		mask |= CEPH_CAP_XATTR_SHARED;
707 	req->r_args.open.mask = cpu_to_le32(mask);
708 	req->r_parent = dir;
709 
710 	if (flags & O_CREAT) {
711 		struct ceph_file_layout lo;
712 
713 		req->r_dentry_drop = CEPH_CAP_FILE_SHARED | CEPH_CAP_AUTH_EXCL;
714 		req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
715 		if (as_ctx.pagelist) {
716 			req->r_pagelist = as_ctx.pagelist;
717 			as_ctx.pagelist = NULL;
718 		}
719 		if (try_async &&
720 		    (req->r_dir_caps =
721 		      try_prep_async_create(dir, dentry, &lo,
722 					    &req->r_deleg_ino))) {
723 			set_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags);
724 			req->r_args.open.flags |= cpu_to_le32(CEPH_O_EXCL);
725 			req->r_callback = ceph_async_create_cb;
726 			err = ceph_mdsc_submit_request(mdsc, dir, req);
727 			if (!err) {
728 				err = ceph_finish_async_create(dir, dentry,
729 							file, mode, req,
730 							&as_ctx, &lo);
731 			} else if (err == -EJUKEBOX) {
732 				restore_deleg_ino(dir, req->r_deleg_ino);
733 				ceph_mdsc_put_request(req);
734 				try_async = false;
735 				goto retry;
736 			}
737 			goto out_req;
738 		}
739 	}
740 
741 	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
742 	err = ceph_mdsc_do_request(mdsc,
743 				   (flags & (O_CREAT|O_TRUNC)) ? dir : NULL,
744 				   req);
745 	err = ceph_handle_snapdir(req, dentry, err);
746 	if (err)
747 		goto out_req;
748 
749 	if ((flags & O_CREAT) && !req->r_reply_info.head->is_dentry)
750 		err = ceph_handle_notrace_create(dir, dentry);
751 
752 	if (d_in_lookup(dentry)) {
753 		dn = ceph_finish_lookup(req, dentry, err);
754 		if (IS_ERR(dn))
755 			err = PTR_ERR(dn);
756 	} else {
757 		/* we were given a hashed negative dentry */
758 		dn = NULL;
759 	}
760 	if (err)
761 		goto out_req;
762 	if (dn || d_really_is_negative(dentry) || d_is_symlink(dentry)) {
763 		/* make vfs retry on splice, ENOENT, or symlink */
764 		dout("atomic_open finish_no_open on dn %p\n", dn);
765 		err = finish_no_open(file, dn);
766 	} else {
767 		dout("atomic_open finish_open on dn %p\n", dn);
768 		if (req->r_op == CEPH_MDS_OP_CREATE && req->r_reply_info.has_create_ino) {
769 			struct inode *newino = d_inode(dentry);
770 
771 			cache_file_layout(dir, newino);
772 			ceph_init_inode_acls(newino, &as_ctx);
773 			file->f_mode |= FMODE_CREATED;
774 		}
775 		err = finish_open(file, dentry, ceph_open);
776 	}
777 out_req:
778 	ceph_mdsc_put_request(req);
779 out_ctx:
780 	ceph_release_acl_sec_ctx(&as_ctx);
781 	dout("atomic_open result=%d\n", err);
782 	return err;
783 }
784 
785 int ceph_release(struct inode *inode, struct file *file)
786 {
787 	struct ceph_inode_info *ci = ceph_inode(inode);
788 
789 	if (S_ISDIR(inode->i_mode)) {
790 		struct ceph_dir_file_info *dfi = file->private_data;
791 		dout("release inode %p dir file %p\n", inode, file);
792 		WARN_ON(!list_empty(&dfi->file_info.rw_contexts));
793 
794 		ceph_put_fmode(ci, dfi->file_info.fmode, 1);
795 
796 		if (dfi->last_readdir)
797 			ceph_mdsc_put_request(dfi->last_readdir);
798 		kfree(dfi->last_name);
799 		kfree(dfi->dir_info);
800 		kmem_cache_free(ceph_dir_file_cachep, dfi);
801 	} else {
802 		struct ceph_file_info *fi = file->private_data;
803 		dout("release inode %p regular file %p\n", inode, file);
804 		WARN_ON(!list_empty(&fi->rw_contexts));
805 
806 		ceph_put_fmode(ci, fi->fmode, 1);
807 
808 		kmem_cache_free(ceph_file_cachep, fi);
809 	}
810 
811 	/* wake up anyone waiting for caps on this inode */
812 	wake_up_all(&ci->i_cap_wq);
813 	return 0;
814 }
815 
816 enum {
817 	HAVE_RETRIED = 1,
818 	CHECK_EOF =    2,
819 	READ_INLINE =  3,
820 };
821 
822 /*
823  * Completely synchronous read and write methods.  Direct from __user
824  * buffer to osd, or directly to user pages (if O_DIRECT).
825  *
826  * If the read spans object boundary, just do multiple reads.  (That's not
827  * atomic, but good enough for now.)
828  *
829  * If we get a short result from the OSD, check against i_size; we need to
830  * only return a short read to the caller if we hit EOF.
831  */
832 static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *to,
833 			      int *retry_op)
834 {
835 	struct file *file = iocb->ki_filp;
836 	struct inode *inode = file_inode(file);
837 	struct ceph_inode_info *ci = ceph_inode(inode);
838 	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
839 	struct ceph_osd_client *osdc = &fsc->client->osdc;
840 	ssize_t ret;
841 	u64 off = iocb->ki_pos;
842 	u64 len = iov_iter_count(to);
843 
844 	dout("sync_read on file %p %llu~%u %s\n", file, off, (unsigned)len,
845 	     (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
846 
847 	if (!len)
848 		return 0;
849 	/*
850 	 * flush any page cache pages in this range.  this
851 	 * will make concurrent normal and sync io slow,
852 	 * but it will at least behave sensibly when they are
853 	 * in sequence.
854 	 */
855 	ret = filemap_write_and_wait_range(inode->i_mapping,
856 					   off, off + len - 1);
857 	if (ret < 0)
858 		return ret;
859 
860 	ret = 0;
861 	while ((len = iov_iter_count(to)) > 0) {
862 		struct ceph_osd_request *req;
863 		struct page **pages;
864 		int num_pages;
865 		size_t page_off;
866 		u64 i_size;
867 		bool more;
868 
869 		req = ceph_osdc_new_request(osdc, &ci->i_layout,
870 					ci->i_vino, off, &len, 0, 1,
871 					CEPH_OSD_OP_READ, CEPH_OSD_FLAG_READ,
872 					NULL, ci->i_truncate_seq,
873 					ci->i_truncate_size, false);
874 		if (IS_ERR(req)) {
875 			ret = PTR_ERR(req);
876 			break;
877 		}
878 
879 		more = len < iov_iter_count(to);
880 
881 		if (unlikely(iov_iter_is_pipe(to))) {
882 			ret = iov_iter_get_pages_alloc(to, &pages, len,
883 						       &page_off);
884 			if (ret <= 0) {
885 				ceph_osdc_put_request(req);
886 				ret = -ENOMEM;
887 				break;
888 			}
889 			num_pages = DIV_ROUND_UP(ret + page_off, PAGE_SIZE);
890 			if (ret < len) {
891 				len = ret;
892 				osd_req_op_extent_update(req, 0, len);
893 				more = false;
894 			}
895 		} else {
896 			num_pages = calc_pages_for(off, len);
897 			page_off = off & ~PAGE_MASK;
898 			pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
899 			if (IS_ERR(pages)) {
900 				ceph_osdc_put_request(req);
901 				ret = PTR_ERR(pages);
902 				break;
903 			}
904 		}
905 
906 		osd_req_op_extent_osd_data_pages(req, 0, pages, len, page_off,
907 						 false, false);
908 		ret = ceph_osdc_start_request(osdc, req, false);
909 		if (!ret)
910 			ret = ceph_osdc_wait_request(osdc, req);
911 
912 		ceph_update_read_latency(&fsc->mdsc->metric,
913 					 req->r_start_latency,
914 					 req->r_end_latency,
915 					 ret);
916 
917 		ceph_osdc_put_request(req);
918 
919 		i_size = i_size_read(inode);
920 		dout("sync_read %llu~%llu got %zd i_size %llu%s\n",
921 		     off, len, ret, i_size, (more ? " MORE" : ""));
922 
923 		if (ret == -ENOENT)
924 			ret = 0;
925 		if (ret >= 0 && ret < len && (off + ret < i_size)) {
926 			int zlen = min(len - ret, i_size - off - ret);
927 			int zoff = page_off + ret;
928 			dout("sync_read zero gap %llu~%llu\n",
929                              off + ret, off + ret + zlen);
930 			ceph_zero_page_vector_range(zoff, zlen, pages);
931 			ret += zlen;
932 		}
933 
934 		if (unlikely(iov_iter_is_pipe(to))) {
935 			if (ret > 0) {
936 				iov_iter_advance(to, ret);
937 				off += ret;
938 			} else {
939 				iov_iter_advance(to, 0);
940 			}
941 			ceph_put_page_vector(pages, num_pages, false);
942 		} else {
943 			int idx = 0;
944 			size_t left = ret > 0 ? ret : 0;
945 			while (left > 0) {
946 				size_t len, copied;
947 				page_off = off & ~PAGE_MASK;
948 				len = min_t(size_t, left, PAGE_SIZE - page_off);
949 				copied = copy_page_to_iter(pages[idx++],
950 							   page_off, len, to);
951 				off += copied;
952 				left -= copied;
953 				if (copied < len) {
954 					ret = -EFAULT;
955 					break;
956 				}
957 			}
958 			ceph_release_page_vector(pages, num_pages);
959 		}
960 
961 		if (ret < 0) {
962 			if (ret == -EBLACKLISTED)
963 				fsc->blacklisted = true;
964 			break;
965 		}
966 
967 		if (off >= i_size || !more)
968 			break;
969 	}
970 
971 	if (off > iocb->ki_pos) {
972 		if (ret >= 0 &&
973 		    iov_iter_count(to) > 0 && off >= i_size_read(inode))
974 			*retry_op = CHECK_EOF;
975 		ret = off - iocb->ki_pos;
976 		iocb->ki_pos = off;
977 	}
978 
979 	dout("sync_read result %zd retry_op %d\n", ret, *retry_op);
980 	return ret;
981 }
982 
983 struct ceph_aio_request {
984 	struct kiocb *iocb;
985 	size_t total_len;
986 	bool write;
987 	bool should_dirty;
988 	int error;
989 	struct list_head osd_reqs;
990 	unsigned num_reqs;
991 	atomic_t pending_reqs;
992 	struct timespec64 mtime;
993 	struct ceph_cap_flush *prealloc_cf;
994 };
995 
996 struct ceph_aio_work {
997 	struct work_struct work;
998 	struct ceph_osd_request *req;
999 };
1000 
1001 static void ceph_aio_retry_work(struct work_struct *work);
1002 
1003 static void ceph_aio_complete(struct inode *inode,
1004 			      struct ceph_aio_request *aio_req)
1005 {
1006 	struct ceph_inode_info *ci = ceph_inode(inode);
1007 	int ret;
1008 
1009 	if (!atomic_dec_and_test(&aio_req->pending_reqs))
1010 		return;
1011 
1012 	if (aio_req->iocb->ki_flags & IOCB_DIRECT)
1013 		inode_dio_end(inode);
1014 
1015 	ret = aio_req->error;
1016 	if (!ret)
1017 		ret = aio_req->total_len;
1018 
1019 	dout("ceph_aio_complete %p rc %d\n", inode, ret);
1020 
1021 	if (ret >= 0 && aio_req->write) {
1022 		int dirty;
1023 
1024 		loff_t endoff = aio_req->iocb->ki_pos + aio_req->total_len;
1025 		if (endoff > i_size_read(inode)) {
1026 			if (ceph_inode_set_size(inode, endoff))
1027 				ceph_check_caps(ci, CHECK_CAPS_AUTHONLY, NULL);
1028 		}
1029 
1030 		spin_lock(&ci->i_ceph_lock);
1031 		ci->i_inline_version = CEPH_INLINE_NONE;
1032 		dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
1033 					       &aio_req->prealloc_cf);
1034 		spin_unlock(&ci->i_ceph_lock);
1035 		if (dirty)
1036 			__mark_inode_dirty(inode, dirty);
1037 
1038 	}
1039 
1040 	ceph_put_cap_refs(ci, (aio_req->write ? CEPH_CAP_FILE_WR :
1041 						CEPH_CAP_FILE_RD));
1042 
1043 	aio_req->iocb->ki_complete(aio_req->iocb, ret, 0);
1044 
1045 	ceph_free_cap_flush(aio_req->prealloc_cf);
1046 	kfree(aio_req);
1047 }
1048 
1049 static void ceph_aio_complete_req(struct ceph_osd_request *req)
1050 {
1051 	int rc = req->r_result;
1052 	struct inode *inode = req->r_inode;
1053 	struct ceph_aio_request *aio_req = req->r_priv;
1054 	struct ceph_osd_data *osd_data = osd_req_op_extent_osd_data(req, 0);
1055 	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
1056 	struct ceph_client_metric *metric = &fsc->mdsc->metric;
1057 
1058 	BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_BVECS);
1059 	BUG_ON(!osd_data->num_bvecs);
1060 
1061 	dout("ceph_aio_complete_req %p rc %d bytes %u\n",
1062 	     inode, rc, osd_data->bvec_pos.iter.bi_size);
1063 
1064 	/* r_start_latency == 0 means the request was not submitted */
1065 	if (req->r_start_latency) {
1066 		if (aio_req->write)
1067 			ceph_update_write_latency(metric, req->r_start_latency,
1068 						  req->r_end_latency, rc);
1069 		else
1070 			ceph_update_read_latency(metric, req->r_start_latency,
1071 						 req->r_end_latency, rc);
1072 	}
1073 
1074 	if (rc == -EOLDSNAPC) {
1075 		struct ceph_aio_work *aio_work;
1076 		BUG_ON(!aio_req->write);
1077 
1078 		aio_work = kmalloc(sizeof(*aio_work), GFP_NOFS);
1079 		if (aio_work) {
1080 			INIT_WORK(&aio_work->work, ceph_aio_retry_work);
1081 			aio_work->req = req;
1082 			queue_work(ceph_inode_to_client(inode)->inode_wq,
1083 				   &aio_work->work);
1084 			return;
1085 		}
1086 		rc = -ENOMEM;
1087 	} else if (!aio_req->write) {
1088 		if (rc == -ENOENT)
1089 			rc = 0;
1090 		if (rc >= 0 && osd_data->bvec_pos.iter.bi_size > rc) {
1091 			struct iov_iter i;
1092 			int zlen = osd_data->bvec_pos.iter.bi_size - rc;
1093 
1094 			/*
1095 			 * If read is satisfied by single OSD request,
1096 			 * it can pass EOF. Otherwise read is within
1097 			 * i_size.
1098 			 */
1099 			if (aio_req->num_reqs == 1) {
1100 				loff_t i_size = i_size_read(inode);
1101 				loff_t endoff = aio_req->iocb->ki_pos + rc;
1102 				if (endoff < i_size)
1103 					zlen = min_t(size_t, zlen,
1104 						     i_size - endoff);
1105 				aio_req->total_len = rc + zlen;
1106 			}
1107 
1108 			iov_iter_bvec(&i, READ, osd_data->bvec_pos.bvecs,
1109 				      osd_data->num_bvecs,
1110 				      osd_data->bvec_pos.iter.bi_size);
1111 			iov_iter_advance(&i, rc);
1112 			iov_iter_zero(zlen, &i);
1113 		}
1114 	}
1115 
1116 	put_bvecs(osd_data->bvec_pos.bvecs, osd_data->num_bvecs,
1117 		  aio_req->should_dirty);
1118 	ceph_osdc_put_request(req);
1119 
1120 	if (rc < 0)
1121 		cmpxchg(&aio_req->error, 0, rc);
1122 
1123 	ceph_aio_complete(inode, aio_req);
1124 	return;
1125 }
1126 
1127 static void ceph_aio_retry_work(struct work_struct *work)
1128 {
1129 	struct ceph_aio_work *aio_work =
1130 		container_of(work, struct ceph_aio_work, work);
1131 	struct ceph_osd_request *orig_req = aio_work->req;
1132 	struct ceph_aio_request *aio_req = orig_req->r_priv;
1133 	struct inode *inode = orig_req->r_inode;
1134 	struct ceph_inode_info *ci = ceph_inode(inode);
1135 	struct ceph_snap_context *snapc;
1136 	struct ceph_osd_request *req;
1137 	int ret;
1138 
1139 	spin_lock(&ci->i_ceph_lock);
1140 	if (__ceph_have_pending_cap_snap(ci)) {
1141 		struct ceph_cap_snap *capsnap =
1142 			list_last_entry(&ci->i_cap_snaps,
1143 					struct ceph_cap_snap,
1144 					ci_item);
1145 		snapc = ceph_get_snap_context(capsnap->context);
1146 	} else {
1147 		BUG_ON(!ci->i_head_snapc);
1148 		snapc = ceph_get_snap_context(ci->i_head_snapc);
1149 	}
1150 	spin_unlock(&ci->i_ceph_lock);
1151 
1152 	req = ceph_osdc_alloc_request(orig_req->r_osdc, snapc, 1,
1153 			false, GFP_NOFS);
1154 	if (!req) {
1155 		ret = -ENOMEM;
1156 		req = orig_req;
1157 		goto out;
1158 	}
1159 
1160 	req->r_flags = /* CEPH_OSD_FLAG_ORDERSNAP | */ CEPH_OSD_FLAG_WRITE;
1161 	ceph_oloc_copy(&req->r_base_oloc, &orig_req->r_base_oloc);
1162 	ceph_oid_copy(&req->r_base_oid, &orig_req->r_base_oid);
1163 
1164 	req->r_ops[0] = orig_req->r_ops[0];
1165 
1166 	req->r_mtime = aio_req->mtime;
1167 	req->r_data_offset = req->r_ops[0].extent.offset;
1168 
1169 	ret = ceph_osdc_alloc_messages(req, GFP_NOFS);
1170 	if (ret) {
1171 		ceph_osdc_put_request(req);
1172 		req = orig_req;
1173 		goto out;
1174 	}
1175 
1176 	ceph_osdc_put_request(orig_req);
1177 
1178 	req->r_callback = ceph_aio_complete_req;
1179 	req->r_inode = inode;
1180 	req->r_priv = aio_req;
1181 
1182 	ret = ceph_osdc_start_request(req->r_osdc, req, false);
1183 out:
1184 	if (ret < 0) {
1185 		req->r_result = ret;
1186 		ceph_aio_complete_req(req);
1187 	}
1188 
1189 	ceph_put_snap_context(snapc);
1190 	kfree(aio_work);
1191 }
1192 
1193 static ssize_t
1194 ceph_direct_read_write(struct kiocb *iocb, struct iov_iter *iter,
1195 		       struct ceph_snap_context *snapc,
1196 		       struct ceph_cap_flush **pcf)
1197 {
1198 	struct file *file = iocb->ki_filp;
1199 	struct inode *inode = file_inode(file);
1200 	struct ceph_inode_info *ci = ceph_inode(inode);
1201 	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
1202 	struct ceph_client_metric *metric = &fsc->mdsc->metric;
1203 	struct ceph_vino vino;
1204 	struct ceph_osd_request *req;
1205 	struct bio_vec *bvecs;
1206 	struct ceph_aio_request *aio_req = NULL;
1207 	int num_pages = 0;
1208 	int flags;
1209 	int ret = 0;
1210 	struct timespec64 mtime = current_time(inode);
1211 	size_t count = iov_iter_count(iter);
1212 	loff_t pos = iocb->ki_pos;
1213 	bool write = iov_iter_rw(iter) == WRITE;
1214 	bool should_dirty = !write && iter_is_iovec(iter);
1215 
1216 	if (write && ceph_snap(file_inode(file)) != CEPH_NOSNAP)
1217 		return -EROFS;
1218 
1219 	dout("sync_direct_%s on file %p %lld~%u snapc %p seq %lld\n",
1220 	     (write ? "write" : "read"), file, pos, (unsigned)count,
1221 	     snapc, snapc ? snapc->seq : 0);
1222 
1223 	if (write) {
1224 		int ret2 = invalidate_inode_pages2_range(inode->i_mapping,
1225 					pos >> PAGE_SHIFT,
1226 					(pos + count - 1) >> PAGE_SHIFT);
1227 		if (ret2 < 0)
1228 			dout("invalidate_inode_pages2_range returned %d\n", ret2);
1229 
1230 		flags = /* CEPH_OSD_FLAG_ORDERSNAP | */ CEPH_OSD_FLAG_WRITE;
1231 	} else {
1232 		flags = CEPH_OSD_FLAG_READ;
1233 	}
1234 
1235 	while (iov_iter_count(iter) > 0) {
1236 		u64 size = iov_iter_count(iter);
1237 		ssize_t len;
1238 
1239 		if (write)
1240 			size = min_t(u64, size, fsc->mount_options->wsize);
1241 		else
1242 			size = min_t(u64, size, fsc->mount_options->rsize);
1243 
1244 		vino = ceph_vino(inode);
1245 		req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
1246 					    vino, pos, &size, 0,
1247 					    1,
1248 					    write ? CEPH_OSD_OP_WRITE :
1249 						    CEPH_OSD_OP_READ,
1250 					    flags, snapc,
1251 					    ci->i_truncate_seq,
1252 					    ci->i_truncate_size,
1253 					    false);
1254 		if (IS_ERR(req)) {
1255 			ret = PTR_ERR(req);
1256 			break;
1257 		}
1258 
1259 		len = iter_get_bvecs_alloc(iter, size, &bvecs, &num_pages);
1260 		if (len < 0) {
1261 			ceph_osdc_put_request(req);
1262 			ret = len;
1263 			break;
1264 		}
1265 		if (len != size)
1266 			osd_req_op_extent_update(req, 0, len);
1267 
1268 		/*
1269 		 * To simplify error handling, allow AIO when IO within i_size
1270 		 * or IO can be satisfied by single OSD request.
1271 		 */
1272 		if (pos == iocb->ki_pos && !is_sync_kiocb(iocb) &&
1273 		    (len == count || pos + count <= i_size_read(inode))) {
1274 			aio_req = kzalloc(sizeof(*aio_req), GFP_KERNEL);
1275 			if (aio_req) {
1276 				aio_req->iocb = iocb;
1277 				aio_req->write = write;
1278 				aio_req->should_dirty = should_dirty;
1279 				INIT_LIST_HEAD(&aio_req->osd_reqs);
1280 				if (write) {
1281 					aio_req->mtime = mtime;
1282 					swap(aio_req->prealloc_cf, *pcf);
1283 				}
1284 			}
1285 			/* ignore error */
1286 		}
1287 
1288 		if (write) {
1289 			/*
1290 			 * throw out any page cache pages in this range. this
1291 			 * may block.
1292 			 */
1293 			truncate_inode_pages_range(inode->i_mapping, pos,
1294 						   PAGE_ALIGN(pos + len) - 1);
1295 
1296 			req->r_mtime = mtime;
1297 		}
1298 
1299 		osd_req_op_extent_osd_data_bvecs(req, 0, bvecs, num_pages, len);
1300 
1301 		if (aio_req) {
1302 			aio_req->total_len += len;
1303 			aio_req->num_reqs++;
1304 			atomic_inc(&aio_req->pending_reqs);
1305 
1306 			req->r_callback = ceph_aio_complete_req;
1307 			req->r_inode = inode;
1308 			req->r_priv = aio_req;
1309 			list_add_tail(&req->r_private_item, &aio_req->osd_reqs);
1310 
1311 			pos += len;
1312 			continue;
1313 		}
1314 
1315 		ret = ceph_osdc_start_request(req->r_osdc, req, false);
1316 		if (!ret)
1317 			ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
1318 
1319 		if (write)
1320 			ceph_update_write_latency(metric, req->r_start_latency,
1321 						  req->r_end_latency, ret);
1322 		else
1323 			ceph_update_read_latency(metric, req->r_start_latency,
1324 						 req->r_end_latency, ret);
1325 
1326 		size = i_size_read(inode);
1327 		if (!write) {
1328 			if (ret == -ENOENT)
1329 				ret = 0;
1330 			if (ret >= 0 && ret < len && pos + ret < size) {
1331 				struct iov_iter i;
1332 				int zlen = min_t(size_t, len - ret,
1333 						 size - pos - ret);
1334 
1335 				iov_iter_bvec(&i, READ, bvecs, num_pages, len);
1336 				iov_iter_advance(&i, ret);
1337 				iov_iter_zero(zlen, &i);
1338 				ret += zlen;
1339 			}
1340 			if (ret >= 0)
1341 				len = ret;
1342 		}
1343 
1344 		put_bvecs(bvecs, num_pages, should_dirty);
1345 		ceph_osdc_put_request(req);
1346 		if (ret < 0)
1347 			break;
1348 
1349 		pos += len;
1350 		if (!write && pos >= size)
1351 			break;
1352 
1353 		if (write && pos > size) {
1354 			if (ceph_inode_set_size(inode, pos))
1355 				ceph_check_caps(ceph_inode(inode),
1356 						CHECK_CAPS_AUTHONLY,
1357 						NULL);
1358 		}
1359 	}
1360 
1361 	if (aio_req) {
1362 		LIST_HEAD(osd_reqs);
1363 
1364 		if (aio_req->num_reqs == 0) {
1365 			kfree(aio_req);
1366 			return ret;
1367 		}
1368 
1369 		ceph_get_cap_refs(ci, write ? CEPH_CAP_FILE_WR :
1370 					      CEPH_CAP_FILE_RD);
1371 
1372 		list_splice(&aio_req->osd_reqs, &osd_reqs);
1373 		inode_dio_begin(inode);
1374 		while (!list_empty(&osd_reqs)) {
1375 			req = list_first_entry(&osd_reqs,
1376 					       struct ceph_osd_request,
1377 					       r_private_item);
1378 			list_del_init(&req->r_private_item);
1379 			if (ret >= 0)
1380 				ret = ceph_osdc_start_request(req->r_osdc,
1381 							      req, false);
1382 			if (ret < 0) {
1383 				req->r_result = ret;
1384 				ceph_aio_complete_req(req);
1385 			}
1386 		}
1387 		return -EIOCBQUEUED;
1388 	}
1389 
1390 	if (ret != -EOLDSNAPC && pos > iocb->ki_pos) {
1391 		ret = pos - iocb->ki_pos;
1392 		iocb->ki_pos = pos;
1393 	}
1394 	return ret;
1395 }
1396 
1397 /*
1398  * Synchronous write, straight from __user pointer or user pages.
1399  *
1400  * If write spans object boundary, just do multiple writes.  (For a
1401  * correct atomic write, we should e.g. take write locks on all
1402  * objects, rollback on failure, etc.)
1403  */
1404 static ssize_t
1405 ceph_sync_write(struct kiocb *iocb, struct iov_iter *from, loff_t pos,
1406 		struct ceph_snap_context *snapc)
1407 {
1408 	struct file *file = iocb->ki_filp;
1409 	struct inode *inode = file_inode(file);
1410 	struct ceph_inode_info *ci = ceph_inode(inode);
1411 	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
1412 	struct ceph_vino vino;
1413 	struct ceph_osd_request *req;
1414 	struct page **pages;
1415 	u64 len;
1416 	int num_pages;
1417 	int written = 0;
1418 	int flags;
1419 	int ret;
1420 	bool check_caps = false;
1421 	struct timespec64 mtime = current_time(inode);
1422 	size_t count = iov_iter_count(from);
1423 
1424 	if (ceph_snap(file_inode(file)) != CEPH_NOSNAP)
1425 		return -EROFS;
1426 
1427 	dout("sync_write on file %p %lld~%u snapc %p seq %lld\n",
1428 	     file, pos, (unsigned)count, snapc, snapc->seq);
1429 
1430 	ret = filemap_write_and_wait_range(inode->i_mapping,
1431 					   pos, pos + count - 1);
1432 	if (ret < 0)
1433 		return ret;
1434 
1435 	ret = invalidate_inode_pages2_range(inode->i_mapping,
1436 					    pos >> PAGE_SHIFT,
1437 					    (pos + count - 1) >> PAGE_SHIFT);
1438 	if (ret < 0)
1439 		dout("invalidate_inode_pages2_range returned %d\n", ret);
1440 
1441 	flags = /* CEPH_OSD_FLAG_ORDERSNAP | */ CEPH_OSD_FLAG_WRITE;
1442 
1443 	while ((len = iov_iter_count(from)) > 0) {
1444 		size_t left;
1445 		int n;
1446 
1447 		vino = ceph_vino(inode);
1448 		req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
1449 					    vino, pos, &len, 0, 1,
1450 					    CEPH_OSD_OP_WRITE, flags, snapc,
1451 					    ci->i_truncate_seq,
1452 					    ci->i_truncate_size,
1453 					    false);
1454 		if (IS_ERR(req)) {
1455 			ret = PTR_ERR(req);
1456 			break;
1457 		}
1458 
1459 		/*
1460 		 * write from beginning of first page,
1461 		 * regardless of io alignment
1462 		 */
1463 		num_pages = (len + PAGE_SIZE - 1) >> PAGE_SHIFT;
1464 
1465 		pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1466 		if (IS_ERR(pages)) {
1467 			ret = PTR_ERR(pages);
1468 			goto out;
1469 		}
1470 
1471 		left = len;
1472 		for (n = 0; n < num_pages; n++) {
1473 			size_t plen = min_t(size_t, left, PAGE_SIZE);
1474 			ret = copy_page_from_iter(pages[n], 0, plen, from);
1475 			if (ret != plen) {
1476 				ret = -EFAULT;
1477 				break;
1478 			}
1479 			left -= ret;
1480 		}
1481 
1482 		if (ret < 0) {
1483 			ceph_release_page_vector(pages, num_pages);
1484 			goto out;
1485 		}
1486 
1487 		req->r_inode = inode;
1488 
1489 		osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0,
1490 						false, true);
1491 
1492 		req->r_mtime = mtime;
1493 		ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
1494 		if (!ret)
1495 			ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
1496 
1497 		ceph_update_write_latency(&fsc->mdsc->metric, req->r_start_latency,
1498 					  req->r_end_latency, ret);
1499 out:
1500 		ceph_osdc_put_request(req);
1501 		if (ret != 0) {
1502 			ceph_set_error_write(ci);
1503 			break;
1504 		}
1505 
1506 		ceph_clear_error_write(ci);
1507 		pos += len;
1508 		written += len;
1509 		if (pos > i_size_read(inode)) {
1510 			check_caps = ceph_inode_set_size(inode, pos);
1511 			if (check_caps)
1512 				ceph_check_caps(ceph_inode(inode),
1513 						CHECK_CAPS_AUTHONLY,
1514 						NULL);
1515 		}
1516 
1517 	}
1518 
1519 	if (ret != -EOLDSNAPC && written > 0) {
1520 		ret = written;
1521 		iocb->ki_pos = pos;
1522 	}
1523 	return ret;
1524 }
1525 
1526 /*
1527  * Wrap generic_file_aio_read with checks for cap bits on the inode.
1528  * Atomically grab references, so that those bits are not released
1529  * back to the MDS mid-read.
1530  *
1531  * Hmm, the sync read case isn't actually async... should it be?
1532  */
1533 static ssize_t ceph_read_iter(struct kiocb *iocb, struct iov_iter *to)
1534 {
1535 	struct file *filp = iocb->ki_filp;
1536 	struct ceph_file_info *fi = filp->private_data;
1537 	size_t len = iov_iter_count(to);
1538 	struct inode *inode = file_inode(filp);
1539 	struct ceph_inode_info *ci = ceph_inode(inode);
1540 	struct page *pinned_page = NULL;
1541 	ssize_t ret;
1542 	int want, got = 0;
1543 	int retry_op = 0, read = 0;
1544 
1545 again:
1546 	dout("aio_read %p %llx.%llx %llu~%u trying to get caps on %p\n",
1547 	     inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len, inode);
1548 
1549 	if (iocb->ki_flags & IOCB_DIRECT)
1550 		ceph_start_io_direct(inode);
1551 	else
1552 		ceph_start_io_read(inode);
1553 
1554 	if (fi->fmode & CEPH_FILE_MODE_LAZY)
1555 		want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
1556 	else
1557 		want = CEPH_CAP_FILE_CACHE;
1558 	ret = ceph_get_caps(filp, CEPH_CAP_FILE_RD, want, -1,
1559 			    &got, &pinned_page);
1560 	if (ret < 0) {
1561 		if (iocb->ki_flags & IOCB_DIRECT)
1562 			ceph_end_io_direct(inode);
1563 		else
1564 			ceph_end_io_read(inode);
1565 		return ret;
1566 	}
1567 
1568 	if ((got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0 ||
1569 	    (iocb->ki_flags & IOCB_DIRECT) ||
1570 	    (fi->flags & CEPH_F_SYNC)) {
1571 
1572 		dout("aio_sync_read %p %llx.%llx %llu~%u got cap refs on %s\n",
1573 		     inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len,
1574 		     ceph_cap_string(got));
1575 
1576 		if (ci->i_inline_version == CEPH_INLINE_NONE) {
1577 			if (!retry_op && (iocb->ki_flags & IOCB_DIRECT)) {
1578 				ret = ceph_direct_read_write(iocb, to,
1579 							     NULL, NULL);
1580 				if (ret >= 0 && ret < len)
1581 					retry_op = CHECK_EOF;
1582 			} else {
1583 				ret = ceph_sync_read(iocb, to, &retry_op);
1584 			}
1585 		} else {
1586 			retry_op = READ_INLINE;
1587 		}
1588 	} else {
1589 		CEPH_DEFINE_RW_CONTEXT(rw_ctx, got);
1590 		dout("aio_read %p %llx.%llx %llu~%u got cap refs on %s\n",
1591 		     inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len,
1592 		     ceph_cap_string(got));
1593 		ceph_add_rw_context(fi, &rw_ctx);
1594 		ret = generic_file_read_iter(iocb, to);
1595 		ceph_del_rw_context(fi, &rw_ctx);
1596 	}
1597 
1598 	dout("aio_read %p %llx.%llx dropping cap refs on %s = %d\n",
1599 	     inode, ceph_vinop(inode), ceph_cap_string(got), (int)ret);
1600 	if (pinned_page) {
1601 		put_page(pinned_page);
1602 		pinned_page = NULL;
1603 	}
1604 	ceph_put_cap_refs(ci, got);
1605 
1606 	if (iocb->ki_flags & IOCB_DIRECT)
1607 		ceph_end_io_direct(inode);
1608 	else
1609 		ceph_end_io_read(inode);
1610 
1611 	if (retry_op > HAVE_RETRIED && ret >= 0) {
1612 		int statret;
1613 		struct page *page = NULL;
1614 		loff_t i_size;
1615 		if (retry_op == READ_INLINE) {
1616 			page = __page_cache_alloc(GFP_KERNEL);
1617 			if (!page)
1618 				return -ENOMEM;
1619 		}
1620 
1621 		statret = __ceph_do_getattr(inode, page,
1622 					    CEPH_STAT_CAP_INLINE_DATA, !!page);
1623 		if (statret < 0) {
1624 			if (page)
1625 				__free_page(page);
1626 			if (statret == -ENODATA) {
1627 				BUG_ON(retry_op != READ_INLINE);
1628 				goto again;
1629 			}
1630 			return statret;
1631 		}
1632 
1633 		i_size = i_size_read(inode);
1634 		if (retry_op == READ_INLINE) {
1635 			BUG_ON(ret > 0 || read > 0);
1636 			if (iocb->ki_pos < i_size &&
1637 			    iocb->ki_pos < PAGE_SIZE) {
1638 				loff_t end = min_t(loff_t, i_size,
1639 						   iocb->ki_pos + len);
1640 				end = min_t(loff_t, end, PAGE_SIZE);
1641 				if (statret < end)
1642 					zero_user_segment(page, statret, end);
1643 				ret = copy_page_to_iter(page,
1644 						iocb->ki_pos & ~PAGE_MASK,
1645 						end - iocb->ki_pos, to);
1646 				iocb->ki_pos += ret;
1647 				read += ret;
1648 			}
1649 			if (iocb->ki_pos < i_size && read < len) {
1650 				size_t zlen = min_t(size_t, len - read,
1651 						    i_size - iocb->ki_pos);
1652 				ret = iov_iter_zero(zlen, to);
1653 				iocb->ki_pos += ret;
1654 				read += ret;
1655 			}
1656 			__free_pages(page, 0);
1657 			return read;
1658 		}
1659 
1660 		/* hit EOF or hole? */
1661 		if (retry_op == CHECK_EOF && iocb->ki_pos < i_size &&
1662 		    ret < len) {
1663 			dout("sync_read hit hole, ppos %lld < size %lld"
1664 			     ", reading more\n", iocb->ki_pos, i_size);
1665 
1666 			read += ret;
1667 			len -= ret;
1668 			retry_op = HAVE_RETRIED;
1669 			goto again;
1670 		}
1671 	}
1672 
1673 	if (ret >= 0)
1674 		ret += read;
1675 
1676 	return ret;
1677 }
1678 
1679 /*
1680  * Take cap references to avoid releasing caps to MDS mid-write.
1681  *
1682  * If we are synchronous, and write with an old snap context, the OSD
1683  * may return EOLDSNAPC.  In that case, retry the write.. _after_
1684  * dropping our cap refs and allowing the pending snap to logically
1685  * complete _before_ this write occurs.
1686  *
1687  * If we are near ENOSPC, write synchronously.
1688  */
1689 static ssize_t ceph_write_iter(struct kiocb *iocb, struct iov_iter *from)
1690 {
1691 	struct file *file = iocb->ki_filp;
1692 	struct ceph_file_info *fi = file->private_data;
1693 	struct inode *inode = file_inode(file);
1694 	struct ceph_inode_info *ci = ceph_inode(inode);
1695 	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
1696 	struct ceph_osd_client *osdc = &fsc->client->osdc;
1697 	struct ceph_cap_flush *prealloc_cf;
1698 	ssize_t count, written = 0;
1699 	int err, want, got;
1700 	bool direct_lock = false;
1701 	u32 map_flags;
1702 	u64 pool_flags;
1703 	loff_t pos;
1704 	loff_t limit = max(i_size_read(inode), fsc->max_file_size);
1705 
1706 	if (ceph_snap(inode) != CEPH_NOSNAP)
1707 		return -EROFS;
1708 
1709 	prealloc_cf = ceph_alloc_cap_flush();
1710 	if (!prealloc_cf)
1711 		return -ENOMEM;
1712 
1713 	if ((iocb->ki_flags & (IOCB_DIRECT | IOCB_APPEND)) == IOCB_DIRECT)
1714 		direct_lock = true;
1715 
1716 retry_snap:
1717 	if (direct_lock)
1718 		ceph_start_io_direct(inode);
1719 	else
1720 		ceph_start_io_write(inode);
1721 
1722 	/* We can write back this queue in page reclaim */
1723 	current->backing_dev_info = inode_to_bdi(inode);
1724 
1725 	if (iocb->ki_flags & IOCB_APPEND) {
1726 		err = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE, false);
1727 		if (err < 0)
1728 			goto out;
1729 	}
1730 
1731 	err = generic_write_checks(iocb, from);
1732 	if (err <= 0)
1733 		goto out;
1734 
1735 	pos = iocb->ki_pos;
1736 	if (unlikely(pos >= limit)) {
1737 		err = -EFBIG;
1738 		goto out;
1739 	} else {
1740 		iov_iter_truncate(from, limit - pos);
1741 	}
1742 
1743 	count = iov_iter_count(from);
1744 	if (ceph_quota_is_max_bytes_exceeded(inode, pos + count)) {
1745 		err = -EDQUOT;
1746 		goto out;
1747 	}
1748 
1749 	err = file_remove_privs(file);
1750 	if (err)
1751 		goto out;
1752 
1753 	err = file_update_time(file);
1754 	if (err)
1755 		goto out;
1756 
1757 	inode_inc_iversion_raw(inode);
1758 
1759 	if (ci->i_inline_version != CEPH_INLINE_NONE) {
1760 		err = ceph_uninline_data(file, NULL);
1761 		if (err < 0)
1762 			goto out;
1763 	}
1764 
1765 	down_read(&osdc->lock);
1766 	map_flags = osdc->osdmap->flags;
1767 	pool_flags = ceph_pg_pool_flags(osdc->osdmap, ci->i_layout.pool_id);
1768 	up_read(&osdc->lock);
1769 	if ((map_flags & CEPH_OSDMAP_FULL) ||
1770 	    (pool_flags & CEPH_POOL_FLAG_FULL)) {
1771 		err = -ENOSPC;
1772 		goto out;
1773 	}
1774 
1775 	dout("aio_write %p %llx.%llx %llu~%zd getting caps. i_size %llu\n",
1776 	     inode, ceph_vinop(inode), pos, count, i_size_read(inode));
1777 	if (fi->fmode & CEPH_FILE_MODE_LAZY)
1778 		want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
1779 	else
1780 		want = CEPH_CAP_FILE_BUFFER;
1781 	got = 0;
1782 	err = ceph_get_caps(file, CEPH_CAP_FILE_WR, want, pos + count,
1783 			    &got, NULL);
1784 	if (err < 0)
1785 		goto out;
1786 
1787 	dout("aio_write %p %llx.%llx %llu~%zd got cap refs on %s\n",
1788 	     inode, ceph_vinop(inode), pos, count, ceph_cap_string(got));
1789 
1790 	if ((got & (CEPH_CAP_FILE_BUFFER|CEPH_CAP_FILE_LAZYIO)) == 0 ||
1791 	    (iocb->ki_flags & IOCB_DIRECT) || (fi->flags & CEPH_F_SYNC) ||
1792 	    (ci->i_ceph_flags & CEPH_I_ERROR_WRITE)) {
1793 		struct ceph_snap_context *snapc;
1794 		struct iov_iter data;
1795 
1796 		spin_lock(&ci->i_ceph_lock);
1797 		if (__ceph_have_pending_cap_snap(ci)) {
1798 			struct ceph_cap_snap *capsnap =
1799 					list_last_entry(&ci->i_cap_snaps,
1800 							struct ceph_cap_snap,
1801 							ci_item);
1802 			snapc = ceph_get_snap_context(capsnap->context);
1803 		} else {
1804 			BUG_ON(!ci->i_head_snapc);
1805 			snapc = ceph_get_snap_context(ci->i_head_snapc);
1806 		}
1807 		spin_unlock(&ci->i_ceph_lock);
1808 
1809 		/* we might need to revert back to that point */
1810 		data = *from;
1811 		if (iocb->ki_flags & IOCB_DIRECT)
1812 			written = ceph_direct_read_write(iocb, &data, snapc,
1813 							 &prealloc_cf);
1814 		else
1815 			written = ceph_sync_write(iocb, &data, pos, snapc);
1816 		if (direct_lock)
1817 			ceph_end_io_direct(inode);
1818 		else
1819 			ceph_end_io_write(inode);
1820 		if (written > 0)
1821 			iov_iter_advance(from, written);
1822 		ceph_put_snap_context(snapc);
1823 	} else {
1824 		/*
1825 		 * No need to acquire the i_truncate_mutex. Because
1826 		 * the MDS revokes Fwb caps before sending truncate
1827 		 * message to us. We can't get Fwb cap while there
1828 		 * are pending vmtruncate. So write and vmtruncate
1829 		 * can not run at the same time
1830 		 */
1831 		written = generic_perform_write(file, from, pos);
1832 		if (likely(written >= 0))
1833 			iocb->ki_pos = pos + written;
1834 		ceph_end_io_write(inode);
1835 	}
1836 
1837 	if (written >= 0) {
1838 		int dirty;
1839 
1840 		spin_lock(&ci->i_ceph_lock);
1841 		ci->i_inline_version = CEPH_INLINE_NONE;
1842 		dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
1843 					       &prealloc_cf);
1844 		spin_unlock(&ci->i_ceph_lock);
1845 		if (dirty)
1846 			__mark_inode_dirty(inode, dirty);
1847 		if (ceph_quota_is_max_bytes_approaching(inode, iocb->ki_pos))
1848 			ceph_check_caps(ci, 0, NULL);
1849 	}
1850 
1851 	dout("aio_write %p %llx.%llx %llu~%u  dropping cap refs on %s\n",
1852 	     inode, ceph_vinop(inode), pos, (unsigned)count,
1853 	     ceph_cap_string(got));
1854 	ceph_put_cap_refs(ci, got);
1855 
1856 	if (written == -EOLDSNAPC) {
1857 		dout("aio_write %p %llx.%llx %llu~%u" "got EOLDSNAPC, retrying\n",
1858 		     inode, ceph_vinop(inode), pos, (unsigned)count);
1859 		goto retry_snap;
1860 	}
1861 
1862 	if (written >= 0) {
1863 		if ((map_flags & CEPH_OSDMAP_NEARFULL) ||
1864 		    (pool_flags & CEPH_POOL_FLAG_NEARFULL))
1865 			iocb->ki_flags |= IOCB_DSYNC;
1866 		written = generic_write_sync(iocb, written);
1867 	}
1868 
1869 	goto out_unlocked;
1870 out:
1871 	if (direct_lock)
1872 		ceph_end_io_direct(inode);
1873 	else
1874 		ceph_end_io_write(inode);
1875 out_unlocked:
1876 	ceph_free_cap_flush(prealloc_cf);
1877 	current->backing_dev_info = NULL;
1878 	return written ? written : err;
1879 }
1880 
1881 /*
1882  * llseek.  be sure to verify file size on SEEK_END.
1883  */
1884 static loff_t ceph_llseek(struct file *file, loff_t offset, int whence)
1885 {
1886 	struct inode *inode = file->f_mapping->host;
1887 	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
1888 	loff_t i_size;
1889 	loff_t ret;
1890 
1891 	inode_lock(inode);
1892 
1893 	if (whence == SEEK_END || whence == SEEK_DATA || whence == SEEK_HOLE) {
1894 		ret = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE, false);
1895 		if (ret < 0)
1896 			goto out;
1897 	}
1898 
1899 	i_size = i_size_read(inode);
1900 	switch (whence) {
1901 	case SEEK_END:
1902 		offset += i_size;
1903 		break;
1904 	case SEEK_CUR:
1905 		/*
1906 		 * Here we special-case the lseek(fd, 0, SEEK_CUR)
1907 		 * position-querying operation.  Avoid rewriting the "same"
1908 		 * f_pos value back to the file because a concurrent read(),
1909 		 * write() or lseek() might have altered it
1910 		 */
1911 		if (offset == 0) {
1912 			ret = file->f_pos;
1913 			goto out;
1914 		}
1915 		offset += file->f_pos;
1916 		break;
1917 	case SEEK_DATA:
1918 		if (offset < 0 || offset >= i_size) {
1919 			ret = -ENXIO;
1920 			goto out;
1921 		}
1922 		break;
1923 	case SEEK_HOLE:
1924 		if (offset < 0 || offset >= i_size) {
1925 			ret = -ENXIO;
1926 			goto out;
1927 		}
1928 		offset = i_size;
1929 		break;
1930 	}
1931 
1932 	ret = vfs_setpos(file, offset, max(i_size, fsc->max_file_size));
1933 
1934 out:
1935 	inode_unlock(inode);
1936 	return ret;
1937 }
1938 
1939 static inline void ceph_zero_partial_page(
1940 	struct inode *inode, loff_t offset, unsigned size)
1941 {
1942 	struct page *page;
1943 	pgoff_t index = offset >> PAGE_SHIFT;
1944 
1945 	page = find_lock_page(inode->i_mapping, index);
1946 	if (page) {
1947 		wait_on_page_writeback(page);
1948 		zero_user(page, offset & (PAGE_SIZE - 1), size);
1949 		unlock_page(page);
1950 		put_page(page);
1951 	}
1952 }
1953 
1954 static void ceph_zero_pagecache_range(struct inode *inode, loff_t offset,
1955 				      loff_t length)
1956 {
1957 	loff_t nearly = round_up(offset, PAGE_SIZE);
1958 	if (offset < nearly) {
1959 		loff_t size = nearly - offset;
1960 		if (length < size)
1961 			size = length;
1962 		ceph_zero_partial_page(inode, offset, size);
1963 		offset += size;
1964 		length -= size;
1965 	}
1966 	if (length >= PAGE_SIZE) {
1967 		loff_t size = round_down(length, PAGE_SIZE);
1968 		truncate_pagecache_range(inode, offset, offset + size - 1);
1969 		offset += size;
1970 		length -= size;
1971 	}
1972 	if (length)
1973 		ceph_zero_partial_page(inode, offset, length);
1974 }
1975 
1976 static int ceph_zero_partial_object(struct inode *inode,
1977 				    loff_t offset, loff_t *length)
1978 {
1979 	struct ceph_inode_info *ci = ceph_inode(inode);
1980 	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
1981 	struct ceph_osd_request *req;
1982 	int ret = 0;
1983 	loff_t zero = 0;
1984 	int op;
1985 
1986 	if (!length) {
1987 		op = offset ? CEPH_OSD_OP_DELETE : CEPH_OSD_OP_TRUNCATE;
1988 		length = &zero;
1989 	} else {
1990 		op = CEPH_OSD_OP_ZERO;
1991 	}
1992 
1993 	req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
1994 					ceph_vino(inode),
1995 					offset, length,
1996 					0, 1, op,
1997 					CEPH_OSD_FLAG_WRITE,
1998 					NULL, 0, 0, false);
1999 	if (IS_ERR(req)) {
2000 		ret = PTR_ERR(req);
2001 		goto out;
2002 	}
2003 
2004 	req->r_mtime = inode->i_mtime;
2005 	ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
2006 	if (!ret) {
2007 		ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
2008 		if (ret == -ENOENT)
2009 			ret = 0;
2010 	}
2011 	ceph_osdc_put_request(req);
2012 
2013 out:
2014 	return ret;
2015 }
2016 
2017 static int ceph_zero_objects(struct inode *inode, loff_t offset, loff_t length)
2018 {
2019 	int ret = 0;
2020 	struct ceph_inode_info *ci = ceph_inode(inode);
2021 	s32 stripe_unit = ci->i_layout.stripe_unit;
2022 	s32 stripe_count = ci->i_layout.stripe_count;
2023 	s32 object_size = ci->i_layout.object_size;
2024 	u64 object_set_size = object_size * stripe_count;
2025 	u64 nearly, t;
2026 
2027 	/* round offset up to next period boundary */
2028 	nearly = offset + object_set_size - 1;
2029 	t = nearly;
2030 	nearly -= do_div(t, object_set_size);
2031 
2032 	while (length && offset < nearly) {
2033 		loff_t size = length;
2034 		ret = ceph_zero_partial_object(inode, offset, &size);
2035 		if (ret < 0)
2036 			return ret;
2037 		offset += size;
2038 		length -= size;
2039 	}
2040 	while (length >= object_set_size) {
2041 		int i;
2042 		loff_t pos = offset;
2043 		for (i = 0; i < stripe_count; ++i) {
2044 			ret = ceph_zero_partial_object(inode, pos, NULL);
2045 			if (ret < 0)
2046 				return ret;
2047 			pos += stripe_unit;
2048 		}
2049 		offset += object_set_size;
2050 		length -= object_set_size;
2051 	}
2052 	while (length) {
2053 		loff_t size = length;
2054 		ret = ceph_zero_partial_object(inode, offset, &size);
2055 		if (ret < 0)
2056 			return ret;
2057 		offset += size;
2058 		length -= size;
2059 	}
2060 	return ret;
2061 }
2062 
2063 static long ceph_fallocate(struct file *file, int mode,
2064 				loff_t offset, loff_t length)
2065 {
2066 	struct ceph_file_info *fi = file->private_data;
2067 	struct inode *inode = file_inode(file);
2068 	struct ceph_inode_info *ci = ceph_inode(inode);
2069 	struct ceph_cap_flush *prealloc_cf;
2070 	int want, got = 0;
2071 	int dirty;
2072 	int ret = 0;
2073 	loff_t endoff = 0;
2074 	loff_t size;
2075 
2076 	if (mode != (FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE))
2077 		return -EOPNOTSUPP;
2078 
2079 	if (!S_ISREG(inode->i_mode))
2080 		return -EOPNOTSUPP;
2081 
2082 	prealloc_cf = ceph_alloc_cap_flush();
2083 	if (!prealloc_cf)
2084 		return -ENOMEM;
2085 
2086 	inode_lock(inode);
2087 
2088 	if (ceph_snap(inode) != CEPH_NOSNAP) {
2089 		ret = -EROFS;
2090 		goto unlock;
2091 	}
2092 
2093 	if (ci->i_inline_version != CEPH_INLINE_NONE) {
2094 		ret = ceph_uninline_data(file, NULL);
2095 		if (ret < 0)
2096 			goto unlock;
2097 	}
2098 
2099 	size = i_size_read(inode);
2100 
2101 	/* Are we punching a hole beyond EOF? */
2102 	if (offset >= size)
2103 		goto unlock;
2104 	if ((offset + length) > size)
2105 		length = size - offset;
2106 
2107 	if (fi->fmode & CEPH_FILE_MODE_LAZY)
2108 		want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
2109 	else
2110 		want = CEPH_CAP_FILE_BUFFER;
2111 
2112 	ret = ceph_get_caps(file, CEPH_CAP_FILE_WR, want, endoff, &got, NULL);
2113 	if (ret < 0)
2114 		goto unlock;
2115 
2116 	ceph_zero_pagecache_range(inode, offset, length);
2117 	ret = ceph_zero_objects(inode, offset, length);
2118 
2119 	if (!ret) {
2120 		spin_lock(&ci->i_ceph_lock);
2121 		ci->i_inline_version = CEPH_INLINE_NONE;
2122 		dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
2123 					       &prealloc_cf);
2124 		spin_unlock(&ci->i_ceph_lock);
2125 		if (dirty)
2126 			__mark_inode_dirty(inode, dirty);
2127 	}
2128 
2129 	ceph_put_cap_refs(ci, got);
2130 unlock:
2131 	inode_unlock(inode);
2132 	ceph_free_cap_flush(prealloc_cf);
2133 	return ret;
2134 }
2135 
2136 /*
2137  * This function tries to get FILE_WR capabilities for dst_ci and FILE_RD for
2138  * src_ci.  Two attempts are made to obtain both caps, and an error is return if
2139  * this fails; zero is returned on success.
2140  */
2141 static int get_rd_wr_caps(struct file *src_filp, int *src_got,
2142 			  struct file *dst_filp,
2143 			  loff_t dst_endoff, int *dst_got)
2144 {
2145 	int ret = 0;
2146 	bool retrying = false;
2147 
2148 retry_caps:
2149 	ret = ceph_get_caps(dst_filp, CEPH_CAP_FILE_WR, CEPH_CAP_FILE_BUFFER,
2150 			    dst_endoff, dst_got, NULL);
2151 	if (ret < 0)
2152 		return ret;
2153 
2154 	/*
2155 	 * Since we're already holding the FILE_WR capability for the dst file,
2156 	 * we would risk a deadlock by using ceph_get_caps.  Thus, we'll do some
2157 	 * retry dance instead to try to get both capabilities.
2158 	 */
2159 	ret = ceph_try_get_caps(file_inode(src_filp),
2160 				CEPH_CAP_FILE_RD, CEPH_CAP_FILE_SHARED,
2161 				false, src_got);
2162 	if (ret <= 0) {
2163 		/* Start by dropping dst_ci caps and getting src_ci caps */
2164 		ceph_put_cap_refs(ceph_inode(file_inode(dst_filp)), *dst_got);
2165 		if (retrying) {
2166 			if (!ret)
2167 				/* ceph_try_get_caps masks EAGAIN */
2168 				ret = -EAGAIN;
2169 			return ret;
2170 		}
2171 		ret = ceph_get_caps(src_filp, CEPH_CAP_FILE_RD,
2172 				    CEPH_CAP_FILE_SHARED, -1, src_got, NULL);
2173 		if (ret < 0)
2174 			return ret;
2175 		/*... drop src_ci caps too, and retry */
2176 		ceph_put_cap_refs(ceph_inode(file_inode(src_filp)), *src_got);
2177 		retrying = true;
2178 		goto retry_caps;
2179 	}
2180 	return ret;
2181 }
2182 
2183 static void put_rd_wr_caps(struct ceph_inode_info *src_ci, int src_got,
2184 			   struct ceph_inode_info *dst_ci, int dst_got)
2185 {
2186 	ceph_put_cap_refs(src_ci, src_got);
2187 	ceph_put_cap_refs(dst_ci, dst_got);
2188 }
2189 
2190 /*
2191  * This function does several size-related checks, returning an error if:
2192  *  - source file is smaller than off+len
2193  *  - destination file size is not OK (inode_newsize_ok())
2194  *  - max bytes quotas is exceeded
2195  */
2196 static int is_file_size_ok(struct inode *src_inode, struct inode *dst_inode,
2197 			   loff_t src_off, loff_t dst_off, size_t len)
2198 {
2199 	loff_t size, endoff;
2200 
2201 	size = i_size_read(src_inode);
2202 	/*
2203 	 * Don't copy beyond source file EOF.  Instead of simply setting length
2204 	 * to (size - src_off), just drop to VFS default implementation, as the
2205 	 * local i_size may be stale due to other clients writing to the source
2206 	 * inode.
2207 	 */
2208 	if (src_off + len > size) {
2209 		dout("Copy beyond EOF (%llu + %zu > %llu)\n",
2210 		     src_off, len, size);
2211 		return -EOPNOTSUPP;
2212 	}
2213 	size = i_size_read(dst_inode);
2214 
2215 	endoff = dst_off + len;
2216 	if (inode_newsize_ok(dst_inode, endoff))
2217 		return -EOPNOTSUPP;
2218 
2219 	if (ceph_quota_is_max_bytes_exceeded(dst_inode, endoff))
2220 		return -EDQUOT;
2221 
2222 	return 0;
2223 }
2224 
2225 static ssize_t ceph_do_objects_copy(struct ceph_inode_info *src_ci, u64 *src_off,
2226 				    struct ceph_inode_info *dst_ci, u64 *dst_off,
2227 				    struct ceph_fs_client *fsc,
2228 				    size_t len, unsigned int flags)
2229 {
2230 	struct ceph_object_locator src_oloc, dst_oloc;
2231 	struct ceph_object_id src_oid, dst_oid;
2232 	size_t bytes = 0;
2233 	u64 src_objnum, src_objoff, dst_objnum, dst_objoff;
2234 	u32 src_objlen, dst_objlen;
2235 	u32 object_size = src_ci->i_layout.object_size;
2236 	int ret;
2237 
2238 	src_oloc.pool = src_ci->i_layout.pool_id;
2239 	src_oloc.pool_ns = ceph_try_get_string(src_ci->i_layout.pool_ns);
2240 	dst_oloc.pool = dst_ci->i_layout.pool_id;
2241 	dst_oloc.pool_ns = ceph_try_get_string(dst_ci->i_layout.pool_ns);
2242 
2243 	while (len >= object_size) {
2244 		ceph_calc_file_object_mapping(&src_ci->i_layout, *src_off,
2245 					      object_size, &src_objnum,
2246 					      &src_objoff, &src_objlen);
2247 		ceph_calc_file_object_mapping(&dst_ci->i_layout, *dst_off,
2248 					      object_size, &dst_objnum,
2249 					      &dst_objoff, &dst_objlen);
2250 		ceph_oid_init(&src_oid);
2251 		ceph_oid_printf(&src_oid, "%llx.%08llx",
2252 				src_ci->i_vino.ino, src_objnum);
2253 		ceph_oid_init(&dst_oid);
2254 		ceph_oid_printf(&dst_oid, "%llx.%08llx",
2255 				dst_ci->i_vino.ino, dst_objnum);
2256 		/* Do an object remote copy */
2257 		ret = ceph_osdc_copy_from(&fsc->client->osdc,
2258 					  src_ci->i_vino.snap, 0,
2259 					  &src_oid, &src_oloc,
2260 					  CEPH_OSD_OP_FLAG_FADVISE_SEQUENTIAL |
2261 					  CEPH_OSD_OP_FLAG_FADVISE_NOCACHE,
2262 					  &dst_oid, &dst_oloc,
2263 					  CEPH_OSD_OP_FLAG_FADVISE_SEQUENTIAL |
2264 					  CEPH_OSD_OP_FLAG_FADVISE_DONTNEED,
2265 					  dst_ci->i_truncate_seq,
2266 					  dst_ci->i_truncate_size,
2267 					  CEPH_OSD_COPY_FROM_FLAG_TRUNCATE_SEQ);
2268 		if (ret) {
2269 			if (ret == -EOPNOTSUPP) {
2270 				fsc->have_copy_from2 = false;
2271 				pr_notice("OSDs don't support copy-from2; disabling copy offload\n");
2272 			}
2273 			dout("ceph_osdc_copy_from returned %d\n", ret);
2274 			if (!bytes)
2275 				bytes = ret;
2276 			goto out;
2277 		}
2278 		len -= object_size;
2279 		bytes += object_size;
2280 		*src_off += object_size;
2281 		*dst_off += object_size;
2282 	}
2283 
2284 out:
2285 	ceph_oloc_destroy(&src_oloc);
2286 	ceph_oloc_destroy(&dst_oloc);
2287 	return bytes;
2288 }
2289 
2290 static ssize_t __ceph_copy_file_range(struct file *src_file, loff_t src_off,
2291 				      struct file *dst_file, loff_t dst_off,
2292 				      size_t len, unsigned int flags)
2293 {
2294 	struct inode *src_inode = file_inode(src_file);
2295 	struct inode *dst_inode = file_inode(dst_file);
2296 	struct ceph_inode_info *src_ci = ceph_inode(src_inode);
2297 	struct ceph_inode_info *dst_ci = ceph_inode(dst_inode);
2298 	struct ceph_cap_flush *prealloc_cf;
2299 	struct ceph_fs_client *src_fsc = ceph_inode_to_client(src_inode);
2300 	loff_t size;
2301 	ssize_t ret = -EIO, bytes;
2302 	u64 src_objnum, dst_objnum, src_objoff, dst_objoff;
2303 	u32 src_objlen, dst_objlen;
2304 	int src_got = 0, dst_got = 0, err, dirty;
2305 
2306 	if (src_inode->i_sb != dst_inode->i_sb) {
2307 		struct ceph_fs_client *dst_fsc = ceph_inode_to_client(dst_inode);
2308 
2309 		if (ceph_fsid_compare(&src_fsc->client->fsid,
2310 				      &dst_fsc->client->fsid)) {
2311 			dout("Copying files across clusters: src: %pU dst: %pU\n",
2312 			     &src_fsc->client->fsid, &dst_fsc->client->fsid);
2313 			return -EXDEV;
2314 		}
2315 	}
2316 	if (ceph_snap(dst_inode) != CEPH_NOSNAP)
2317 		return -EROFS;
2318 
2319 	/*
2320 	 * Some of the checks below will return -EOPNOTSUPP, which will force a
2321 	 * fallback to the default VFS copy_file_range implementation.  This is
2322 	 * desirable in several cases (for ex, the 'len' is smaller than the
2323 	 * size of the objects, or in cases where that would be more
2324 	 * efficient).
2325 	 */
2326 
2327 	if (ceph_test_mount_opt(src_fsc, NOCOPYFROM))
2328 		return -EOPNOTSUPP;
2329 
2330 	if (!src_fsc->have_copy_from2)
2331 		return -EOPNOTSUPP;
2332 
2333 	/*
2334 	 * Striped file layouts require that we copy partial objects, but the
2335 	 * OSD copy-from operation only supports full-object copies.  Limit
2336 	 * this to non-striped file layouts for now.
2337 	 */
2338 	if ((src_ci->i_layout.stripe_unit != dst_ci->i_layout.stripe_unit) ||
2339 	    (src_ci->i_layout.stripe_count != 1) ||
2340 	    (dst_ci->i_layout.stripe_count != 1) ||
2341 	    (src_ci->i_layout.object_size != dst_ci->i_layout.object_size)) {
2342 		dout("Invalid src/dst files layout\n");
2343 		return -EOPNOTSUPP;
2344 	}
2345 
2346 	if (len < src_ci->i_layout.object_size)
2347 		return -EOPNOTSUPP; /* no remote copy will be done */
2348 
2349 	prealloc_cf = ceph_alloc_cap_flush();
2350 	if (!prealloc_cf)
2351 		return -ENOMEM;
2352 
2353 	/* Start by sync'ing the source and destination files */
2354 	ret = file_write_and_wait_range(src_file, src_off, (src_off + len));
2355 	if (ret < 0) {
2356 		dout("failed to write src file (%zd)\n", ret);
2357 		goto out;
2358 	}
2359 	ret = file_write_and_wait_range(dst_file, dst_off, (dst_off + len));
2360 	if (ret < 0) {
2361 		dout("failed to write dst file (%zd)\n", ret);
2362 		goto out;
2363 	}
2364 
2365 	/*
2366 	 * We need FILE_WR caps for dst_ci and FILE_RD for src_ci as other
2367 	 * clients may have dirty data in their caches.  And OSDs know nothing
2368 	 * about caps, so they can't safely do the remote object copies.
2369 	 */
2370 	err = get_rd_wr_caps(src_file, &src_got,
2371 			     dst_file, (dst_off + len), &dst_got);
2372 	if (err < 0) {
2373 		dout("get_rd_wr_caps returned %d\n", err);
2374 		ret = -EOPNOTSUPP;
2375 		goto out;
2376 	}
2377 
2378 	ret = is_file_size_ok(src_inode, dst_inode, src_off, dst_off, len);
2379 	if (ret < 0)
2380 		goto out_caps;
2381 
2382 	/* Drop dst file cached pages */
2383 	ret = invalidate_inode_pages2_range(dst_inode->i_mapping,
2384 					    dst_off >> PAGE_SHIFT,
2385 					    (dst_off + len) >> PAGE_SHIFT);
2386 	if (ret < 0) {
2387 		dout("Failed to invalidate inode pages (%zd)\n", ret);
2388 		ret = 0; /* XXX */
2389 	}
2390 	ceph_calc_file_object_mapping(&src_ci->i_layout, src_off,
2391 				      src_ci->i_layout.object_size,
2392 				      &src_objnum, &src_objoff, &src_objlen);
2393 	ceph_calc_file_object_mapping(&dst_ci->i_layout, dst_off,
2394 				      dst_ci->i_layout.object_size,
2395 				      &dst_objnum, &dst_objoff, &dst_objlen);
2396 	/* object-level offsets need to the same */
2397 	if (src_objoff != dst_objoff) {
2398 		ret = -EOPNOTSUPP;
2399 		goto out_caps;
2400 	}
2401 
2402 	/*
2403 	 * Do a manual copy if the object offset isn't object aligned.
2404 	 * 'src_objlen' contains the bytes left until the end of the object,
2405 	 * starting at the src_off
2406 	 */
2407 	if (src_objoff) {
2408 		dout("Initial partial copy of %u bytes\n", src_objlen);
2409 
2410 		/*
2411 		 * we need to temporarily drop all caps as we'll be calling
2412 		 * {read,write}_iter, which will get caps again.
2413 		 */
2414 		put_rd_wr_caps(src_ci, src_got, dst_ci, dst_got);
2415 		ret = do_splice_direct(src_file, &src_off, dst_file,
2416 				       &dst_off, src_objlen, flags);
2417 		/* Abort on short copies or on error */
2418 		if (ret < src_objlen) {
2419 			dout("Failed partial copy (%zd)\n", ret);
2420 			goto out;
2421 		}
2422 		len -= ret;
2423 		err = get_rd_wr_caps(src_file, &src_got,
2424 				     dst_file, (dst_off + len), &dst_got);
2425 		if (err < 0)
2426 			goto out;
2427 		err = is_file_size_ok(src_inode, dst_inode,
2428 				      src_off, dst_off, len);
2429 		if (err < 0)
2430 			goto out_caps;
2431 	}
2432 
2433 	size = i_size_read(dst_inode);
2434 	bytes = ceph_do_objects_copy(src_ci, &src_off, dst_ci, &dst_off,
2435 				     src_fsc, len, flags);
2436 	if (bytes <= 0) {
2437 		if (!ret)
2438 			ret = bytes;
2439 		goto out_caps;
2440 	}
2441 	dout("Copied %zu bytes out of %zu\n", bytes, len);
2442 	len -= bytes;
2443 	ret += bytes;
2444 
2445 	file_update_time(dst_file);
2446 	inode_inc_iversion_raw(dst_inode);
2447 
2448 	if (dst_off > size) {
2449 		/* Let the MDS know about dst file size change */
2450 		if (ceph_inode_set_size(dst_inode, dst_off) ||
2451 		    ceph_quota_is_max_bytes_approaching(dst_inode, dst_off))
2452 			ceph_check_caps(dst_ci, CHECK_CAPS_AUTHONLY, NULL);
2453 	}
2454 	/* Mark Fw dirty */
2455 	spin_lock(&dst_ci->i_ceph_lock);
2456 	dst_ci->i_inline_version = CEPH_INLINE_NONE;
2457 	dirty = __ceph_mark_dirty_caps(dst_ci, CEPH_CAP_FILE_WR, &prealloc_cf);
2458 	spin_unlock(&dst_ci->i_ceph_lock);
2459 	if (dirty)
2460 		__mark_inode_dirty(dst_inode, dirty);
2461 
2462 out_caps:
2463 	put_rd_wr_caps(src_ci, src_got, dst_ci, dst_got);
2464 
2465 	/*
2466 	 * Do the final manual copy if we still have some bytes left, unless
2467 	 * there were errors in remote object copies (len >= object_size).
2468 	 */
2469 	if (len && (len < src_ci->i_layout.object_size)) {
2470 		dout("Final partial copy of %zu bytes\n", len);
2471 		bytes = do_splice_direct(src_file, &src_off, dst_file,
2472 					 &dst_off, len, flags);
2473 		if (bytes > 0)
2474 			ret += bytes;
2475 		else
2476 			dout("Failed partial copy (%zd)\n", bytes);
2477 	}
2478 
2479 out:
2480 	ceph_free_cap_flush(prealloc_cf);
2481 
2482 	return ret;
2483 }
2484 
2485 static ssize_t ceph_copy_file_range(struct file *src_file, loff_t src_off,
2486 				    struct file *dst_file, loff_t dst_off,
2487 				    size_t len, unsigned int flags)
2488 {
2489 	ssize_t ret;
2490 
2491 	ret = __ceph_copy_file_range(src_file, src_off, dst_file, dst_off,
2492 				     len, flags);
2493 
2494 	if (ret == -EOPNOTSUPP || ret == -EXDEV)
2495 		ret = generic_copy_file_range(src_file, src_off, dst_file,
2496 					      dst_off, len, flags);
2497 	return ret;
2498 }
2499 
2500 const struct file_operations ceph_file_fops = {
2501 	.open = ceph_open,
2502 	.release = ceph_release,
2503 	.llseek = ceph_llseek,
2504 	.read_iter = ceph_read_iter,
2505 	.write_iter = ceph_write_iter,
2506 	.mmap = ceph_mmap,
2507 	.fsync = ceph_fsync,
2508 	.lock = ceph_lock,
2509 	.flock = ceph_flock,
2510 	.splice_read = generic_file_splice_read,
2511 	.splice_write = iter_file_splice_write,
2512 	.unlocked_ioctl = ceph_ioctl,
2513 	.compat_ioctl = compat_ptr_ioctl,
2514 	.fallocate	= ceph_fallocate,
2515 	.copy_file_range = ceph_copy_file_range,
2516 };
2517