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(i: iter))
92	maxsize = iov_iter_count(i: 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(i: iter, pages, maxsize: maxsize - size,
101	ITER_GET_BVECS_PAGES, start: &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(bv: &bvecs[bvec_idx], page: pages[idx], len, offset: 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(i: iter);
132	ssize_t bytes;
133	int npages;
134
135	iov_iter_truncate(i: iter, count: maxsize);
136	npages = iov_iter_npages(i: iter, INT_MAX);
137	iov_iter_reexpand(i: iter, count: orig_count);
138
139	/*
140	* __iter_get_bvecs() may populate only part of the array -- zero it
141	* out.
142	*/
143	bv = kvmalloc_array(n: npages, size: sizeof(*bv), GFP_KERNEL | __GFP_ZERO);
144	if (!bv)
145	return -ENOMEM;
146
147	bytes = __iter_get_bvecs(iter, maxsize, bvecs: bv);
148	if (bytes < 0) {
149	/*
150	* No pages were pinned -- just free the array.
151	*/
152	kvfree(addr: 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(page: bvecs[i].bv_page);
170	}
171	}
172	kvfree(addr: 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, mode: want_auth);
191	if (IS_ERR(ptr: 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(inode: &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(k: 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(k: 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, mode: fmode, count: 1);
236	fi->fmode = fmode;
237
238	spin_lock_init(&fi->rw_contexts_lock);
239	INIT_LIST_HEAD(list: &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, update: file->f_mode & FMODE_WRITE);
252	ceph_put_fmode(ci, mode: fi->fmode, count: 1);
253	kmem_cache_free(s: ceph_file_cachep, objp: 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, will_modify: 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(sb: 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(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(lock: &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, flags: 0);
319	return 0;
320	}
321	spin_unlock(lock: &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(sb: inode->i_sb, flags, create_mode: 0);
336	if (IS_ERR(ptr: req)) {
337	err = PTR_ERR(ptr: 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(sb: 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, filp: 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(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(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, check: 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(lock: &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, flags: 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(lock: &ci->i_ceph_lock);
427	return ceph_init_file(inode, file, fmode);
428	}
429
430	spin_unlock(lock: &ci->i_ceph_lock);
431
432	doutc(cl, "open fmode %d wants %s\n", fmode, ceph_cap_string(wanted));
433	req = prepare_open_request(sb: inode->i_sb, flags, create_mode: 0);
434	if (IS_ERR(ptr: req)) {
435	err = PTR_ERR(ptr: 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, fmode: 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(inode: dst);
456	struct ceph_inode_info *csrc = ceph_inode(inode: src);
457
458	spin_lock(lock: &cdst->i_ceph_lock);
459	if ((__ceph_caps_issued(ci: cdst, NULL) & CEPH_CAP_DIR_CREATE) &&
460	!ceph_file_layout_is_valid(layout: &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(lock: &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(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(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(xa: &ci->i_auth_cap->session->s_delegated_inos))
490	goto no_async;
491
492	if (!ceph_file_layout_is_valid(layout: &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(lock: &dentry->d_lock);
502	di->lease_shared_gen = atomic_read(v: &ci->i_shared_gen);
503	spin_unlock(lock: &dentry->d_lock);
504	} else if (atomic_read(v: &ci->i_shared_gen) !=
505	READ_ONCE(di->lease_shared_gen)) {
506	goto no_async;
507	}
508
509	ino = ceph_get_deleg_ino(session: ci->i_auth_cap->session);
510	if (!ino)
511	goto no_async;
512
513	*pino = ino;
514	ceph_take_cap_refs(ci, caps: want, snap_rwsem_locked: 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(lock: &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(inode: dir);
527	struct ceph_inode_info *ci = ceph_inode(inode: dir);
528	struct ceph_mds_session *s = NULL;
529
530	spin_lock(lock: &ci->i_ceph_lock);
531	if (ci->i_auth_cap)
532	s = ceph_get_mds_session(s: ci->i_auth_cap->session);
533	spin_unlock(lock: &ci->i_ceph_lock);
534	if (s) {
535	int err = ceph_restore_deleg_ino(session: 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(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(word: &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(lock: &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(mapping: req->r_parent->i_mapping, error: result);
584
585	if (result) {
586	int pathlen = 0;
587	u64 base = 0;
588	char *path = ceph_mdsc_build_path(mdsc, dentry: req->r_dentry, plen: &pathlen,
589	base: &base, for_wire: 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, len: pathlen);
595
596	ceph_dir_clear_complete(inode: req->r_parent);
597	if (!d_unhashed(dentry))
598	d_drop(dentry);
599
600	if (dinode) {
601	mapping_set_error(mapping: dinode->i_mapping, error: result);
602	ceph_inode_shutdown(inode: dinode);
603	wake_async_create_waiters(inode: dinode, session: req->r_session);
604	}
605	}
606
607	if (tinode) {
608	u64 ino = ceph_vino(inode: 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(mapping: tinode->i_mapping, error: result);
616	wake_async_create_waiters(inode: tinode, session: 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(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(sb: 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(tv: &now);
646
647	iinfo.inline_version = CEPH_INLINE_NONE;
648	iinfo.change_attr = 1;
649	ceph_encode_timespec64(tv: &iinfo.btime, ts: &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(fl: lo, legacy: &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(sem: &mdsc->snap_rwsem);
700	ret = ceph_fill_inode(inode, NULL, iinfo: &iinfo, NULL, session: req->r_session,
701	cap_fmode: req->r_fmode, NULL);
702	up_read(sem: &mdsc->snap_rwsem);
703	if (ret) {
704	doutc(cl, "failed to fill inode: %d\n", ret);
705	ceph_dir_clear_complete(inode: 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(inode: 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, open: ceph_open);
734	}
735
736	spin_lock(lock: &dentry->d_lock);
737	di->flags &= ~CEPH_DENTRY_ASYNC_CREATE;
738	wake_up_bit(word: &di->flags, CEPH_DENTRY_ASYNC_CREATE_BIT);
739	spin_unlock(lock: &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(sb: 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(inode: dir))
782	return -EDQUOT;
783
784	new_inode = ceph_new_inode(dir, dentry, mode: &mode, as_ctx: &as_ctx);
785	if (IS_ERR(ptr: new_inode)) {
786	err = PTR_ERR(ptr: 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(head: &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(sb: dir->i_sb, flags, create_mode: mode);
800	if (IS_ERR(ptr: req)) {
801	err = PTR_ERR(ptr: 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(in: 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(inode: dir);
814	if (IS_ENCRYPTED(dir)) {
815	set_bit(CEPH_MDS_R_FSCRYPT_FILE, addr: &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: &as_ctx);
829
830	if (try_async && (req->r_dir_caps =
831	try_prep_async_create(dir, dentry, lo: &lo,
832	pino: &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, addr: &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(sb: dir->i_sb, vino, newino: new_inode);
843	if (IS_ERR(ptr: new_inode)) {
844	err = PTR_ERR(ptr: 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(lock: &dentry->d_lock);
851	di->flags |= CEPH_DENTRY_ASYNC_CREATE;
852	spin_unlock(lock: &dentry->d_lock);
853
854	err = ceph_mdsc_submit_request(mdsc, dir, req);
855	if (!err) {
856	err = ceph_finish_async_create(dir, inode: new_inode,
857	dentry, file,
858	mode, req,
859	as_ctx: &as_ctx, lo: &lo);
860	new_inode = NULL;
861	} else if (err == -EJUKEBOX) {
862	restore_deleg_ino(dir, ino: req->r_deleg_ino);
863	ceph_mdsc_put_request(req);
864	discard_new_inode(new_inode);
865	ceph_release_acl_sec_ctx(as_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, addr: &req->r_req_flags);
878	req->r_new_inode = new_inode;
879	new_inode = NULL;
880	err = ceph_mdsc_do_request(mdsc, dir: (flags & O_CREAT) ? dir : NULL, req);
881	if (err == -ENOENT) {
882	dentry = ceph_handle_snapdir(req, dentry);
883	if (IS_ERR(ptr: dentry)) {
884	err = PTR_ERR(ptr: 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(ptr: dn))
896	err = PTR_ERR(ptr: 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, dentry: dn);
907	} else {
908	if (IS_ENCRYPTED(dir) &&
909	!fscrypt_has_permitted_context(parent: dir, child: 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(dst: dir, src: newino);
921	ceph_init_inode_acls(inode: newino, as_ctx: &as_ctx);
922	file->f_mode |= FMODE_CREATED;
923	}
924	err = finish_open(file, dentry, open: 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: &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, mode: dfi->file_info.fmode, count: 1);
947
948	if (dfi->last_readdir)
949	ceph_mdsc_put_request(req: dfi->last_readdir);
950	kfree(objp: dfi->last_name);
951	kfree(objp: dfi->dir_info);
952	kmem_cache_free(s: ceph_dir_file_cachep, objp: 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, update: file->f_mode & FMODE_WRITE);
960	ceph_put_fmode(ci, mode: fi->fmode, count: 1);
961
962	kmem_cache_free(s: ceph_file_cachep, objp: 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(i: 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(mapping: inode->i_mapping,
1016	lstart: off, lend: off + len - 1);
1017	if (ret < 0)
1018	return ret;
1019
1020	ret = 0;
1021	while ((len = iov_iter_count(i: 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, off: &read_off, len: &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, layout: &ci->i_layout,
1041	vino: ci->i_vino, offset: read_off, len: &read_len, which: 0, num_ops: 1,
1042	opcode: sparse ? CEPH_OSD_OP_SPARSE_READ :
1043	CEPH_OSD_OP_READ,
1044	flags: CEPH_OSD_FLAG_READ,
1045	NULL, truncate_seq: ci->i_truncate_seq,
1046	truncate_size: ci->i_truncate_size, use_mempool: false);
1047	if (IS_ERR(ptr: req)) {
1048	ret = PTR_ERR(ptr: 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(i: to);
1056
1057	num_pages = calc_pages_for(off: read_off, len: read_len);
1058	page_off = offset_in_page(off);
1059	pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1060	if (IS_ERR(ptr: pages)) {
1061	ceph_osdc_put_request(req);
1062	ret = PTR_ERR(ptr: pages);
1063	break;
1064	}
1065
1066	osd_req_op_extent_osd_data_pages(req, which: 0, pages, length: read_len,
1067	offset_in_page(read_off),
1068	pages_from_pool: false, own_pages: false);
1069
1070	op = &req->r_ops[0];
1071	if (sparse) {
1072	extent_cnt = __ceph_sparse_read_ext_count(inode, len: read_len);
1073	ret = ceph_alloc_sparse_ext_map(op, cnt: 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(m: &fsc->mdsc->metric,
1084	r_start: req->r_start_latency,
1085	r_end: req->r_end_latency,
1086	size: read_len, rc: 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, page: pages,
1105	off: read_off, map: op->extent.sparse_ext,
1106	ext_cnt: 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(off: zoff, len: 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(page: pages[idx++],
1153	offset: page_off, bytes: plen, i: 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(f: 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, ki_pos: &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(v: &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, size: endoff))
1250	ceph_check_caps(ci, CHECK_CAPS_AUTHONLY);
1251	}
1252
1253	spin_lock(lock: &ci->i_ceph_lock);
1254	dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
1255	pcf: &aio_req->prealloc_cf);
1256	spin_unlock(lock: &ci->i_ceph_lock);
1257	if (dirty)
1258	__mark_inode_dirty(inode, dirty);
1259
1260	}
1261
1262	ceph_put_cap_refs(ci, had: (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(cf: aio_req->prealloc_cf);
1268	kfree(objp: 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(osd_req: req, which: 0);
1277	struct ceph_osd_req_op *op = &req->r_ops[0];
1278	struct ceph_client_metric *metric = &ceph_sb_to_mdsc(sb: 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(size: 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(wq: ceph_inode_to_fs_client(inode)->inode_wq,
1298	work: &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: &i, ITER_DEST, bvec: osd_data->bvec_pos.bvecs,
1326	nr_segs: osd_data->num_bvecs, count: len);
1327	iov_iter_advance(i: &i, bytes: rc);
1328	iov_iter_zero(bytes: 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(m: metric, r_start: req->r_start_latency,
1336	r_end: req->r_end_latency, size: len, rc);
1337	else
1338	ceph_update_read_metrics(m: metric, r_start: req->r_start_latency,
1339	r_end: req->r_end_latency, size: len, rc);
1340	}
1341
1342	put_bvecs(bvecs: osd_data->bvec_pos.bvecs, num_bvecs: osd_data->num_bvecs,
1343	should_dirty: 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(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(sc: capsnap->context);
1372	} else {
1373	BUG_ON(!ci->i_head_snapc);
1374	snapc = ceph_get_snap_context(sc: ci->i_head_snapc);
1375	}
1376	spin_unlock(lock: &ci->i_ceph_lock);
1377
1378	req = ceph_osdc_alloc_request(osdc: orig_req->r_osdc, snapc, num_ops: 1,
1379	use_mempool: 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(dest: &req->r_base_oloc, src: &orig_req->r_base_oloc);
1388	ceph_oid_copy(dest: &req->r_base_oid, src: &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(req: 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(osdc: 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(sc: snapc);
1416	kfree(objp: 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(f: 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(i: iter);
1439	loff_t pos = iocb->ki_pos;
1440	bool write = iov_iter_rw(i: iter) == WRITE;
1441	bool should_dirty = !write && user_backed_iter(i: iter);
1442	bool sparse = ceph_test_mount_opt(fsc, SPARSEREAD);
1443
1444	if (write && ceph_snap(inode: file_inode(f: 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, dio_write: true);
1455
1456	ret2 = invalidate_inode_pages2_range(mapping: inode->i_mapping,
1457	start: pos >> PAGE_SHIFT,
1458	end: (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(i: iter) > 0) {
1469	u64 size = iov_iter_count(i: 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, layout: &ci->i_layout,
1482	vino, offset: pos, len: &size, which: 0,
1483	num_ops: 1,
1484	opcode: write ? CEPH_OSD_OP_WRITE : readop,
1485	flags, snapc,
1486	truncate_seq: ci->i_truncate_seq,
1487	truncate_size: ci->i_truncate_size,
1488	use_mempool: false);
1489	if (IS_ERR(ptr: req)) {
1490	ret = PTR_ERR(ptr: req);
1491	break;
1492	}
1493
1494	len = iter_get_bvecs_alloc(iter, maxsize: size, bvecs: &bvecs, num_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(osd_req: req, which: 0, length: 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(kiocb: iocb) &&
1508	(len == count || pos + count <= i_size_read(inode))) {
1509	aio_req = kzalloc(size: 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(list: &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, lstart: pos,
1529	PAGE_ALIGN(pos + len) - 1);
1530
1531	req->r_mtime = mtime;
1532	}
1533
1534	osd_req_op_extent_osd_data_bvecs(osd_req: req, which: 0, bvecs, num_bvecs: num_pages, bytes: len);
1535	op = &req->r_ops[0];
1536	if (sparse) {
1537	extent_cnt = __ceph_sparse_read_ext_count(inode, len: size);
1538	ret = ceph_alloc_sparse_ext_map(op, cnt: 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(v: &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(new: &req->r_private_item, head: &aio_req->osd_reqs);
1554
1555	pos += len;
1556	continue;
1557	}
1558
1559	ceph_osdc_start_request(osdc: req->r_osdc, req);
1560	ret = ceph_osdc_wait_request(osdc: &fsc->client->osdc, req);
1561
1562	if (write)
1563	ceph_update_write_metrics(m: metric, r_start: req->r_start_latency,
1564	r_end: req->r_end_latency, size: len, rc: ret);
1565	else
1566	ceph_update_read_metrics(m: metric, r_start: req->r_start_latency,
1567	r_end: req->r_end_latency, size: len, rc: 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: &i, ITER_DEST, bvec: bvecs, nr_segs: num_pages, count: len);
1582	iov_iter_advance(i: &i, bytes: ret);
1583	iov_iter_zero(bytes: zlen, &i);
1584	ret += zlen;
1585	}
1586	if (ret >= 0)
1587	len = ret;
1588	}
1589
1590	put_bvecs(bvecs, num_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, size: pos))
1601	ceph_check_caps(ci: 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(objp: aio_req);
1611	return ret;
1612	}
1613
1614	ceph_get_cap_refs(ci, caps: write ? CEPH_CAP_FILE_WR :
1615	CEPH_CAP_FILE_RD);
1616
1617	list_splice(list: &aio_req->osd_reqs, head: &osd_reqs);
1618	inode_dio_begin(inode);
1619	while (!list_empty(head: &osd_reqs)) {
1620	req = list_first_entry(&osd_reqs,
1621	struct ceph_osd_request,
1622	r_private_item);
1623	list_del_init(entry: &req->r_private_item);
1624	if (ret >= 0)
1625	ceph_osdc_start_request(osdc: 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(f: 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(i: from);
1667
1668	if (ceph_snap(inode: file_inode(f: 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(mapping: inode->i_mapping,
1675	lstart: pos, lend: pos + count - 1);
1676	if (ret < 0)
1677	return ret;
1678
1679	ceph_fscache_invalidate(inode, dio_write: false);
1680
1681	while ((len = iov_iter_count(i: 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, off: &write_pos, len: &write_len);
1695
1696	/* clamp the length to the end of first object */
1697	ceph_calc_file_object_mapping(l: &ci->i_layout, off: write_pos,
1698	len: write_len, objno: &objnum, objoff: &objoff,
1699	xlen: &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(off: write_pos, len: write_len);
1725	pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1726	if (IS_ERR(ptr: pages)) {
1727	ret = PTR_ERR(ptr: 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, layout: &ci->i_layout,
1750	vino: ci->i_vino, offset: first ? first_pos : last_pos,
1751	len: &read_len, which: 0, num_ops: (first && last) ? 2 : 1,
1752	opcode: CEPH_OSD_OP_SPARSE_READ, flags: CEPH_OSD_FLAG_READ,
1753	NULL, truncate_seq: ci->i_truncate_seq,
1754	truncate_size: ci->i_truncate_size, use_mempool: false);
1755	if (IS_ERR(ptr: req)) {
1756	ceph_release_page_vector(pages, num_pages);
1757	ret = PTR_ERR(ptr: 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, which: 0, pages,
1774	CEPH_FSCRYPT_BLOCK_SIZE,
1775	offset_in_page(first_pos),
1776	pages_from_pool: false, own_pages: false);
1777	/* We only expect a single extent here */
1778	ret = __ceph_alloc_sparse_ext_map(op, cnt: 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(osd_req: req, which: 1,
1792	opcode: CEPH_OSD_OP_SPARSE_READ,
1793	offset: last_pos, CEPH_FSCRYPT_BLOCK_SIZE,
1794	truncate_size: ci->i_truncate_size,
1795	truncate_seq: ci->i_truncate_seq);
1796	}
1797
1798	ret = __ceph_alloc_sparse_ext_map(op, cnt: 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, which: first ? 1 : 0,
1806	pages: &pages[num_pages - 1],
1807	CEPH_FSCRYPT_BLOCK_SIZE,
1808	offset_in_page(last_pos),
1809	pages_from_pool: false, own_pages: 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(m: &fsc->mdsc->metric,
1817	r_start: req->r_start_latency,
1818	r_end: req->r_end_latency,
1819	size: read_len, rc: 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(page: pages[0], start: 0,
1837	offset_in_page(first_pos));
1838	if (last)
1839	zero_user_segment(page: 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(page: pages[0], start: 0,
1853	offset_in_page(first_pos));
1854	else
1855	zero_user_segment(page: 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(page: 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	page: pages[0], CEPH_FSCRYPT_BLOCK_SIZE,
1891	offset_in_page(first_pos),
1892	lblk_num: 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	page: pages[num_pages - 1],
1901	CEPH_FSCRYPT_BLOCK_SIZE,
1902	offset_in_page(last_pos),
1903	lblk_num: 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(page: pages[n], offset: off, bytes: plen, i: 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, page: pages,
1934	off: write_pos, len: 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, layout: &ci->i_layout,
1944	vino: ci->i_vino, offset: write_pos, len: &write_len,
1945	which: rmw ? 1 : 0, num_ops: rmw ? 2 : 1,
1946	opcode: CEPH_OSD_OP_WRITE,
1947	flags: CEPH_OSD_FLAG_WRITE,
1948	snapc, truncate_seq: ci->i_truncate_seq,
1949	truncate_size: ci->i_truncate_size, use_mempool: false);
1950	if (IS_ERR(ptr: req)) {
1951	ret = PTR_ERR(ptr: 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, which: rmw ? 1 : 0, pages, length: write_len,
1958	offset_in_page(write_pos), pages_from_pool: false,
1959	own_pages: 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(osd_req: req, which: 0, opcode: CEPH_OSD_OP_ASSERT_VER, flags: 0);
1973	req->r_ops[0].assert_ver.ver = assert_ver;
1974	} else {
1975	osd_req_op_init(osd_req: req, which: 0, opcode: CEPH_OSD_OP_CREATE,
1976	flags: 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(m: &fsc->mdsc->metric, r_start: req->r_start_latency,
1984	r_end: req->r_end_latency, size: len, rc: 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	mapping: inode->i_mapping,
2014	start: pos >> PAGE_SHIFT,
2015	end: (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, size: pos);
2026	if (check_caps)
2027	ceph_check_caps(ci: 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(i: to);
2053	struct inode *inode = file_inode(f: 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, endoff: -1, got: &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, iter: 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: &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(cf: fi, ctx: &rw_ctx);
2115	ret = generic_file_read_iter(iocb, to);
2116	ceph_del_rw_context(cf: fi, ctx: &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, had: 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, locked_page: page, mask, force: !!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, start: statret, end);
2162	ret = copy_page_to_iter(page,
2163	offset: iocb->ki_pos & ~PAGE_MASK,
2164	bytes: end - iocb->ki_pos, i: 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(bytes: zlen, to);
2172	iocb->ki_pos += ret;
2173	read += ret;
2174	}
2175	__free_pages(page, order: 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(f: 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(filp: in, CEPH_CAP_FILE_RD, want, endoff: -1, got: &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, had: 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(cf: fi, ctx: &rw_ctx);
2249	ret = filemap_splice_read(in, ppos, pipe, len, flags);
2250	ceph_del_rw_context(cf: fi, ctx: &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, had: 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(f: 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, force: 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(i: from, count: limit - pos);
2324	}
2325
2326	count = iov_iter_count(i: from);
2327	if (ceph_quota_is_max_bytes_exceeded(inode, newlen: pos + count)) {
2328	err = -EDQUOT;
2329	goto out;
2330	}
2331
2332	down_read(sem: &osdc->lock);
2333	map_flags = osdc->osdmap->flags;
2334	pool_flags = ceph_pg_pool_flags(map: osdc->osdmap, id: ci->i_layout.pool_id);
2335	up_read(sem: &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(filp: file, CEPH_CAP_FILE_WR, want, endoff: pos + count, got: &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(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(sc: capsnap->context);
2380	} else {
2381	BUG_ON(!ci->i_head_snapc);
2382	snapc = ceph_get_snap_context(sc: ci->i_head_snapc);
2383	}
2384	spin_unlock(lock: &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, iter: &data, snapc,
2390	pcf: &prealloc_cf);
2391	else
2392	written = ceph_sync_write(iocb, from: &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(i: from, bytes: written);
2399	ceph_put_snap_context(sc: 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(lock: &ci->i_ceph_lock);
2416	dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
2417	pcf: &prealloc_cf);
2418	spin_unlock(lock: &ci->i_ceph_lock);
2419	if (dirty)
2420	__mark_inode_dirty(inode, dirty);
2421	if (ceph_quota_is_max_bytes_approaching(inode, newlen: 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, had: 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, count: written);
2441	}
2442
2443	goto out_unlocked;
2444	out_caps:
2445	ceph_put_cap_refs(ci, had: 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(cf: 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(f: file);
2463	int ret;
2464
2465	ret = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE, force: 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(mapping: inode->i_mapping, index);
2479	if (page) {
2480	wait_on_page_writeback(page);
2481	zero_user(page, start: 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, end: offset + size - 1);
2502	offset += size;
2503	length -= size;
2504	}
2505	if (length)
2506	ceph_zero_partial_page(inode, offset, size: 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, layout: &ci->i_layout,
2530	vino: ceph_vino(inode),
2531	offset, len: length,
2532	which: 0, num_ops: 1, opcode: op,
2533	flags: CEPH_OSD_FLAG_WRITE,
2534	NULL, truncate_seq: 0, truncate_size: 0, use_mempool: false);
2535	if (IS_ERR(ptr: req)) {
2536	ret = PTR_ERR(ptr: req);
2537	goto out;
2538	}
2539
2540	req->r_mtime = inode_get_mtime(inode);
2541	ceph_osdc_start_request(osdc: &fsc->client->osdc, req);
2542	ret = ceph_osdc_wait_request(osdc: &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, length: &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, offset: 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, length: &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(f: 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(filp: file, CEPH_CAP_FILE_WR, want, endoff, got: &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(mapping: inode->i_mapping);
2656	ceph_fscache_invalidate(inode, dio_write: false);
2657	ceph_zero_pagecache_range(inode, offset, length);
2658	ret = ceph_zero_objects(inode, offset, length);
2659
2660	if (!ret) {
2661	spin_lock(lock: &ci->i_ceph_lock);
2662	dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
2663	pcf: &prealloc_cf);
2664	spin_unlock(lock: &ci->i_ceph_lock);
2665	if (dirty)
2666	__mark_inode_dirty(inode, dirty);
2667	}
2668	filemap_invalidate_unlock(mapping: inode->i_mapping);
2669
2670	put_caps:
2671	ceph_put_cap_refs(ci, had: got);
2672	unlock:
2673	inode_unlock(inode);
2674	ceph_free_cap_flush(cf: 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(filp: dst_filp, CEPH_CAP_FILE_WR, CEPH_CAP_FILE_BUFFER,
2692	endoff: dst_endoff, got: 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(inode: file_inode(f: src_filp),
2702	CEPH_CAP_FILE_RD, CEPH_CAP_FILE_SHARED,
2703	nonblock: false, got: src_got);
2704	if (ret <= 0) {
2705	/* Start by dropping dst_ci caps and getting src_ci caps */
2706	ceph_put_cap_refs(ci: ceph_inode(inode: file_inode(f: dst_filp)), had: *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(filp: src_filp, CEPH_CAP_FILE_RD,
2714	CEPH_CAP_FILE_SHARED, endoff: -1, got: src_got);
2715	if (ret < 0)
2716	return ret;
2717	/*... drop src_ci caps too, and retry */
2718	ceph_put_cap_refs(ci: ceph_inode(inode: file_inode(f: src_filp)), had: *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(ci: src_ci, had: src_got);
2729	ceph_put_cap_refs(ci: dst_ci, had: 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(inode: src_inode);
2742	loff_t size, endoff;
2743
2744	size = i_size_read(inode: 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(inode: dst_inode);
2757
2758	endoff = dst_off + len;
2759	if (inode_newsize_ok(dst_inode, offset: endoff))
2760	return -EOPNOTSUPP;
2761
2762	if (ceph_quota_is_max_bytes_exceeded(inode: dst_inode, newlen: 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, num_ops: 1, use_mempool: false, GFP_KERNEL);
2787	if (!req)
2788	return ERR_PTR(error: -ENOMEM);
2789
2790	req->r_flags = CEPH_OSD_FLAG_WRITE;
2791
2792	ceph_oloc_copy(dest: &req->r_t.base_oloc, src: dst_oloc);
2793	ceph_oid_copy(dest: &req->r_t.base_oid, src: dst_oid);
2794
2795	ret = osd_req_op_copy_from_init(req, src_snapid, src_version: 0,
2796	src_oid, src_oloc,
2797	src_fadvise_flags,
2798	dst_fadvise_flags,
2799	truncate_seq,
2800	truncate_size,
2801	copy_from_flags: 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(error: 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(l: &src_ci->i_layout, off: *src_off,
2840	len: object_size, objno: &src_objnum,
2841	objoff: &src_objoff, xlen: &src_objlen);
2842	ceph_calc_file_object_mapping(l: &dst_ci->i_layout, off: *dst_off,
2843	len: object_size, objno: &dst_objnum,
2844	objoff: &dst_objoff, xlen: &dst_objlen);
2845	ceph_oid_init(oid: &src_oid);
2846	ceph_oid_printf(oid: &src_oid, fmt: "%llx.%08llx",
2847	src_ci->i_vino.ino, src_objnum);
2848	ceph_oid_init(oid: &dst_oid);
2849	ceph_oid_printf(oid: &dst_oid, fmt: "%llx.%08llx",
2850	dst_ci->i_vino.ino, dst_objnum);
2851	/* Do an object remote copy */
2852	req = ceph_alloc_copyfrom_request(osdc, src_snapid: src_ci->i_vino.snap,
2853	src_oid: &src_oid, src_oloc: &src_oloc,
2854	dst_oid: &dst_oid, dst_oloc: &dst_oloc,
2855	truncate_seq: dst_ci->i_truncate_seq,
2856	truncate_size: dst_ci->i_truncate_size);
2857	if (IS_ERR(ptr: req))
2858	ret = PTR_ERR(ptr: req);
2859	else {
2860	ceph_osdc_start_request(osdc, req);
2861	ret = ceph_osdc_wait_request(osdc, req);
2862	ceph_update_copyfrom_metrics(m: &fsc->mdsc->metric,
2863	r_start: req->r_start_latency,
2864	r_end: req->r_end_latency,
2865	size: object_size, rc: 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(oloc: &src_oloc);
2887	ceph_oloc_destroy(oloc: &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(f: src_file);
2896	struct inode *dst_inode = file_inode(f: dst_file);
2897	struct ceph_inode_info *src_ci = ceph_inode(inode: src_inode);
2898	struct ceph_inode_info *dst_ci = ceph_inode(inode: dst_inode);
2899	struct ceph_cap_flush *prealloc_cf;
2900	struct ceph_fs_client *src_fsc = ceph_inode_to_fs_client(inode: 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(inode: dst_inode);
2910
2911	if (ceph_fsid_compare(a: &src_fsc->client->fsid,
2912	b: &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(inode: 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(file: src_file, start: src_off, end: (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(file: dst_file, start: dst_off, end: (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_filp: src_file, src_got: &src_got,
2977	dst_filp: dst_file, dst_endoff: (dst_off + len), dst_got: &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(inode: dst_inode, dio_write: false);
2990	ret = invalidate_inode_pages2_range(mapping: dst_inode->i_mapping,
2991	start: dst_off >> PAGE_SHIFT,
2992	end: (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(l: &src_ci->i_layout, off: src_off,
2999	len: src_ci->i_layout.object_size,
3000	objno: &src_objnum, objoff: &src_objoff, xlen: &src_objlen);
3001	ceph_calc_file_object_mapping(l: &dst_ci->i_layout, off: dst_off,
3002	len: dst_ci->i_layout.object_size,
3003	objno: &dst_objnum, objoff: &dst_objoff, xlen: &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(in: src_file, ppos: &src_off, out: dst_file, opos: &dst_off,
3024	len: 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_filp: src_file, src_got: &src_got,
3032	dst_filp: dst_file, dst_endoff: (dst_off + len), dst_got: &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(inode: dst_inode);
3042	bytes = ceph_do_objects_copy(src_ci, src_off: &src_off, dst_ci, dst_off: &dst_off,
3043	fsc: 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(file: dst_file);
3054	inode_inc_iversion_raw(inode: dst_inode);
3055
3056	if (dst_off > size) {
3057	/* Let the MDS know about dst file size change */
3058	if (ceph_inode_set_size(inode: dst_inode, size: dst_off) ||
3059	ceph_quota_is_max_bytes_approaching(inode: dst_inode, newlen: dst_off))
3060	ceph_check_caps(ci: dst_ci, CHECK_CAPS_AUTHONLY | CHECK_CAPS_FLUSH);
3061	}
3062	/* Mark Fw dirty */
3063	spin_lock(lock: &dst_ci->i_ceph_lock);
3064	dirty = __ceph_mark_dirty_caps(ci: dst_ci, CEPH_CAP_FILE_WR, pcf: &prealloc_cf);
3065	spin_unlock(lock: &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(in: src_file, ppos: &src_off, out: dst_file,
3079	opos: &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(cf: 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(in: src_file, pos_in: src_off, out: dst_file,
3103	pos_out: 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