linux - fs/ceph/addr.c

2658 lines · 71.3 KiB

1	// SPDX-License-Identifier: GPL-2.0
2	#include <linux/ceph/ceph_debug.h>
3
4	#include <linux/backing-dev.h>
5	#include <linux/fs.h>
6	#include <linux/mm.h>
7	#include <linux/swap.h>
8	#include <linux/pagemap.h>
9	#include <linux/slab.h>
10	#include <linux/folio_batch.h>
11	#include <linux/task_io_accounting_ops.h>
12	#include <linux/signal.h>
13	#include <linux/iversion.h>
14	#include <linux/ktime.h>
15	#include <linux/netfs.h>
16	#include <trace/events/netfs.h>
17
18	#include "super.h"
19	#include "mds_client.h"
20	#include "cache.h"
21	#include "metric.h"
22	#include "subvolume_metrics.h"
23	#include "crypto.h"
24	#include <linux/ceph/osd_client.h>
25	#include <linux/ceph/striper.h>
26
27	/*
28	* Ceph address space ops.
29	*
30	* There are a few funny things going on here.
31	*
32	* The page->private field is used to reference a struct
33	* ceph_snap_context for _every_ dirty page. This indicates which
34	* snapshot the page was logically dirtied in, and thus which snap
35	* context needs to be associated with the osd write during writeback.
36	*
37	* Similarly, struct ceph_inode_info maintains a set of counters to
38	* count dirty pages on the inode. In the absence of snapshots,
39	* i_wrbuffer_ref == i_wrbuffer_ref_head == the dirty page count.
40	*
41	* When a snapshot is taken (that is, when the client receives
42	* notification that a snapshot was taken), each inode with caps and
43	* with dirty pages (dirty pages implies there is a cap) gets a new
44	* ceph_cap_snap in the i_cap_snaps list (which is sorted in ascending
45	* order, new snaps go to the tail). The i_wrbuffer_ref_head count is
46	* moved to capsnap->dirty. (Unless a sync write is currently in
47	* progress. In that case, the capsnap is said to be "pending", new
48	* writes cannot start, and the capsnap isn't "finalized" until the
49	* write completes (or fails) and a final size/mtime for the inode for
50	* that snap can be settled upon.) i_wrbuffer_ref_head is reset to 0.
51	*
52	* On writeback, we must submit writes to the osd IN SNAP ORDER. So,
53	* we look for the first capsnap in i_cap_snaps and write out pages in
54	* that snap context _only_. Then we move on to the next capsnap,
55	* eventually reaching the "live" or "head" context (i.e., pages that
56	* are not yet snapped) and are writing the most recently dirtied
57	* pages.
58	*
59	* Invalidate and so forth must take care to ensure the dirty page
60	* accounting is preserved.
61	*/
62
63	#define CONGESTION_ON_THRESH(congestion_kb) (congestion_kb >> (PAGE_SHIFT-10))
64	#define CONGESTION_OFF_THRESH(congestion_kb) \
65	(CONGESTION_ON_THRESH(congestion_kb) - \
66	(CONGESTION_ON_THRESH(congestion_kb) >> 2))
67
68	static int ceph_netfs_check_write_begin(struct file *file, uoff_t pos, unsigned int len,
69	struct folio foliop, void _fsdata);
70
71	static inline struct ceph_snap_context page_snap_context(struct page page)
72	{
73	if (PagePrivate(page))
74	return (void *)page->private;
75	return NULL;
76	}
77
78	/*
79	* Dirty a page. Optimistically adjust accounting, on the assumption
80	* that we won't race with invalidate. If we do, readjust.
81	*/
82	static bool ceph_dirty_folio(struct address_space mapping, struct folio folio)
83	{
84	struct inode *inode = mapping->host;
85	struct ceph_client *cl = ceph_inode_to_client(inode);
86	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
87	struct ceph_inode_info *ci;
88	struct ceph_snap_context *snapc;
89
90	if (folio_test_dirty(folio)) {
91	doutc(cl, "%llx.%llx %p idx %lu -- already dirty\n",
92	ceph_vinop(inode), folio, folio->index);
93	VM_BUG_ON_FOLIO(!folio_test_private(folio), folio);
94	return false;
95	}
96
97	atomic64_inc(&mdsc->dirty_folios);
98
99	ci = ceph_inode(inode);
100
101	/* dirty the head */
102	spin_lock(&ci->i_ceph_lock);
103	if (__ceph_have_pending_cap_snap(ci)) {
104	struct ceph_cap_snap *capsnap =
105	list_last_entry(&ci->i_cap_snaps,
106	struct ceph_cap_snap,
107	ci_item);
108	snapc = ceph_get_snap_context(capsnap->context);
109	capsnap->dirty_pages++;
110	} else {
111	BUG_ON(!ci->i_head_snapc);
112	snapc = ceph_get_snap_context(ci->i_head_snapc);
113	++ci->i_wrbuffer_ref_head;
114	}
115	if (ci->i_wrbuffer_ref == 0)
116	ihold(inode);
117	++ci->i_wrbuffer_ref;
118	doutc(cl, "%llx.%llx %p idx %lu head %d/%d -> %d/%d "
119	"snapc %p seq %lld (%d snaps)\n",
120	ceph_vinop(inode), folio, folio->index,
121	ci->i_wrbuffer_ref-1, ci->i_wrbuffer_ref_head-1,
122	ci->i_wrbuffer_ref, ci->i_wrbuffer_ref_head,
123	snapc, snapc->seq, snapc->num_snaps);
124	spin_unlock(&ci->i_ceph_lock);
125
126	/*
127	* Reference snap context in folio->private. Also set
128	* PagePrivate so that we get invalidate_folio callback.
129	*/
130	VM_WARN_ON_FOLIO(folio->private, folio);
131	folio_attach_private(folio, snapc);
132
133	return ceph_fscache_dirty_folio(mapping, folio);
134	}
135
136	/*
137	* If we are truncating the full folio (i.e. offset == 0), adjust the
138	* dirty folio counters appropriately. Only called if there is private
139	* data on the folio.
140	*/
141	static void ceph_invalidate_folio(struct folio *folio, size_t offset,
142	size_t length)
143	{
144	struct inode *inode = folio->mapping->host;
145	struct ceph_client *cl = ceph_inode_to_client(inode);
146	struct ceph_inode_info *ci = ceph_inode(inode);
147	struct ceph_snap_context *snapc;
148
149
150	if (offset != 0 \|\| length != folio_size(folio)) {
151	doutc(cl, "%llx.%llx idx %lu partial dirty page %zu~%zu\n",
152	ceph_vinop(inode), folio->index, offset, length);
153	return;
154	}
155
156	WARN_ON(!folio_test_locked(folio));
157	if (folio_test_private(folio)) {
158	doutc(cl, "%llx.%llx idx %lu full dirty page\n",
159	ceph_vinop(inode), folio->index);
160
161	snapc = folio_detach_private(folio);
162	ceph_put_wrbuffer_cap_refs(ci, 1, snapc);
163	ceph_put_snap_context(snapc);
164	}
165
166	netfs_invalidate_folio(folio, offset, length);
167	}
168
169	static void ceph_netfs_expand_readahead(struct netfs_io_request *rreq)
170	{
171	struct inode *inode = rreq->inode;
172	struct ceph_inode_info *ci = ceph_inode(inode);
173	struct ceph_file_layout *lo = &ci->i_layout;
174	unsigned long max_pages = inode->i_sb->s_bdi->ra_pages;
175	loff_t end = rreq->start + rreq->len, new_end;
176	struct ceph_netfs_request_data *priv = rreq->netfs_priv;
177	unsigned long max_len;
178	u32 blockoff;
179
180	if (priv) {
181	/* Readahead is disabled by posix_fadvise POSIX_FADV_RANDOM */
182	if (priv->file_ra_disabled)
183	max_pages = 0;
184	else
185	max_pages = priv->file_ra_pages;
186
187	}
188
189	/* Readahead is disabled */
190	if (!max_pages)
191	return;
192
193	max_len = max_pages << PAGE_SHIFT;
194
195	/*
196	* Try to expand the length forward by rounding up it to the next
197	* block, but do not exceed the file size, unless the original
198	* request already exceeds it.
199	*/
200	new_end = umin(round_up(end, lo->stripe_unit), rreq->i_size);
201	if (new_end > end && new_end <= rreq->start + max_len)
202	rreq->len = new_end - rreq->start;
203
204	/* Try to expand the start downward */
205	div_u64_rem(rreq->start, lo->stripe_unit, &blockoff);
206	if (rreq->len + blockoff <= max_len) {
207	rreq->start -= blockoff;
208	rreq->len += blockoff;
209	}
210	}
211
212	static void finish_netfs_read(struct ceph_osd_request *req)
213	{
214	struct inode *inode = req->r_inode;
215	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
216	struct ceph_client *cl = fsc->client;
217	struct ceph_osd_data *osd_data = osd_req_op_extent_osd_data(req, 0);
218	struct netfs_io_subrequest *subreq = req->r_priv;
219	struct ceph_osd_req_op *op = &req->r_ops[0];
220	int err = req->r_result;
221	bool sparse = (op->op == CEPH_OSD_OP_SPARSE_READ);
222
223	ceph_update_read_metrics(&fsc->mdsc->metric, req->r_start_latency,
224	req->r_end_latency, osd_data->length, err);
225
226	doutc(cl, "result %d subreq->len=%zu i_size=%lld\n", req->r_result,
227	subreq->len, i_size_read(req->r_inode));
228
229	/* no object means success but no data */
230	if (err == -ENOENT) {
231	__set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags);
232	__set_bit(NETFS_SREQ_MADE_PROGRESS, &subreq->flags);
233	err = 0;
234	} else if (err == -EBLOCKLISTED) {
235	fsc->blocklisted = true;
236	}
237
238	if (err >= 0) {
239	if (sparse && err > 0)
240	err = ceph_sparse_ext_map_end(op);
241	if (err < subreq->len &&
242	subreq->rreq->origin != NETFS_UNBUFFERED_READ &&
243	subreq->rreq->origin != NETFS_DIO_READ)
244	__set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags);
245	if (IS_ENCRYPTED(inode) && err > 0) {
246	err = ceph_fscrypt_decrypt_extents(inode,
247	osd_data->pages, subreq->start,
248	op->extent.sparse_ext,
249	op->extent.sparse_ext_cnt);
250	if (err > subreq->len)
251	err = subreq->len;
252	}
253	if (err > 0)
254	__set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags);
255	}
256
257	if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES) {
258	int num_pages = calc_pages_for(osd_data->alignment,
259	osd_data->length);
260
261	for (int i = 0; i < num_pages; i++)
262	put_page(osd_data->pages[i]);
263	kvfree(osd_data->pages);
264	}
265	if (err > 0) {
266	ceph_subvolume_metrics_record_io(fsc->mdsc, ceph_inode(inode),
267	false, err,
268	req->r_start_latency,
269	req->r_end_latency);
270	subreq->transferred = err;
271	err = 0;
272	}
273	subreq->error = err;
274	trace_netfs_sreq(subreq, netfs_sreq_trace_io_progress);
275	netfs_read_subreq_terminated(subreq);
276	iput(req->r_inode);
277	ceph_dec_osd_stopping_blocker(fsc->mdsc);
278	}
279
280	static bool ceph_netfs_issue_op_inline(struct netfs_io_subrequest *subreq)
281	{
282	struct netfs_io_request *rreq = subreq->rreq;
283	struct inode *inode = rreq->inode;
284	struct ceph_mds_reply_info_parsed *rinfo;
285	struct ceph_mds_reply_info_in *iinfo;
286	struct ceph_mds_request *req;
287	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
288	struct ceph_inode_info *ci = ceph_inode(inode);
289	ssize_t err = 0;
290	size_t len;
291	int mode;
292
293	if (rreq->origin != NETFS_UNBUFFERED_READ &&
294	rreq->origin != NETFS_DIO_READ)
295	__set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags);
296	__clear_bit(NETFS_SREQ_COPY_TO_CACHE, &subreq->flags);
297
298	if (subreq->start >= inode->i_size)
299	goto out;
300
301	/* We need to fetch the inline data. */
302	mode = ceph_try_to_choose_auth_mds(inode, CEPH_STAT_CAP_INLINE_DATA);
303	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_GETATTR, mode);
304	if (IS_ERR(req)) {
305	err = PTR_ERR(req);
306	goto out;
307	}
308	req->r_ino1 = ci->i_vino;
309	req->r_args.getattr.mask = cpu_to_le32(CEPH_STAT_CAP_INLINE_DATA);
310	req->r_num_caps = 2;
311
312	trace_netfs_sreq(subreq, netfs_sreq_trace_submit);
313	err = ceph_mdsc_do_request(mdsc, NULL, req);
314	if (err < 0)
315	goto out;
316
317	rinfo = &req->r_reply_info;
318	iinfo = &rinfo->targeti;
319	if (iinfo->inline_version == CEPH_INLINE_NONE) {
320	/* The data got uninlined */
321	ceph_mdsc_put_request(req);
322	return false;
323	}
324
325	len = min_t(size_t, iinfo->inline_len - subreq->start, subreq->len);
326	err = copy_to_iter(iinfo->inline_data + subreq->start, len, &subreq->io_iter);
327	if (err == 0) {
328	err = -EFAULT;
329	} else {
330	subreq->transferred += err;
331	err = 0;
332	}
333
334	ceph_mdsc_put_request(req);
335	out:
336	subreq->error = err;
337	trace_netfs_sreq(subreq, netfs_sreq_trace_io_progress);
338	netfs_read_subreq_terminated(subreq);
339	return true;
340	}
341
342	static int ceph_netfs_prepare_read(struct netfs_io_subrequest *subreq)
343	{
344	struct netfs_io_request *rreq = subreq->rreq;
345	struct inode *inode = rreq->inode;
346	struct ceph_inode_info *ci = ceph_inode(inode);
347	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
348	u64 objno, objoff;
349	u32 xlen;
350
351	/* Truncate the extent at the end of the current block */
352	ceph_calc_file_object_mapping(&ci->i_layout, subreq->start, subreq->len,
353	&objno, &objoff, &xlen);
354	rreq->io_streams[0].sreq_max_len = umin(xlen, fsc->mount_options->rsize);
355	return 0;
356	}
357
358	static void ceph_netfs_issue_read(struct netfs_io_subrequest *subreq)
359	{
360	struct netfs_io_request *rreq = subreq->rreq;
361	struct inode *inode = rreq->inode;
362	struct ceph_inode_info *ci = ceph_inode(inode);
363	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
364	struct ceph_client *cl = fsc->client;
365	struct ceph_osd_request *req = NULL;
366	struct ceph_vino vino = ceph_vino(inode);
367	int err;
368	u64 len;
369	bool sparse = IS_ENCRYPTED(inode) \|\| ceph_test_mount_opt(fsc, SPARSEREAD);
370	u64 off = subreq->start;
371	int extent_cnt;
372
373	if (ceph_inode_is_shutdown(inode)) {
374	err = -EIO;
375	goto out;
376	}
377
378	if (ceph_has_inline_data(ci) && ceph_netfs_issue_op_inline(subreq))
379	return;
380
381	// TODO: This rounding here is slightly dodgy. It should work, for
382	// now, as the cache only deals in blocks that are a multiple of
383	// PAGE_SIZE and fscrypt blocks are at most PAGE_SIZE. What needs to
384	// happen is for the fscrypt driving to be moved into netfslib and the
385	// data in the cache also to be stored encrypted.
386	len = subreq->len;
387	ceph_fscrypt_adjust_off_and_len(inode, &off, &len);
388
389	req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout, vino,
390	off, &len, 0, 1, sparse ? CEPH_OSD_OP_SPARSE_READ : CEPH_OSD_OP_READ,
391	CEPH_OSD_FLAG_READ, NULL, ci->i_truncate_seq,
392	ci->i_truncate_size, false);
393	if (IS_ERR(req)) {
394	err = PTR_ERR(req);
395	req = NULL;
396	goto out;
397	}
398
399	if (sparse) {
400	extent_cnt = __ceph_sparse_read_ext_count(inode, len);
401	err = ceph_alloc_sparse_ext_map(&req->r_ops[0], extent_cnt);
402	if (err)
403	goto out;
404	}
405
406	doutc(cl, "%llx.%llx pos=%llu orig_len=%zu len=%llu\n",
407	ceph_vinop(inode), subreq->start, subreq->len, len);
408
409	/*
410	* FIXME: For now, use CEPH_OSD_DATA_TYPE_PAGES instead of _ITER for
411	* encrypted inodes. We'd need infrastructure that handles an iov_iter
412	* instead of page arrays, and we don't have that as of yet. Once the
413	* dust settles on the write helpers and encrypt/decrypt routines for
414	* netfs, we should be able to rework this.
415	*/
416	if (IS_ENCRYPTED(inode)) {
417	struct page **pages;
418	size_t page_off;
419
420	/*
421	* FIXME: io_iter.count needs to be corrected to aligned
422	* length. Otherwise, iov_iter_get_pages_alloc2() operates
423	* with the initial unaligned length value. As a result,
424	* ceph_msg_data_cursor_init() triggers BUG_ON() in the case
425	* if msg->sparse_read_total > msg->data_length.
426	*/
427	subreq->io_iter.count = len;
428
429	err = iov_iter_get_pages_alloc2(&subreq->io_iter, &pages, len, &page_off);
430	if (err < 0) {
431	doutc(cl, "%llx.%llx failed to allocate pages, %d\n",
432	ceph_vinop(inode), err);
433	goto out;
434	}
435
436	/* should always give us a page-aligned read */
437	WARN_ON_ONCE(page_off);
438	len = err;
439	err = 0;
440
441	osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0, false,
442	false);
443	} else {
444	osd_req_op_extent_osd_iter(req, 0, &subreq->io_iter);
445	}
446	if (!ceph_inc_osd_stopping_blocker(fsc->mdsc)) {
447	err = -EIO;
448	goto out;
449	}
450	req->r_callback = finish_netfs_read;
451	req->r_priv = subreq;
452	req->r_inode = inode;
453	ihold(inode);
454
455	trace_netfs_sreq(subreq, netfs_sreq_trace_submit);
456	ceph_osdc_start_request(req->r_osdc, req);
457	out:
458	ceph_osdc_put_request(req);
459	if (err) {
460	subreq->error = err;
461	netfs_read_subreq_terminated(subreq);
462	}
463	doutc(cl, "%llx.%llx result %d\n", ceph_vinop(inode), err);
464	}
465
466	static int ceph_init_request(struct netfs_io_request rreq, struct file file)
467	{
468	struct inode *inode = rreq->inode;
469	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
470	struct ceph_client *cl = ceph_inode_to_client(inode);
471	int got = 0, want = CEPH_CAP_FILE_CACHE;
472	struct ceph_netfs_request_data *priv;
473	int ret = 0;
474
475	/* [DEPRECATED] Use PG_private_2 to mark folio being written to the cache. */
476	__set_bit(NETFS_RREQ_USE_PGPRIV2, &rreq->flags);
477
478	if (rreq->origin != NETFS_READAHEAD)
479	return 0;
480
481	priv = kzalloc_obj(*priv, GFP_NOFS);
482	if (!priv)
483	return -ENOMEM;
484
485	if (file) {
486	struct ceph_rw_context *rw_ctx;
487	struct ceph_file_info *fi = file->private_data;
488
489	priv->file_ra_pages = file->f_ra.ra_pages;
490	priv->file_ra_disabled = file->f_mode & FMODE_RANDOM;
491
492	rw_ctx = ceph_find_rw_context(fi);
493	if (rw_ctx) {
494	rreq->netfs_priv = priv;
495	return 0;
496	}
497	}
498
499	/*
500	* readahead callers do not necessarily hold Fcb caps
501	* (e.g. fadvise, madvise).
502	*/
503	ret = ceph_try_get_caps(inode, CEPH_CAP_FILE_RD, want, true, &got);
504	if (ret < 0) {
505	doutc(cl, "%llx.%llx, error getting cap\n", ceph_vinop(inode));
506	goto out;
507	}
508
509	if (!(got & want)) {
510	doutc(cl, "%llx.%llx, no cache cap\n", ceph_vinop(inode));
511	ret = -EACCES;
512	goto out;
513	}
514	if (ret == 0) {
515	ret = -EACCES;
516	goto out;
517	}
518
519	priv->caps = got;
520	rreq->netfs_priv = priv;
521	rreq->io_streams[0].sreq_max_len = fsc->mount_options->rsize;
522
523	out:
524	if (ret < 0) {
525	if (got)
526	ceph_put_cap_refs(ceph_inode(inode), got);
527	kfree(priv);
528	}
529
530	return ret;
531	}
532
533	static void ceph_netfs_free_request(struct netfs_io_request *rreq)
534	{
535	struct ceph_netfs_request_data *priv = rreq->netfs_priv;
536
537	if (!priv)
538	return;
539
540	if (priv->caps)
541	ceph_put_cap_refs(ceph_inode(rreq->inode), priv->caps);
542	kfree(priv);
543	rreq->netfs_priv = NULL;
544	}
545
546	const struct netfs_request_ops ceph_netfs_ops = {
547	.init_request = ceph_init_request,
548	.free_request = ceph_netfs_free_request,
549	.prepare_read = ceph_netfs_prepare_read,
550	.issue_read = ceph_netfs_issue_read,
551	.expand_readahead = ceph_netfs_expand_readahead,
552	.check_write_begin = ceph_netfs_check_write_begin,
553	};
554
555	#ifdef CONFIG_CEPH_FSCACHE
556	static void ceph_set_page_fscache(struct page *page)
557	{
558	folio_start_private_2(page_folio(page)); /* [DEPRECATED] */
559	}
560
561	static void ceph_fscache_write_terminated(void *priv, ssize_t error)
562	{
563	struct inode *inode = priv;
564
565	if (IS_ERR_VALUE(error) && error != -ENOBUFS)
566	ceph_fscache_invalidate(inode, false);
567	}
568
569	static void ceph_fscache_write_to_cache(struct inode *inode, u64 off, u64 len, bool caching)
570	{
571	struct ceph_inode_info *ci = ceph_inode(inode);
572	struct fscache_cookie *cookie = ceph_fscache_cookie(ci);
573
574	fscache_write_to_cache(cookie, inode->i_mapping, off, len, i_size_read(inode),
575	ceph_fscache_write_terminated, inode, true, caching);
576	}
577	#else
578	static inline void ceph_set_page_fscache(struct page *page)
579	{
580	}
581
582	static inline void ceph_fscache_write_to_cache(struct inode *inode, u64 off, u64 len, bool caching)
583	{
584	}
585	#endif /* CONFIG_CEPH_FSCACHE */
586
587	struct ceph_writeback_ctl
588	{
589	loff_t i_size;
590	u64 truncate_size;
591	u32 truncate_seq;
592	bool size_stable;
593
594	bool head_snapc;
595	struct ceph_snap_context *snapc;
596	struct ceph_snap_context *last_snapc;
597
598	bool done;
599	bool should_loop;
600	bool range_whole;
601	pgoff_t start_index;
602	pgoff_t index;
603	pgoff_t end;
604	xa_mark_t tag;
605
606	pgoff_t strip_unit_end;
607	unsigned int wsize;
608	unsigned int nr_folios;
609	unsigned int max_pages;
610	unsigned int locked_pages;
611
612	int op_idx;
613	int num_ops;
614	u64 offset;
615	u64 len;
616
617	struct folio_batch fbatch;
618	unsigned int processed_in_fbatch;
619
620	bool from_pool;
621	struct page **pages;
622	struct page **data_pages;
623	};
624
625	/*
626	* Get ref for the oldest snapc for an inode with dirty data... that is, the
627	* only snap context we are allowed to write back.
628	*/
629	static struct ceph_snap_context *
630	get_oldest_context(struct inode inode, struct ceph_writeback_ctl ctl,
631	struct ceph_snap_context *page_snapc)
632	{
633	struct ceph_inode_info *ci = ceph_inode(inode);
634	struct ceph_client *cl = ceph_inode_to_client(inode);
635	struct ceph_snap_context *snapc = NULL;
636	struct ceph_cap_snap *capsnap = NULL;
637
638	spin_lock(&ci->i_ceph_lock);
639	list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
640	doutc(cl, " capsnap %p snapc %p has %d dirty pages\n",
641	capsnap, capsnap->context, capsnap->dirty_pages);
642	if (!capsnap->dirty_pages)
643	continue;
644
645	/* get i_size, truncate_{seq,size} for page_snapc? */
646	if (snapc && capsnap->context != page_snapc)
647	continue;
648
649	if (ctl) {
650	if (capsnap->writing) {
651	ctl->i_size = i_size_read(inode);
652	ctl->size_stable = false;
653	} else {
654	ctl->i_size = capsnap->size;
655	ctl->size_stable = true;
656	}
657	ctl->truncate_size = capsnap->truncate_size;
658	ctl->truncate_seq = capsnap->truncate_seq;
659	ctl->head_snapc = false;
660	}
661
662	if (snapc)
663	break;
664
665	snapc = ceph_get_snap_context(capsnap->context);
666	if (!page_snapc \|\|
667	page_snapc == snapc \|\|
668	page_snapc->seq > snapc->seq)
669	break;
670	}
671	if (!snapc && ci->i_wrbuffer_ref_head) {
672	snapc = ceph_get_snap_context(ci->i_head_snapc);
673	doutc(cl, " head snapc %p has %d dirty pages\n", snapc,
674	ci->i_wrbuffer_ref_head);
675	if (ctl) {
676	ctl->i_size = i_size_read(inode);
677	ctl->truncate_size = ci->i_truncate_size;
678	ctl->truncate_seq = ci->i_truncate_seq;
679	ctl->size_stable = false;
680	ctl->head_snapc = true;
681	}
682	}
683	spin_unlock(&ci->i_ceph_lock);
684	return snapc;
685	}
686
687	static u64 get_writepages_data_length(struct inode *inode,
688	struct page *page, u64 start)
689	{
690	struct ceph_inode_info *ci = ceph_inode(inode);
691	struct ceph_snap_context *snapc;
692	struct ceph_cap_snap *capsnap = NULL;
693	u64 end = i_size_read(inode);
694	u64 ret;
695
696	snapc = page_snap_context(ceph_fscrypt_pagecache_page(page));
697	if (snapc != ci->i_head_snapc) {
698	bool found = false;
699	spin_lock(&ci->i_ceph_lock);
700	list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
701	if (capsnap->context == snapc) {
702	if (!capsnap->writing)
703	end = capsnap->size;
704	found = true;
705	break;
706	}
707	}
708	spin_unlock(&ci->i_ceph_lock);
709	WARN_ON(!found);
710	}
711	if (end > ceph_fscrypt_page_offset(page) + thp_size(page))
712	end = ceph_fscrypt_page_offset(page) + thp_size(page);
713	ret = end > start ? end - start : 0;
714	if (ret && fscrypt_is_bounce_page(page))
715	ret = round_up(ret, CEPH_FSCRYPT_BLOCK_SIZE);
716	return ret;
717	}
718
719	/*
720	* Write a folio, but leave it locked.
721	*
722	* If we get a write error, mark the mapping for error, but still adjust the
723	* dirty page accounting (i.e., folio is no longer dirty).
724	*/
725	static int write_folio_nounlock(struct folio *folio,
726	struct writeback_control *wbc)
727	{
728	struct page *page = &folio->page;
729	struct inode *inode = folio->mapping->host;
730	struct ceph_inode_info *ci = ceph_inode(inode);
731	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
732	struct ceph_client *cl = fsc->client;
733	struct ceph_snap_context snapc, oldest;
734	loff_t page_off = folio_pos(folio);
735	int err;
736	loff_t len = folio_size(folio);
737	loff_t wlen;
738	struct ceph_writeback_ctl ceph_wbc;
739	struct ceph_osd_client *osdc = &fsc->client->osdc;
740	struct ceph_osd_request *req;
741	bool caching = ceph_is_cache_enabled(inode);
742	struct page *bounce_page = NULL;
743
744	doutc(cl, "%llx.%llx folio %p idx %lu\n", ceph_vinop(inode), folio,
745	folio->index);
746
747	if (ceph_inode_is_shutdown(inode))
748	return -EIO;
749
750	/* verify this is a writeable snap context */
751	snapc = page_snap_context(&folio->page);
752	if (!snapc) {
753	doutc(cl, "%llx.%llx folio %p not dirty?\n", ceph_vinop(inode),
754	folio);
755	return 0;
756	}
757	oldest = get_oldest_context(inode, &ceph_wbc, snapc);
758	if (snapc->seq > oldest->seq) {
759	doutc(cl, "%llx.%llx folio %p snapc %p not writeable - noop\n",
760	ceph_vinop(inode), folio, snapc);
761	/* we should only noop if called by kswapd */
762	WARN_ON(!(current->flags & PF_MEMALLOC));
763	ceph_put_snap_context(oldest);
764	folio_redirty_for_writepage(wbc, folio);
765	return 0;
766	}
767	ceph_put_snap_context(oldest);
768
769	/* is this a partial page at end of file? */
770	if (page_off >= ceph_wbc.i_size) {
771	doutc(cl, "%llx.%llx folio at %lu beyond eof %llu\n",
772	ceph_vinop(inode), folio->index, ceph_wbc.i_size);
773	folio_invalidate(folio, 0, folio_size(folio));
774	return 0;
775	}
776
777	if (ceph_wbc.i_size < page_off + len)
778	len = ceph_wbc.i_size - page_off;
779
780	wlen = IS_ENCRYPTED(inode) ? round_up(len, CEPH_FSCRYPT_BLOCK_SIZE) : len;
781	doutc(cl, "%llx.%llx folio %p index %lu on %llu~%llu snapc %p seq %lld\n",
782	ceph_vinop(inode), folio, folio->index, page_off, wlen, snapc,
783	snapc->seq);
784
785	if (atomic_long_inc_return(&fsc->writeback_count) >
786	CONGESTION_ON_THRESH(fsc->mount_options->congestion_kb))
787	fsc->write_congested = true;
788
789	req = ceph_osdc_new_request(osdc, &ci->i_layout, ceph_vino(inode),
790	page_off, &wlen, 0, 1, CEPH_OSD_OP_WRITE,
791	CEPH_OSD_FLAG_WRITE, snapc,
792	ceph_wbc.truncate_seq,
793	ceph_wbc.truncate_size, true);
794	if (IS_ERR(req)) {
795	folio_redirty_for_writepage(wbc, folio);
796	if (atomic_long_dec_return(&fsc->writeback_count) <
797	CONGESTION_OFF_THRESH(fsc->mount_options->congestion_kb))
798	fsc->write_congested = false;
799	return PTR_ERR(req);
800	}
801
802	if (wlen < len)
803	len = wlen;
804
805	folio_start_writeback(folio);
806	if (caching)
807	ceph_set_page_fscache(&folio->page);
808	ceph_fscache_write_to_cache(inode, page_off, len, caching);
809
810	if (IS_ENCRYPTED(inode)) {
811	bounce_page = fscrypt_encrypt_pagecache_blocks(folio,
812	CEPH_FSCRYPT_BLOCK_SIZE, 0,
813	GFP_NOFS);
814	if (IS_ERR(bounce_page)) {
815	folio_redirty_for_writepage(wbc, folio);
816	folio_end_writeback(folio);
817	ceph_osdc_put_request(req);
818	if (atomic_long_dec_return(&fsc->writeback_count) <
819	CONGESTION_OFF_THRESH(fsc->mount_options->congestion_kb))
820	fsc->write_congested = false;
821	return PTR_ERR(bounce_page);
822	}
823	}
824
825	/* it may be a short write due to an object boundary */
826	WARN_ON_ONCE(len > folio_size(folio));
827	osd_req_op_extent_osd_data_pages(req, 0,
828	bounce_page ? &bounce_page : &page, wlen, 0,
829	false, false);
830	doutc(cl, "%llx.%llx %llu~%llu (%llu bytes, %sencrypted)\n",
831	ceph_vinop(inode), page_off, len, wlen,
832	IS_ENCRYPTED(inode) ? "" : "not ");
833
834	req->r_mtime = inode_get_mtime(inode);
835	ceph_osdc_start_request(osdc, req);
836	err = ceph_osdc_wait_request(osdc, req);
837
838	ceph_update_write_metrics(&fsc->mdsc->metric, req->r_start_latency,
839	req->r_end_latency, len, err);
840	if (err >= 0 && len > 0)
841	ceph_subvolume_metrics_record_io(fsc->mdsc, ci, true, len,
842	req->r_start_latency,
843	req->r_end_latency);
844	fscrypt_free_bounce_page(bounce_page);
845	ceph_osdc_put_request(req);
846	if (err == 0)
847	err = len;
848
849	if (err < 0) {
850	struct writeback_control tmp_wbc;
851	if (!wbc)
852	wbc = &tmp_wbc;
853	if (err == -ERESTARTSYS) {
854	/* killed by SIGKILL */
855	doutc(cl, "%llx.%llx interrupted page %p\n",
856	ceph_vinop(inode), folio);
857	folio_redirty_for_writepage(wbc, folio);
858	folio_end_writeback(folio);
859	if (atomic_long_dec_return(&fsc->writeback_count) <
860	CONGESTION_OFF_THRESH(fsc->mount_options->congestion_kb))
861	fsc->write_congested = false;
862	return err;
863	}
864	if (err == -EBLOCKLISTED)
865	fsc->blocklisted = true;
866	doutc(cl, "%llx.%llx setting mapping error %d %p\n",
867	ceph_vinop(inode), err, folio);
868	mapping_set_error(&inode->i_data, err);
869	wbc->pages_skipped++;
870	} else {
871	doutc(cl, "%llx.%llx cleaned page %p\n",
872	ceph_vinop(inode), folio);
873	err = 0; /* vfs expects us to return 0 */
874	}
875	oldest = folio_detach_private(folio);
876	WARN_ON_ONCE(oldest != snapc);
877	folio_end_writeback(folio);
878	ceph_put_wrbuffer_cap_refs(ci, 1, snapc);
879	ceph_put_snap_context(snapc); /* page's reference */
880
881	if (atomic_long_dec_return(&fsc->writeback_count) <
882	CONGESTION_OFF_THRESH(fsc->mount_options->congestion_kb))
883	fsc->write_congested = false;
884
885	return err;
886	}
887
888	/*
889	* async writeback completion handler.
890	*
891	* If we get an error, set the mapping error bit, but not the individual
892	* page error bits.
893	*/
894	static void writepages_finish(struct ceph_osd_request *req)
895	{
896	struct inode *inode = req->r_inode;
897	struct ceph_inode_info *ci = ceph_inode(inode);
898	struct ceph_client *cl = ceph_inode_to_client(inode);
899	struct ceph_osd_data *osd_data;
900	struct page *page;
901	int num_pages, total_pages = 0;
902	int i, j;
903	int rc = req->r_result;
904	struct ceph_snap_context *snapc = req->r_snapc;
905	struct address_space *mapping = inode->i_mapping;
906	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
907	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
908	unsigned int len = 0;
909	bool remove_page;
910
911	doutc(cl, "%llx.%llx rc %d\n", ceph_vinop(inode), rc);
912	if (rc < 0) {
913	mapping_set_error(mapping, rc);
914	ceph_set_error_write(ci);
915	if (rc == -EBLOCKLISTED)
916	fsc->blocklisted = true;
917	} else {
918	ceph_clear_error_write(ci);
919	}
920
921	/*
922	* We lost the cache cap, need to truncate the page before
923	* it is unlocked, otherwise we'd truncate it later in the
924	* page truncation thread, possibly losing some data that
925	* raced its way in
926	*/
927	remove_page = !(ceph_caps_issued(ci) &
928	(CEPH_CAP_FILE_CACHE\|CEPH_CAP_FILE_LAZYIO));
929
930	/* clean all pages */
931	for (i = 0; i < req->r_num_ops; i++) {
932	if (req->r_ops[i].op != CEPH_OSD_OP_WRITE) {
933	pr_warn_client(cl,
934	"%llx.%llx incorrect op %d req %p index %d tid %llu\n",
935	ceph_vinop(inode), req->r_ops[i].op, req, i,
936	req->r_tid);
937	break;
938	}
939
940	osd_data = osd_req_op_extent_osd_data(req, i);
941	BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGES);
942	len += osd_data->length;
943	num_pages = calc_pages_for((u64)osd_data->alignment,
944	(u64)osd_data->length);
945	total_pages += num_pages;
946	for (j = 0; j < num_pages; j++) {
947	page = osd_data->pages[j];
948	if (fscrypt_is_bounce_page(page)) {
949	page = fscrypt_pagecache_page(page);
950	fscrypt_free_bounce_page(osd_data->pages[j]);
951	osd_data->pages[j] = page;
952	}
953	BUG_ON(!page);
954	WARN_ON(!PageUptodate(page));
955
956	if (atomic_long_dec_return(&fsc->writeback_count) <
957	CONGESTION_OFF_THRESH(
958	fsc->mount_options->congestion_kb))
959	fsc->write_congested = false;
960
961	ceph_put_snap_context(detach_page_private(page));
962	end_page_writeback(page);
963
964	if (atomic64_dec_return(&mdsc->dirty_folios) <= 0) {
965	wake_up_all(&mdsc->flush_end_wq);
966	WARN_ON(atomic64_read(&mdsc->dirty_folios) < 0);
967	}
968
969	doutc(cl, "unlocking %p\n", page);
970
971	if (remove_page)
972	generic_error_remove_folio(inode->i_mapping,
973	page_folio(page));
974
975	unlock_page(page);
976	}
977	doutc(cl, "%llx.%llx wrote %llu bytes cleaned %d pages\n",
978	ceph_vinop(inode), osd_data->length,
979	rc >= 0 ? num_pages : 0);
980
981	release_pages(osd_data->pages, num_pages);
982	}
983
984	ceph_update_write_metrics(&fsc->mdsc->metric, req->r_start_latency,
985	req->r_end_latency, len, rc);
986
987	if (rc >= 0 && len > 0)
988	ceph_subvolume_metrics_record_io(mdsc, ci, true, len,
989	req->r_start_latency,
990	req->r_end_latency);
991
992	ceph_put_wrbuffer_cap_refs(ci, total_pages, snapc);
993
994	osd_data = osd_req_op_extent_osd_data(req, 0);
995	if (osd_data->pages_from_pool)
996	mempool_free(osd_data->pages, ceph_wb_pagevec_pool);
997	else
998	kfree(osd_data->pages);
999	ceph_osdc_put_request(req);
1000	ceph_dec_osd_stopping_blocker(fsc->mdsc);
1001	}
1002
1003	static inline
1004	bool is_forced_umount(struct address_space *mapping)
1005	{
1006	struct inode *inode = mapping->host;
1007	struct ceph_inode_info *ci = ceph_inode(inode);
1008	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1009	struct ceph_client *cl = fsc->client;
1010
1011	if (ceph_inode_is_shutdown(inode)) {
1012	if (ci->i_wrbuffer_ref > 0) {
1013	pr_warn_ratelimited_client(cl,
1014	"%llx.%llx %lld forced umount\n",
1015	ceph_vinop(inode), ceph_ino(inode));
1016	}
1017	mapping_set_error(mapping, -EIO);
1018	return true;
1019	}
1020
1021	return false;
1022	}
1023
1024	static inline
1025	unsigned int ceph_define_write_size(struct address_space *mapping)
1026	{
1027	struct inode *inode = mapping->host;
1028	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1029	struct ceph_inode_info *ci = ceph_inode(inode);
1030	unsigned int wsize = ci->i_layout.stripe_unit;
1031
1032	if (fsc->mount_options->wsize < wsize)
1033	wsize = fsc->mount_options->wsize;
1034
1035	return wsize;
1036	}
1037
1038	static inline
1039	void ceph_folio_batch_init(struct ceph_writeback_ctl *ceph_wbc)
1040	{
1041	folio_batch_init(&ceph_wbc->fbatch);
1042	ceph_wbc->processed_in_fbatch = 0;
1043	}
1044
1045	static inline
1046	void ceph_folio_batch_reinit(struct ceph_writeback_ctl *ceph_wbc)
1047	{
1048	folio_batch_release(&ceph_wbc->fbatch);
1049	ceph_folio_batch_init(ceph_wbc);
1050	}
1051
1052	static inline
1053	void ceph_init_writeback_ctl(struct address_space *mapping,
1054	struct writeback_control *wbc,
1055	struct ceph_writeback_ctl *ceph_wbc)
1056	{
1057	ceph_wbc->snapc = NULL;
1058	ceph_wbc->last_snapc = NULL;
1059
1060	ceph_wbc->strip_unit_end = 0;
1061	ceph_wbc->wsize = ceph_define_write_size(mapping);
1062
1063	ceph_wbc->nr_folios = 0;
1064	ceph_wbc->max_pages = 0;
1065	ceph_wbc->locked_pages = 0;
1066
1067	ceph_wbc->done = false;
1068	ceph_wbc->should_loop = false;
1069	ceph_wbc->range_whole = false;
1070
1071	ceph_wbc->start_index = wbc->range_cyclic ? mapping->writeback_index : 0;
1072	ceph_wbc->index = ceph_wbc->start_index;
1073	ceph_wbc->end = -1;
1074
1075	ceph_wbc->tag = wbc_to_tag(wbc);
1076
1077	ceph_wbc->op_idx = -1;
1078	ceph_wbc->num_ops = 0;
1079	ceph_wbc->offset = 0;
1080	ceph_wbc->len = 0;
1081	ceph_wbc->from_pool = false;
1082
1083	ceph_folio_batch_init(ceph_wbc);
1084
1085	ceph_wbc->pages = NULL;
1086	ceph_wbc->data_pages = NULL;
1087	}
1088
1089	static inline
1090	int ceph_define_writeback_range(struct address_space *mapping,
1091	struct writeback_control *wbc,
1092	struct ceph_writeback_ctl *ceph_wbc)
1093	{
1094	struct inode *inode = mapping->host;
1095	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1096	struct ceph_client *cl = fsc->client;
1097
1098	/* find oldest snap context with dirty data */
1099	ceph_wbc->snapc = get_oldest_context(inode, ceph_wbc, NULL);
1100	if (!ceph_wbc->snapc) {
1101	/* hmm, why does writepages get called when there
1102	is no dirty data? */
1103	doutc(cl, " no snap context with dirty data?\n");
1104	return -ENODATA;
1105	}
1106
1107	doutc(cl, " oldest snapc is %p seq %lld (%d snaps)\n",
1108	ceph_wbc->snapc, ceph_wbc->snapc->seq,
1109	ceph_wbc->snapc->num_snaps);
1110
1111	ceph_wbc->should_loop = false;
1112
1113	if (ceph_wbc->head_snapc && ceph_wbc->snapc != ceph_wbc->last_snapc) {
1114	/* where to start/end? */
1115	if (wbc->range_cyclic) {
1116	ceph_wbc->index = ceph_wbc->start_index;
1117	ceph_wbc->end = -1;
1118	if (ceph_wbc->index > 0)
1119	ceph_wbc->should_loop = true;
1120	doutc(cl, " cyclic, start at %lu\n", ceph_wbc->index);
1121	} else {
1122	ceph_wbc->index = wbc->range_start >> PAGE_SHIFT;
1123	ceph_wbc->end = wbc->range_end >> PAGE_SHIFT;
1124	if (wbc->range_start == 0 && wbc->range_end == LLONG_MAX)
1125	ceph_wbc->range_whole = true;
1126	doutc(cl, " not cyclic, %lu to %lu\n",
1127	ceph_wbc->index, ceph_wbc->end);
1128	}
1129	} else if (!ceph_wbc->head_snapc) {
1130	/* Do not respect wbc->range_{start,end}. Dirty pages
1131	* in that range can be associated with newer snapc.
1132	* They are not writeable until we write all dirty pages
1133	* associated with 'snapc' get written */
1134	if (ceph_wbc->index > 0)
1135	ceph_wbc->should_loop = true;
1136	doutc(cl, " non-head snapc, range whole\n");
1137	}
1138
1139	ceph_put_snap_context(ceph_wbc->last_snapc);
1140	ceph_wbc->last_snapc = ceph_wbc->snapc;
1141
1142	return 0;
1143	}
1144
1145	static inline
1146	bool has_writeback_done(struct ceph_writeback_ctl *ceph_wbc)
1147	{
1148	return ceph_wbc->done && ceph_wbc->index > ceph_wbc->end;
1149	}
1150
1151	static inline
1152	bool can_next_page_be_processed(struct ceph_writeback_ctl *ceph_wbc,
1153	unsigned index)
1154	{
1155	return index < ceph_wbc->nr_folios &&
1156	ceph_wbc->locked_pages < ceph_wbc->max_pages;
1157	}
1158
1159	static
1160	int ceph_check_page_before_write(struct address_space *mapping,
1161	struct writeback_control *wbc,
1162	struct ceph_writeback_ctl *ceph_wbc,
1163	struct folio *folio)
1164	{
1165	struct inode *inode = mapping->host;
1166	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1167	struct ceph_client *cl = fsc->client;
1168	struct ceph_snap_context *pgsnapc;
1169
1170	/* only dirty folios, or our accounting breaks */
1171	if (unlikely(!folio_test_dirty(folio) \|\| folio->mapping != mapping)) {
1172	doutc(cl, "!dirty or !mapping %p\n", folio);
1173	return -ENODATA;
1174	}
1175
1176	/* only if matching snap context */
1177	pgsnapc = page_snap_context(&folio->page);
1178	if (pgsnapc != ceph_wbc->snapc) {
1179	doutc(cl, "folio snapc %p %lld != oldest %p %lld\n",
1180	pgsnapc, pgsnapc->seq,
1181	ceph_wbc->snapc, ceph_wbc->snapc->seq);
1182
1183	if (!ceph_wbc->should_loop && !ceph_wbc->head_snapc &&
1184	wbc->sync_mode != WB_SYNC_NONE)
1185	ceph_wbc->should_loop = true;
1186
1187	return -ENODATA;
1188	}
1189
1190	if (folio_pos(folio) >= ceph_wbc->i_size) {
1191	doutc(cl, "folio at %lu beyond eof %llu\n",
1192	folio->index, ceph_wbc->i_size);
1193
1194	if ((ceph_wbc->size_stable \|\|
1195	folio_pos(folio) >= i_size_read(inode)) &&
1196	folio_clear_dirty_for_io(folio))
1197	folio_invalidate(folio, 0, folio_size(folio));
1198
1199	return -ENODATA;
1200	}
1201
1202	if (ceph_wbc->strip_unit_end &&
1203	(folio->index > ceph_wbc->strip_unit_end)) {
1204	doutc(cl, "end of strip unit %p\n", folio);
1205	return -E2BIG;
1206	}
1207
1208	return 0;
1209	}
1210
1211	static inline
1212	void __ceph_allocate_page_array(struct ceph_writeback_ctl *ceph_wbc,
1213	unsigned int max_pages)
1214	{
1215	ceph_wbc->pages = kmalloc_objs(*ceph_wbc->pages, max_pages, GFP_NOFS);
1216	if (!ceph_wbc->pages) {
1217	ceph_wbc->from_pool = true;
1218	ceph_wbc->pages = mempool_alloc(ceph_wb_pagevec_pool, GFP_NOFS);
1219	BUG_ON(!ceph_wbc->pages);
1220	}
1221	}
1222
1223	static inline
1224	void ceph_allocate_page_array(struct address_space *mapping,
1225	struct ceph_writeback_ctl *ceph_wbc,
1226	struct folio *folio)
1227	{
1228	struct inode *inode = mapping->host;
1229	struct ceph_inode_info *ci = ceph_inode(inode);
1230	u64 objnum;
1231	u64 objoff;
1232	u32 xlen;
1233
1234	/* prepare async write request */
1235	ceph_wbc->offset = (u64)folio_pos(folio);
1236	ceph_calc_file_object_mapping(&ci->i_layout,
1237	ceph_wbc->offset, ceph_wbc->wsize,
1238	&objnum, &objoff, &xlen);
1239
1240	ceph_wbc->num_ops = 1;
1241	ceph_wbc->strip_unit_end = folio->index + ((xlen - 1) >> PAGE_SHIFT);
1242
1243	BUG_ON(ceph_wbc->pages);
1244	ceph_wbc->max_pages = calc_pages_for(0, (u64)xlen);
1245	__ceph_allocate_page_array(ceph_wbc, ceph_wbc->max_pages);
1246
1247	ceph_wbc->len = 0;
1248	}
1249
1250	static inline
1251	bool is_folio_index_contiguous(const struct ceph_writeback_ctl *ceph_wbc,
1252	const struct folio *folio)
1253	{
1254	return folio->index == (ceph_wbc->offset + ceph_wbc->len) >> PAGE_SHIFT;
1255	}
1256
1257	static inline
1258	bool is_num_ops_too_big(struct ceph_writeback_ctl *ceph_wbc)
1259	{
1260	return ceph_wbc->num_ops >=
1261	(ceph_wbc->from_pool ? CEPH_OSD_SLAB_OPS : CEPH_OSD_MAX_OPS);
1262	}
1263
1264	static inline
1265	bool is_write_congestion_happened(struct ceph_fs_client *fsc)
1266	{
1267	return atomic_long_inc_return(&fsc->writeback_count) >
1268	CONGESTION_ON_THRESH(fsc->mount_options->congestion_kb);
1269	}
1270
1271	static inline int move_dirty_folio_in_page_array(struct address_space *mapping,
1272	struct writeback_control *wbc,
1273	struct ceph_writeback_ctl ceph_wbc, struct folio folio)
1274	{
1275	struct inode *inode = mapping->host;
1276	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1277	struct ceph_client *cl = fsc->client;
1278	struct page **pages = ceph_wbc->pages;
1279	unsigned int index = ceph_wbc->locked_pages;
1280	gfp_t gfp_flags = ceph_wbc->locked_pages ? GFP_NOWAIT : GFP_NOFS;
1281
1282	if (IS_ENCRYPTED(inode)) {
1283	pages[index] = fscrypt_encrypt_pagecache_blocks(folio,
1284	PAGE_SIZE,
1285	0,
1286	gfp_flags);
1287	if (IS_ERR(pages[index])) {
1288	int err = PTR_ERR(pages[index]);
1289
1290	if (err == -EINVAL) {
1291	pr_err_client(cl, "inode->i_blkbits=%hhu\n",
1292	inode->i_blkbits);
1293	}
1294
1295	/* better not fail on first page! */
1296	BUG_ON(ceph_wbc->locked_pages == 0);
1297
1298	pages[index] = NULL;
1299	return err;
1300	}
1301	} else {
1302	pages[index] = &folio->page;
1303	}
1304
1305	ceph_wbc->locked_pages++;
1306
1307	return 0;
1308	}
1309
1310	static
1311	void ceph_process_folio_batch(struct address_space *mapping,
1312	struct writeback_control *wbc,
1313	struct ceph_writeback_ctl *ceph_wbc)
1314	{
1315	struct inode *inode = mapping->host;
1316	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1317	struct ceph_client *cl = fsc->client;
1318	struct folio *folio = NULL;
1319	unsigned i;
1320	int rc;
1321
1322	for (i = 0; can_next_page_be_processed(ceph_wbc, i); i++) {
1323	folio = ceph_wbc->fbatch.folios[i];
1324
1325	if (!folio)
1326	continue;
1327
1328	doutc(cl, "? %p idx %lu, folio_test_writeback %#x, "
1329	"folio_test_dirty %#x, folio_test_locked %#x\n",
1330	folio, folio->index, folio_test_writeback(folio),
1331	folio_test_dirty(folio),
1332	folio_test_locked(folio));
1333
1334	if (folio_test_writeback(folio) \|\|
1335	folio_test_private_2(folio) /* [DEPRECATED] */) {
1336	doutc(cl, "waiting on writeback %p\n", folio);
1337	folio_wait_writeback(folio);
1338	folio_wait_private_2(folio); /* [DEPRECATED] */
1339	continue;
1340	}
1341
1342	if (ceph_wbc->locked_pages == 0)
1343	folio_lock(folio);
1344	else if (!folio_trylock(folio))
1345	break;
1346
1347	rc = ceph_check_page_before_write(mapping, wbc,
1348	ceph_wbc, folio);
1349	if (rc == -ENODATA) {
1350	folio_unlock(folio);
1351	folio_put(folio);
1352	ceph_wbc->fbatch.folios[i] = NULL;
1353	continue;
1354	} else if (rc == -E2BIG) {
1355	folio_unlock(folio);
1356	break;
1357	}
1358
1359	if (!folio_clear_dirty_for_io(folio)) {
1360	doutc(cl, "%p !folio_clear_dirty_for_io\n", folio);
1361	folio_unlock(folio);
1362	folio_put(folio);
1363	ceph_wbc->fbatch.folios[i] = NULL;
1364	continue;
1365	}
1366
1367	/*
1368	* We have something to write. If this is
1369	* the first locked page this time through,
1370	* calculate max possible write size and
1371	* allocate a page array
1372	*/
1373	if (ceph_wbc->locked_pages == 0) {
1374	ceph_allocate_page_array(mapping, ceph_wbc, folio);
1375	} else if (!is_folio_index_contiguous(ceph_wbc, folio)) {
1376	if (is_num_ops_too_big(ceph_wbc)) {
1377	folio_redirty_for_writepage(wbc, folio);
1378	folio_unlock(folio);
1379	break;
1380	}
1381
1382	ceph_wbc->num_ops++;
1383	ceph_wbc->offset = (u64)folio_pos(folio);
1384	ceph_wbc->len = 0;
1385	}
1386
1387	/* note position of first page in fbatch */
1388	doutc(cl, "%llx.%llx will write folio %p idx %lu\n",
1389	ceph_vinop(inode), folio, folio->index);
1390
1391	fsc->write_congested = is_write_congestion_happened(fsc);
1392
1393	rc = move_dirty_folio_in_page_array(mapping, wbc, ceph_wbc,
1394	folio);
1395	if (rc) {
1396	/* Did we just begin a new contiguous op? Nevermind! */
1397	if (ceph_wbc->len == 0)
1398	ceph_wbc->num_ops--;
1399
1400	folio_redirty_for_writepage(wbc, folio);
1401	folio_unlock(folio);
1402	break;
1403	}
1404
1405	ceph_wbc->fbatch.folios[i] = NULL;
1406	ceph_wbc->len += folio_size(folio);
1407	}
1408
1409	ceph_wbc->processed_in_fbatch = i;
1410	}
1411
1412	static inline
1413	void ceph_shift_unused_folios_left(struct folio_batch *fbatch)
1414	{
1415	unsigned j, n = 0;
1416
1417	/* shift unused page to beginning of fbatch */
1418	for (j = 0; j < folio_batch_count(fbatch); j++) {
1419	if (!fbatch->folios[j])
1420	continue;
1421
1422	if (n < j) {
1423	fbatch->folios[n] = fbatch->folios[j];
1424	}
1425
1426	n++;
1427	}
1428
1429	fbatch->nr = n;
1430	}
1431
1432	static void ceph_undo_wrbuffer_claim(struct inode inode, struct folio folio)
1433	{
1434	struct ceph_snap_context *snapc = folio_detach_private(folio);
1435
1436	if (!snapc)
1437	return;
1438	ceph_put_wrbuffer_cap_refs(ceph_inode(inode), 1, snapc);
1439	ceph_put_snap_context(snapc);
1440	}
1441
1442	static
1443	int ceph_submit_write(struct address_space *mapping,
1444	struct writeback_control *wbc,
1445	struct ceph_writeback_ctl *ceph_wbc)
1446	{
1447	struct inode *inode = mapping->host;
1448	struct ceph_inode_info *ci = ceph_inode(inode);
1449	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1450	struct ceph_client *cl = fsc->client;
1451	struct ceph_vino vino = ceph_vino(inode);
1452	struct ceph_osd_request *req = NULL;
1453	struct page *page = NULL;
1454	bool caching = ceph_is_cache_enabled(inode);
1455	u64 offset;
1456	u64 len;
1457	unsigned i;
1458
1459	new_request:
1460	offset = ceph_fscrypt_page_offset(ceph_wbc->pages[0]);
1461	len = ceph_wbc->wsize;
1462
1463	req = ceph_osdc_new_request(&fsc->client->osdc,
1464	&ci->i_layout, vino,
1465	offset, &len, 0, ceph_wbc->num_ops,
1466	CEPH_OSD_OP_WRITE, CEPH_OSD_FLAG_WRITE,
1467	ceph_wbc->snapc, ceph_wbc->truncate_seq,
1468	ceph_wbc->truncate_size, false);
1469	if (IS_ERR(req)) {
1470	req = ceph_osdc_new_request(&fsc->client->osdc,
1471	&ci->i_layout, vino,
1472	offset, &len, 0,
1473	min(ceph_wbc->num_ops,
1474	CEPH_OSD_SLAB_OPS),
1475	CEPH_OSD_OP_WRITE,
1476	CEPH_OSD_FLAG_WRITE,
1477	ceph_wbc->snapc,
1478	ceph_wbc->truncate_seq,
1479	ceph_wbc->truncate_size,
1480	true);
1481	BUG_ON(IS_ERR(req));
1482	}
1483
1484	page = ceph_wbc->pages[ceph_wbc->locked_pages - 1];
1485	BUG_ON(len < ceph_fscrypt_page_offset(page) + thp_size(page) - offset);
1486
1487	if (!ceph_inc_osd_stopping_blocker(fsc->mdsc)) {
1488	for (i = 0; i < folio_batch_count(&ceph_wbc->fbatch); i++) {
1489	struct folio *folio = ceph_wbc->fbatch.folios[i];
1490
1491	if (!folio)
1492	continue;
1493
1494	page = &folio->page;
1495	redirty_page_for_writepage(wbc, page);
1496	unlock_page(page);
1497	}
1498
1499	for (i = 0; i < ceph_wbc->locked_pages; i++) {
1500	page = ceph_fscrypt_pagecache_page(ceph_wbc->pages[i]);
1501
1502	if (!page)
1503	continue;
1504
1505	ceph_undo_wrbuffer_claim(inode, page_folio(page));
1506	redirty_page_for_writepage(wbc, page);
1507	unlock_page(page);
1508	}
1509
1510	ceph_osdc_put_request(req);
1511	return -EIO;
1512	}
1513
1514	req->r_callback = writepages_finish;
1515	req->r_inode = inode;
1516
1517	/* Format the osd request message and submit the write */
1518	len = 0;
1519	ceph_wbc->data_pages = ceph_wbc->pages;
1520	ceph_wbc->op_idx = 0;
1521	for (i = 0; i < ceph_wbc->locked_pages; i++) {
1522	u64 cur_offset;
1523
1524	page = ceph_fscrypt_pagecache_page(ceph_wbc->pages[i]);
1525	cur_offset = page_offset(page);
1526
1527	/*
1528	* Discontinuity in page range? Ceph can handle that by just passing
1529	* multiple extents in the write op.
1530	*/
1531	if (offset + len != cur_offset) {
1532	/* If it's full, stop here */
1533	if (ceph_wbc->op_idx + 1 == req->r_num_ops)
1534	break;
1535
1536	/* Kick off an fscache write with what we have so far. */
1537	ceph_fscache_write_to_cache(inode, offset, len, caching);
1538
1539	/* Start a new extent */
1540	osd_req_op_extent_dup_last(req, ceph_wbc->op_idx,
1541	cur_offset - offset);
1542
1543	doutc(cl, "got pages at %llu~%llu\n", offset, len);
1544
1545	osd_req_op_extent_osd_data_pages(req, ceph_wbc->op_idx,
1546	ceph_wbc->data_pages,
1547	len, 0,
1548	ceph_wbc->from_pool,
1549	false);
1550	osd_req_op_extent_update(req, ceph_wbc->op_idx, len);
1551
1552	len = 0;
1553	offset = cur_offset;
1554	ceph_wbc->data_pages = ceph_wbc->pages + i;
1555	ceph_wbc->op_idx++;
1556	}
1557
1558	set_page_writeback(page);
1559
1560	if (caching)
1561	ceph_set_page_fscache(page);
1562
1563	len += thp_size(page);
1564	}
1565
1566	ceph_fscache_write_to_cache(inode, offset, len, caching);
1567
1568	if (ceph_wbc->size_stable) {
1569	len = min(len, ceph_wbc->i_size - offset);
1570	} else if (i == ceph_wbc->locked_pages) {
1571	/* writepages_finish() clears writeback pages
1572	* according to the data length, so make sure
1573	* data length covers all locked pages */
1574	u64 min_len = len + 1 - thp_size(page);
1575	len = get_writepages_data_length(inode,
1576	ceph_wbc->pages[i - 1],
1577	offset);
1578	len = max(len, min_len);
1579	}
1580
1581	if (IS_ENCRYPTED(inode))
1582	len = round_up(len, CEPH_FSCRYPT_BLOCK_SIZE);
1583
1584	doutc(cl, "got pages at %llu~%llu\n", offset, len);
1585
1586	if (IS_ENCRYPTED(inode) &&
1587	((offset \| len) & ~CEPH_FSCRYPT_BLOCK_MASK)) {
1588	pr_warn_client(cl,
1589	"bad encrypted write offset=%lld len=%llu\n",
1590	offset, len);
1591	}
1592
1593	osd_req_op_extent_osd_data_pages(req, ceph_wbc->op_idx,
1594	ceph_wbc->data_pages, len,
1595	0, ceph_wbc->from_pool, false);
1596	osd_req_op_extent_update(req, ceph_wbc->op_idx, len);
1597
1598	BUG_ON(ceph_wbc->op_idx + 1 != req->r_num_ops);
1599
1600	ceph_wbc->from_pool = false;
1601	if (i < ceph_wbc->locked_pages) {
1602	BUG_ON(ceph_wbc->num_ops <= req->r_num_ops);
1603	ceph_wbc->num_ops -= req->r_num_ops;
1604	ceph_wbc->locked_pages -= i;
1605
1606	/* allocate new pages array for next request */
1607	ceph_wbc->data_pages = ceph_wbc->pages;
1608	__ceph_allocate_page_array(ceph_wbc, ceph_wbc->locked_pages);
1609	memcpy(ceph_wbc->pages, ceph_wbc->data_pages + i,
1610	ceph_wbc->locked_pages * sizeof(*ceph_wbc->pages));
1611	memset(ceph_wbc->data_pages + i, 0,
1612	ceph_wbc->locked_pages * sizeof(*ceph_wbc->pages));
1613	} else {
1614	BUG_ON(ceph_wbc->num_ops != req->r_num_ops);
1615	/* request message now owns the pages array */
1616	ceph_wbc->pages = NULL;
1617	}
1618
1619	req->r_mtime = inode_get_mtime(inode);
1620	ceph_osdc_start_request(&fsc->client->osdc, req);
1621	req = NULL;
1622
1623	wbc->nr_to_write -= i;
1624	if (ceph_wbc->pages)
1625	goto new_request;
1626
1627	return 0;
1628	}
1629
1630	static
1631	void ceph_wait_until_current_writes_complete(struct address_space *mapping,
1632	struct writeback_control *wbc,
1633	struct ceph_writeback_ctl *ceph_wbc)
1634	{
1635	struct page *page;
1636	unsigned i, nr;
1637
1638	if (wbc->sync_mode != WB_SYNC_NONE &&
1639	ceph_wbc->start_index == 0 && /* all dirty pages were checked */
1640	!ceph_wbc->head_snapc) {
1641	ceph_wbc->index = 0;
1642
1643	while ((ceph_wbc->index <= ceph_wbc->end) &&
1644	(nr = filemap_get_folios_tag(mapping,
1645	&ceph_wbc->index,
1646	(pgoff_t)-1,
1647	PAGECACHE_TAG_WRITEBACK,
1648	&ceph_wbc->fbatch))) {
1649	for (i = 0; i < nr; i++) {
1650	page = &ceph_wbc->fbatch.folios[i]->page;
1651	if (page_snap_context(page) != ceph_wbc->snapc)
1652	continue;
1653	wait_on_page_writeback(page);
1654	}
1655
1656	folio_batch_release(&ceph_wbc->fbatch);
1657	cond_resched();
1658	}
1659	}
1660	}
1661
1662	/*
1663	* initiate async writeback
1664	*/
1665	static int ceph_writepages_start(struct address_space *mapping,
1666	struct writeback_control *wbc)
1667	{
1668	struct inode *inode = mapping->host;
1669	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
1670	struct ceph_client *cl = fsc->client;
1671	struct ceph_writeback_ctl ceph_wbc;
1672	int rc = 0;
1673
1674	if (wbc->sync_mode == WB_SYNC_NONE && fsc->write_congested)
1675	return 0;
1676
1677	doutc(cl, "%llx.%llx (mode=%s)\n", ceph_vinop(inode),
1678	wbc->sync_mode == WB_SYNC_NONE ? "NONE" :
1679	(wbc->sync_mode == WB_SYNC_ALL ? "ALL" : "HOLD"));
1680
1681	if (is_forced_umount(mapping)) {
1682	/* we're in a forced umount, don't write! */
1683	return -EIO;
1684	}
1685
1686	ceph_init_writeback_ctl(mapping, wbc, &ceph_wbc);
1687
1688	if (!ceph_inc_osd_stopping_blocker(fsc->mdsc)) {
1689	rc = -EIO;
1690	goto out;
1691	}
1692
1693	retry:
1694	rc = ceph_define_writeback_range(mapping, wbc, &ceph_wbc);
1695	if (rc == -ENODATA) {
1696	/* hmm, why does writepages get called when there
1697	is no dirty data? */
1698	rc = 0;
1699	goto dec_osd_stopping_blocker;
1700	}
1701
1702	if (wbc->sync_mode == WB_SYNC_ALL \|\| wbc->tagged_writepages)
1703	tag_pages_for_writeback(mapping, ceph_wbc.index, ceph_wbc.end);
1704
1705	while (!has_writeback_done(&ceph_wbc)) {
1706	BUG_ON(ceph_wbc.locked_pages);
1707	BUG_ON(ceph_wbc.pages);
1708
1709	ceph_wbc.max_pages = ceph_wbc.wsize >> PAGE_SHIFT;
1710
1711	get_more_pages:
1712	ceph_folio_batch_reinit(&ceph_wbc);
1713
1714	ceph_wbc.nr_folios = filemap_get_folios_tag(mapping,
1715	&ceph_wbc.index,
1716	ceph_wbc.end,
1717	ceph_wbc.tag,
1718	&ceph_wbc.fbatch);
1719	doutc(cl, "pagevec_lookup_range_tag for tag %#x got %d\n",
1720	ceph_wbc.tag, ceph_wbc.nr_folios);
1721
1722	if (!ceph_wbc.nr_folios && !ceph_wbc.locked_pages)
1723	break;
1724
1725	process_folio_batch:
1726	ceph_process_folio_batch(mapping, wbc, &ceph_wbc);
1727	ceph_shift_unused_folios_left(&ceph_wbc.fbatch);
1728
1729	/* did we get anything? */
1730	if (!ceph_wbc.locked_pages)
1731	goto release_folios;
1732
1733	if (ceph_wbc.processed_in_fbatch) {
1734	if (folio_batch_count(&ceph_wbc.fbatch) == 0 &&
1735	ceph_wbc.locked_pages < ceph_wbc.max_pages) {
1736	doutc(cl, "reached end fbatch, trying for more\n");
1737	goto get_more_pages;
1738	}
1739	}
1740
1741	rc = ceph_submit_write(mapping, wbc, &ceph_wbc);
1742	if (rc)
1743	goto release_folios;
1744
1745	ceph_wbc.locked_pages = 0;
1746	ceph_wbc.strip_unit_end = 0;
1747
1748	if (folio_batch_count(&ceph_wbc.fbatch) > 0) {
1749	ceph_wbc.nr_folios =
1750	folio_batch_count(&ceph_wbc.fbatch);
1751	goto process_folio_batch;
1752	}
1753
1754	/*
1755	* We stop writing back only if we are not doing
1756	* integrity sync. In case of integrity sync we have to
1757	* keep going until we have written all the pages
1758	* we tagged for writeback prior to entering this loop.
1759	*/
1760	if (wbc->nr_to_write <= 0 && wbc->sync_mode == WB_SYNC_NONE)
1761	ceph_wbc.done = true;
1762
1763	release_folios:
1764	doutc(cl, "folio_batch release on %d folios (%p)\n",
1765	(int)ceph_wbc.fbatch.nr,
1766	ceph_wbc.fbatch.nr ? ceph_wbc.fbatch.folios[0] : NULL);
1767	folio_batch_release(&ceph_wbc.fbatch);
1768	}
1769
1770	if (ceph_wbc.should_loop && !ceph_wbc.done) {
1771	/* more to do; loop back to beginning of file */
1772	doutc(cl, "looping back to beginning of file\n");
1773	/* OK even when start_index == 0 */
1774	ceph_wbc.end = ceph_wbc.start_index - 1;
1775
1776	/* to write dirty pages associated with next snapc,
1777	* we need to wait until current writes complete */
1778	ceph_wait_until_current_writes_complete(mapping, wbc, &ceph_wbc);
1779
1780	ceph_wbc.start_index = 0;
1781	ceph_wbc.index = 0;
1782	goto retry;
1783	}
1784
1785	if (wbc->range_cyclic \|\| (ceph_wbc.range_whole && wbc->nr_to_write > 0))
1786	mapping->writeback_index = ceph_wbc.index;
1787
1788	dec_osd_stopping_blocker:
1789	ceph_dec_osd_stopping_blocker(fsc->mdsc);
1790
1791	out:
1792	ceph_put_snap_context(ceph_wbc.last_snapc);
1793	doutc(cl, "%llx.%llx dend - startone, rc = %d\n", ceph_vinop(inode),
1794	rc);
1795
1796	return rc;
1797	}
1798
1799	/*
1800	* See if a given @snapc is either writeable, or already written.
1801	*/
1802	static int context_is_writeable_or_written(struct inode *inode,
1803	struct ceph_snap_context *snapc)
1804	{
1805	struct ceph_snap_context *oldest = get_oldest_context(inode, NULL, NULL);
1806	int ret = !oldest \|\| snapc->seq <= oldest->seq;
1807
1808	ceph_put_snap_context(oldest);
1809	return ret;
1810	}
1811
1812	/**
1813	* ceph_find_incompatible - find an incompatible context and return it
1814	* @folio: folio being dirtied
1815	*
1816	* We are only allowed to write into/dirty a folio if the folio is
1817	* clean, or already dirty within the same snap context. Returns a
1818	* conflicting context if there is one, NULL if there isn't, or a
1819	* negative error code on other errors.
1820	*
1821	* Must be called with folio lock held.
1822	*/
1823	static struct ceph_snap_context *
1824	ceph_find_incompatible(struct folio *folio)
1825	{
1826	struct inode *inode = folio->mapping->host;
1827	struct ceph_client *cl = ceph_inode_to_client(inode);
1828	struct ceph_inode_info *ci = ceph_inode(inode);
1829
1830	if (ceph_inode_is_shutdown(inode)) {
1831	doutc(cl, " %llx.%llx folio %p is shutdown\n",
1832	ceph_vinop(inode), folio);
1833	return ERR_PTR(-ESTALE);
1834	}
1835
1836	for (;;) {
1837	struct ceph_snap_context snapc, oldest;
1838
1839	folio_wait_writeback(folio);
1840
1841	snapc = page_snap_context(&folio->page);
1842	if (!snapc \|\| snapc == ci->i_head_snapc)
1843	break;
1844
1845	/*
1846	* this folio is already dirty in another (older) snap
1847	* context! is it writeable now?
1848	*/
1849	oldest = get_oldest_context(inode, NULL, NULL);
1850	if (snapc->seq > oldest->seq) {
1851	/* not writeable -- return it for the caller to deal with */
1852	ceph_put_snap_context(oldest);
1853	doutc(cl, " %llx.%llx folio %p snapc %p not current or oldest\n",
1854	ceph_vinop(inode), folio, snapc);
1855	return ceph_get_snap_context(snapc);
1856	}
1857	ceph_put_snap_context(oldest);
1858
1859	/* yay, writeable, do it now (without dropping folio lock) */
1860	doutc(cl, " %llx.%llx folio %p snapc %p not current, but oldest\n",
1861	ceph_vinop(inode), folio, snapc);
1862	if (folio_clear_dirty_for_io(folio)) {
1863	int r = write_folio_nounlock(folio, NULL);
1864	if (r < 0)
1865	return ERR_PTR(r);
1866	}
1867	}
1868	return NULL;
1869	}
1870
1871	static int ceph_netfs_check_write_begin(struct file *file, uoff_t pos, unsigned int len,
1872	struct folio foliop, void _fsdata)
1873	{
1874	struct inode *inode = file_inode(file);
1875	struct ceph_inode_info *ci = ceph_inode(inode);
1876	struct ceph_snap_context *snapc;
1877
1878	snapc = ceph_find_incompatible(*foliop);
1879	if (snapc) {
1880	int r;
1881
1882	folio_unlock(*foliop);
1883	folio_put(*foliop);
1884	*foliop = NULL;
1885	if (IS_ERR(snapc))
1886	return PTR_ERR(snapc);
1887
1888	ceph_queue_writeback(inode);
1889	r = wait_event_killable(ci->i_cap_wq,
1890	context_is_writeable_or_written(inode, snapc));
1891	ceph_put_snap_context(snapc);
1892	return r == 0 ? -EAGAIN : r;
1893	}
1894	return 0;
1895	}
1896
1897	/*
1898	* We are only allowed to write into/dirty the page if the page is
1899	* clean, or already dirty within the same snap context.
1900	*/
1901	static int ceph_write_begin(const struct kiocb *iocb,
1902	struct address_space *mapping,
1903	loff_t pos, unsigned len,
1904	struct folio foliop, void fsdata)
1905	{
1906	struct file *file = iocb->ki_filp;
1907	struct inode *inode = file_inode(file);
1908	struct ceph_inode_info *ci = ceph_inode(inode);
1909	int r;
1910
1911	r = netfs_write_begin(&ci->netfs, file, inode->i_mapping, pos, len, foliop, NULL);
1912	if (r < 0)
1913	return r;
1914
1915	folio_wait_private_2(foliop); / [DEPRECATED] */
1916	WARN_ON_ONCE(!folio_test_locked(*foliop));
1917	return 0;
1918	}
1919
1920	/*
1921	* we don't do anything in here that simple_write_end doesn't do
1922	* except adjust dirty page accounting
1923	*/
1924	static int ceph_write_end(const struct kiocb *iocb,
1925	struct address_space *mapping, loff_t pos,
1926	unsigned len, unsigned copied,
1927	struct folio folio, void fsdata)
1928	{
1929	struct file *file = iocb->ki_filp;
1930	struct inode *inode = file_inode(file);
1931	struct ceph_client *cl = ceph_inode_to_client(inode);
1932	bool check_cap = false;
1933
1934	doutc(cl, "%llx.%llx file %p folio %p %d~%d (%d)\n", ceph_vinop(inode),
1935	file, folio, (int)pos, (int)copied, (int)len);
1936
1937	if (!folio_test_uptodate(folio)) {
1938	/* just return that nothing was copied on a short copy */
1939	if (copied < len) {
1940	copied = 0;
1941	goto out;
1942	}
1943	folio_mark_uptodate(folio);
1944	}
1945
1946	/* did file size increase? */
1947	if (pos+copied > i_size_read(inode))
1948	check_cap = ceph_inode_set_size(inode, pos+copied);
1949
1950	folio_mark_dirty(folio);
1951
1952	out:
1953	folio_unlock(folio);
1954	folio_put(folio);
1955
1956	if (check_cap)
1957	ceph_check_caps(ceph_inode(inode), CHECK_CAPS_AUTHONLY);
1958
1959	return copied;
1960	}
1961
1962	const struct address_space_operations ceph_aops = {
1963	.read_folio = netfs_read_folio,
1964	.readahead = netfs_readahead,
1965	.writepages = ceph_writepages_start,
1966	.write_begin = ceph_write_begin,
1967	.write_end = ceph_write_end,
1968	.dirty_folio = ceph_dirty_folio,
1969	.invalidate_folio = ceph_invalidate_folio,
1970	.release_folio = netfs_release_folio,
1971	.direct_IO = noop_direct_IO,
1972	.migrate_folio = filemap_migrate_folio,
1973	};
1974
1975	static void ceph_block_sigs(sigset_t *oldset)
1976	{
1977	sigset_t mask;
1978	siginitsetinv(&mask, sigmask(SIGKILL));
1979	sigprocmask(SIG_BLOCK, &mask, oldset);
1980	}
1981
1982	static void ceph_restore_sigs(sigset_t *oldset)
1983	{
1984	sigprocmask(SIG_SETMASK, oldset, NULL);
1985	}
1986
1987	/*
1988	* vm ops
1989	*/
1990	static vm_fault_t ceph_filemap_fault(struct vm_fault *vmf)
1991	{
1992	struct vm_area_struct *vma = vmf->vma;
1993	struct inode *inode = file_inode(vma->vm_file);
1994	struct ceph_inode_info *ci = ceph_inode(inode);
1995	struct ceph_client *cl = ceph_inode_to_client(inode);
1996	struct ceph_file_info *fi = vma->vm_file->private_data;
1997	loff_t off = (loff_t)vmf->pgoff << PAGE_SHIFT;
1998	int want, got, err;
1999	sigset_t oldset;
2000	vm_fault_t ret = VM_FAULT_SIGBUS;
2001
2002	if (ceph_inode_is_shutdown(inode))
2003	return ret;
2004
2005	ceph_block_sigs(&oldset);
2006
2007	doutc(cl, "%llx.%llx %llu trying to get caps\n",
2008	ceph_vinop(inode), off);
2009	if (fi->fmode & CEPH_FILE_MODE_LAZY)
2010	want = CEPH_CAP_FILE_CACHE \| CEPH_CAP_FILE_LAZYIO;
2011	else
2012	want = CEPH_CAP_FILE_CACHE;
2013
2014	got = 0;
2015	err = ceph_get_caps(vma->vm_file, CEPH_CAP_FILE_RD, want, -1, &got);
2016	if (err < 0)
2017	goto out_restore;
2018
2019	doutc(cl, "%llx.%llx %llu got cap refs on %s\n", ceph_vinop(inode),
2020	off, ceph_cap_string(got));
2021
2022	if ((got & (CEPH_CAP_FILE_CACHE \| CEPH_CAP_FILE_LAZYIO)) \|\|
2023	!ceph_has_inline_data(ci)) {
2024	CEPH_DEFINE_RW_CONTEXT(rw_ctx, got);
2025	ceph_add_rw_context(fi, &rw_ctx);
2026	ret = filemap_fault(vmf);
2027	ceph_del_rw_context(fi, &rw_ctx);
2028	doutc(cl, "%llx.%llx %llu drop cap refs %s ret %x\n",
2029	ceph_vinop(inode), off, ceph_cap_string(got), ret);
2030	} else
2031	err = -EAGAIN;
2032
2033	ceph_put_cap_refs(ci, got);
2034
2035	if (err != -EAGAIN)
2036	goto out_restore;
2037
2038	/* read inline data */
2039	if (off >= PAGE_SIZE) {
2040	/* does not support inline data > PAGE_SIZE */
2041	ret = VM_FAULT_SIGBUS;
2042	} else {
2043	struct address_space *mapping = inode->i_mapping;
2044	struct page *page;
2045
2046	filemap_invalidate_lock_shared(mapping);
2047	page = find_or_create_page(mapping, 0,
2048	mapping_gfp_constraint(mapping, ~__GFP_FS));
2049	if (!page) {
2050	ret = VM_FAULT_OOM;
2051	goto out_inline;
2052	}
2053	err = __ceph_do_getattr(inode, page,
2054	CEPH_STAT_CAP_INLINE_DATA, true);
2055	if (err < 0 \|\| off >= i_size_read(inode)) {
2056	unlock_page(page);
2057	put_page(page);
2058	ret = vmf_error(err);
2059	goto out_inline;
2060	}
2061	if (err < PAGE_SIZE)
2062	zero_user_segment(page, err, PAGE_SIZE);
2063	else
2064	flush_dcache_page(page);
2065	SetPageUptodate(page);
2066	vmf->page = page;
2067	ret = VM_FAULT_MAJOR \| VM_FAULT_LOCKED;
2068	out_inline:
2069	filemap_invalidate_unlock_shared(mapping);
2070	doutc(cl, "%llx.%llx %llu read inline data ret %x\n",
2071	ceph_vinop(inode), off, ret);
2072	}
2073	out_restore:
2074	ceph_restore_sigs(&oldset);
2075	if (err < 0)
2076	ret = vmf_error(err);
2077
2078	return ret;
2079	}
2080
2081	static vm_fault_t ceph_page_mkwrite(struct vm_fault *vmf)
2082	{
2083	struct vm_area_struct *vma = vmf->vma;
2084	struct inode *inode = file_inode(vma->vm_file);
2085	struct ceph_client *cl = ceph_inode_to_client(inode);
2086	struct ceph_inode_info *ci = ceph_inode(inode);
2087	struct ceph_file_info *fi = vma->vm_file->private_data;
2088	struct ceph_cap_flush *prealloc_cf;
2089	struct folio *folio = page_folio(vmf->page);
2090	loff_t off = folio_pos(folio);
2091	loff_t size = i_size_read(inode);
2092	size_t len;
2093	int want, got, err;
2094	sigset_t oldset;
2095	vm_fault_t ret = VM_FAULT_SIGBUS;
2096
2097	if (ceph_inode_is_shutdown(inode))
2098	return ret;
2099
2100	prealloc_cf = ceph_alloc_cap_flush();
2101	if (!prealloc_cf)
2102	return VM_FAULT_OOM;
2103
2104	sb_start_pagefault(inode->i_sb);
2105	ceph_block_sigs(&oldset);
2106
2107	if (off + folio_size(folio) <= size)
2108	len = folio_size(folio);
2109	else
2110	len = offset_in_folio(folio, size);
2111
2112	doutc(cl, "%llx.%llx %llu~%zd getting caps i_size %llu\n",
2113	ceph_vinop(inode), off, len, size);
2114	if (fi->fmode & CEPH_FILE_MODE_LAZY)
2115	want = CEPH_CAP_FILE_BUFFER \| CEPH_CAP_FILE_LAZYIO;
2116	else
2117	want = CEPH_CAP_FILE_BUFFER;
2118
2119	got = 0;
2120	err = ceph_get_caps(vma->vm_file, CEPH_CAP_FILE_WR, want, off + len, &got);
2121	if (err < 0)
2122	goto out_free;
2123
2124	doutc(cl, "%llx.%llx %llu~%zd got cap refs on %s\n", ceph_vinop(inode),
2125	off, len, ceph_cap_string(got));
2126
2127	/* Update time before taking folio lock */
2128	file_update_time(vma->vm_file);
2129	inode_inc_iversion_raw(inode);
2130
2131	do {
2132	struct ceph_snap_context *snapc;
2133
2134	folio_lock(folio);
2135
2136	if (folio_mkwrite_check_truncate(folio, inode) < 0) {
2137	folio_unlock(folio);
2138	ret = VM_FAULT_NOPAGE;
2139	break;
2140	}
2141
2142	snapc = ceph_find_incompatible(folio);
2143	if (!snapc) {
2144	/* success. we'll keep the folio locked. */
2145	folio_mark_dirty(folio);
2146	ret = VM_FAULT_LOCKED;
2147	break;
2148	}
2149
2150	folio_unlock(folio);
2151
2152	if (IS_ERR(snapc)) {
2153	ret = VM_FAULT_SIGBUS;
2154	break;
2155	}
2156
2157	ceph_queue_writeback(inode);
2158	err = wait_event_killable(ci->i_cap_wq,
2159	context_is_writeable_or_written(inode, snapc));
2160	ceph_put_snap_context(snapc);
2161	} while (err == 0);
2162
2163	if (ret == VM_FAULT_LOCKED) {
2164	int dirty;
2165	spin_lock(&ci->i_ceph_lock);
2166	dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
2167	&prealloc_cf);
2168	spin_unlock(&ci->i_ceph_lock);
2169	if (dirty)
2170	__mark_inode_dirty(inode, dirty);
2171	}
2172
2173	doutc(cl, "%llx.%llx %llu~%zd dropping cap refs on %s ret %x\n",
2174	ceph_vinop(inode), off, len, ceph_cap_string(got), ret);
2175	ceph_put_cap_refs_async(ci, got);
2176	out_free:
2177	ceph_restore_sigs(&oldset);
2178	sb_end_pagefault(inode->i_sb);
2179	ceph_free_cap_flush(prealloc_cf);
2180	if (err < 0)
2181	ret = vmf_error(err);
2182	return ret;
2183	}
2184
2185	void ceph_fill_inline_data(struct inode inode, struct page locked_page,
2186	char *data, size_t len)
2187	{
2188	struct ceph_client *cl = ceph_inode_to_client(inode);
2189	struct address_space *mapping = inode->i_mapping;
2190	struct page *page;
2191
2192	if (locked_page) {
2193	page = locked_page;
2194	} else {
2195	if (i_size_read(inode) == 0)
2196	return;
2197	page = find_or_create_page(mapping, 0,
2198	mapping_gfp_constraint(mapping,
2199	~__GFP_FS));
2200	if (!page)
2201	return;
2202	if (PageUptodate(page)) {
2203	unlock_page(page);
2204	put_page(page);
2205	return;
2206	}
2207	}
2208
2209	doutc(cl, "%p %llx.%llx len %zu locked_page %p\n", inode,
2210	ceph_vinop(inode), len, locked_page);
2211
2212	if (len > 0) {
2213	void *kaddr = kmap_atomic(page);
2214	memcpy(kaddr, data, len);
2215	kunmap_atomic(kaddr);
2216	}
2217
2218	if (page != locked_page) {
2219	if (len < PAGE_SIZE)
2220	zero_user_segment(page, len, PAGE_SIZE);
2221	else
2222	flush_dcache_page(page);
2223
2224	SetPageUptodate(page);
2225	unlock_page(page);
2226	put_page(page);
2227	}
2228	}
2229
2230	int ceph_uninline_data(struct file *file)
2231	{
2232	struct inode *inode = file_inode(file);
2233	struct ceph_inode_info *ci = ceph_inode(inode);
2234	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
2235	struct ceph_client *cl = fsc->client;
2236	struct ceph_osd_request *req = NULL;
2237	struct ceph_cap_flush *prealloc_cf = NULL;
2238	struct folio *folio = NULL;
2239	struct ceph_snap_context *snapc = NULL;
2240	u64 inline_version = CEPH_INLINE_NONE;
2241	struct page *pages[1];
2242	int err = 0;
2243	u64 len;
2244
2245	spin_lock(&ci->i_ceph_lock);
2246	inline_version = ci->i_inline_version;
2247	spin_unlock(&ci->i_ceph_lock);
2248
2249	doutc(cl, "%llx.%llx inline_version %llu\n", ceph_vinop(inode),
2250	inline_version);
2251
2252	if (ceph_inode_is_shutdown(inode)) {
2253	err = -EIO;
2254	goto out;
2255	}
2256
2257	if (inline_version == CEPH_INLINE_NONE)
2258	return 0;
2259
2260	prealloc_cf = ceph_alloc_cap_flush();
2261	if (!prealloc_cf)
2262	return -ENOMEM;
2263
2264	if (inline_version == 1) /* initial version, no data */
2265	goto out_uninline;
2266
2267	down_read(&fsc->mdsc->snap_rwsem);
2268	spin_lock(&ci->i_ceph_lock);
2269	if (__ceph_have_pending_cap_snap(ci)) {
2270	struct ceph_cap_snap *capsnap =
2271	list_last_entry(&ci->i_cap_snaps,
2272	struct ceph_cap_snap,
2273	ci_item);
2274	snapc = ceph_get_snap_context(capsnap->context);
2275	} else {
2276	if (!ci->i_head_snapc) {
2277	ci->i_head_snapc = ceph_get_snap_context(
2278	ci->i_snap_realm->cached_context);
2279	}
2280	snapc = ceph_get_snap_context(ci->i_head_snapc);
2281	}
2282	spin_unlock(&ci->i_ceph_lock);
2283	up_read(&fsc->mdsc->snap_rwsem);
2284
2285	folio = read_mapping_folio(inode->i_mapping, 0, file);
2286	if (IS_ERR(folio)) {
2287	err = PTR_ERR(folio);
2288	goto out;
2289	}
2290
2291	folio_lock(folio);
2292
2293	len = i_size_read(inode);
2294	if (len > folio_size(folio))
2295	len = folio_size(folio);
2296
2297	req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
2298	ceph_vino(inode), 0, &len, 0, 1,
2299	CEPH_OSD_OP_CREATE, CEPH_OSD_FLAG_WRITE,
2300	snapc, 0, 0, false);
2301	if (IS_ERR(req)) {
2302	err = PTR_ERR(req);
2303	goto out_unlock;
2304	}
2305
2306	req->r_mtime = inode_get_mtime(inode);
2307	ceph_osdc_start_request(&fsc->client->osdc, req);
2308	err = ceph_osdc_wait_request(&fsc->client->osdc, req);
2309	ceph_osdc_put_request(req);
2310	if (err < 0)
2311	goto out_unlock;
2312
2313	req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
2314	ceph_vino(inode), 0, &len, 1, 3,
2315	CEPH_OSD_OP_WRITE, CEPH_OSD_FLAG_WRITE,
2316	snapc, ci->i_truncate_seq,
2317	ci->i_truncate_size, false);
2318	if (IS_ERR(req)) {
2319	err = PTR_ERR(req);
2320	goto out_unlock;
2321	}
2322
2323	pages[0] = folio_page(folio, 0);
2324	osd_req_op_extent_osd_data_pages(req, 1, pages, len, 0, false, false);
2325
2326	{
2327	__le64 xattr_buf = cpu_to_le64(inline_version);
2328	err = osd_req_op_xattr_init(req, 0, CEPH_OSD_OP_CMPXATTR,
2329	"inline_version", &xattr_buf,
2330	sizeof(xattr_buf),
2331	CEPH_OSD_CMPXATTR_OP_GT,
2332	CEPH_OSD_CMPXATTR_MODE_U64);
2333	if (err)
2334	goto out_put_req;
2335	}
2336
2337	{
2338	char xattr_buf[32];
2339	int xattr_len = snprintf(xattr_buf, sizeof(xattr_buf),
2340	"%llu", inline_version);
2341	err = osd_req_op_xattr_init(req, 2, CEPH_OSD_OP_SETXATTR,
2342	"inline_version",
2343	xattr_buf, xattr_len, 0, 0);
2344	if (err)
2345	goto out_put_req;
2346	}
2347
2348	req->r_mtime = inode_get_mtime(inode);
2349	ceph_osdc_start_request(&fsc->client->osdc, req);
2350	err = ceph_osdc_wait_request(&fsc->client->osdc, req);
2351
2352	ceph_update_write_metrics(&fsc->mdsc->metric, req->r_start_latency,
2353	req->r_end_latency, len, err);
2354
2355	out_uninline:
2356	if (!err) {
2357	int dirty;
2358
2359	/* Set to CAP_INLINE_NONE and dirty the caps */
2360	down_read(&fsc->mdsc->snap_rwsem);
2361	spin_lock(&ci->i_ceph_lock);
2362	ci->i_inline_version = CEPH_INLINE_NONE;
2363	dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR, &prealloc_cf);
2364	spin_unlock(&ci->i_ceph_lock);
2365	up_read(&fsc->mdsc->snap_rwsem);
2366	if (dirty)
2367	__mark_inode_dirty(inode, dirty);
2368	}
2369	out_put_req:
2370	ceph_osdc_put_request(req);
2371	if (err == -ECANCELED)
2372	err = 0;
2373	out_unlock:
2374	if (folio) {
2375	folio_unlock(folio);
2376	folio_put(folio);
2377	}
2378	out:
2379	ceph_put_snap_context(snapc);
2380	ceph_free_cap_flush(prealloc_cf);
2381	doutc(cl, "%llx.%llx inline_version %llu = %d\n",
2382	ceph_vinop(inode), inline_version, err);
2383	return err;
2384	}
2385
2386	static const struct vm_operations_struct ceph_vmops = {
2387	.fault = ceph_filemap_fault,
2388	.page_mkwrite = ceph_page_mkwrite,
2389	};
2390
2391	int ceph_mmap_prepare(struct vm_area_desc *desc)
2392	{
2393	struct address_space *mapping = desc->file->f_mapping;
2394
2395	if (!mapping->a_ops->read_folio)
2396	return -ENOEXEC;
2397	desc->vm_ops = &ceph_vmops;
2398	return 0;
2399	}
2400
2401	enum {
2402	POOL_READ = 1,
2403	POOL_WRITE = 2,
2404	};
2405
2406	static int __ceph_pool_perm_get(struct ceph_inode_info *ci,
2407	s64 pool, struct ceph_string *pool_ns)
2408	{
2409	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(&ci->netfs.inode);
2410	struct ceph_mds_client *mdsc = fsc->mdsc;
2411	struct ceph_client *cl = fsc->client;
2412	struct ceph_osd_request rd_req = NULL, wr_req = NULL;
2413	struct rb_node *p, parent;
2414	struct ceph_pool_perm *perm;
2415	struct page **pages;
2416	size_t pool_ns_len;
2417	int err = 0, err2 = 0, have = 0;
2418
2419	down_read(&mdsc->pool_perm_rwsem);
2420	p = &mdsc->pool_perm_tree.rb_node;
2421	while (*p) {
2422	perm = rb_entry(*p, struct ceph_pool_perm, node);
2423	if (pool < perm->pool)
2424	p = &(*p)->rb_left;
2425	else if (pool > perm->pool)
2426	p = &(*p)->rb_right;
2427	else {
2428	int ret = ceph_compare_string(pool_ns,
2429	perm->pool_ns,
2430	perm->pool_ns_len);
2431	if (ret < 0)
2432	p = &(*p)->rb_left;
2433	else if (ret > 0)
2434	p = &(*p)->rb_right;
2435	else {
2436	have = perm->perm;
2437	break;
2438	}
2439	}
2440	}
2441	up_read(&mdsc->pool_perm_rwsem);
2442	if (*p)
2443	goto out;
2444
2445	if (pool_ns)
2446	doutc(cl, "pool %lld ns %.*s no perm cached\n", pool,
2447	(int)pool_ns->len, pool_ns->str);
2448	else
2449	doutc(cl, "pool %lld no perm cached\n", pool);
2450
2451	down_write(&mdsc->pool_perm_rwsem);
2452	p = &mdsc->pool_perm_tree.rb_node;
2453	parent = NULL;
2454	while (*p) {
2455	parent = *p;
2456	perm = rb_entry(parent, struct ceph_pool_perm, node);
2457	if (pool < perm->pool)
2458	p = &(*p)->rb_left;
2459	else if (pool > perm->pool)
2460	p = &(*p)->rb_right;
2461	else {
2462	int ret = ceph_compare_string(pool_ns,
2463	perm->pool_ns,
2464	perm->pool_ns_len);
2465	if (ret < 0)
2466	p = &(*p)->rb_left;
2467	else if (ret > 0)
2468	p = &(*p)->rb_right;
2469	else {
2470	have = perm->perm;
2471	break;
2472	}
2473	}
2474	}
2475	if (*p) {
2476	up_write(&mdsc->pool_perm_rwsem);
2477	goto out;
2478	}
2479
2480	rd_req = ceph_osdc_alloc_request(&fsc->client->osdc, NULL,
2481	1, false, GFP_KERNEL);
2482	if (!rd_req) {
2483	err = -ENOMEM;
2484	goto out_unlock;
2485	}
2486
2487	rd_req->r_flags = CEPH_OSD_FLAG_READ;
2488	osd_req_op_init(rd_req, 0, CEPH_OSD_OP_STAT, 0);
2489	rd_req->r_base_oloc.pool = pool;
2490	if (pool_ns)
2491	rd_req->r_base_oloc.pool_ns = ceph_get_string(pool_ns);
2492	ceph_oid_printf(&rd_req->r_base_oid, "%llx.00000000", ci->i_vino.ino);
2493
2494	err = ceph_osdc_alloc_messages(rd_req, GFP_KERNEL);
2495	if (err)
2496	goto out_unlock;
2497
2498	wr_req = ceph_osdc_alloc_request(&fsc->client->osdc, NULL,
2499	1, false, GFP_KERNEL);
2500	if (!wr_req) {
2501	err = -ENOMEM;
2502	goto out_unlock;
2503	}
2504
2505	wr_req->r_flags = CEPH_OSD_FLAG_WRITE;
2506	osd_req_op_init(wr_req, 0, CEPH_OSD_OP_CREATE, CEPH_OSD_OP_FLAG_EXCL);
2507	ceph_oloc_copy(&wr_req->r_base_oloc, &rd_req->r_base_oloc);
2508	ceph_oid_copy(&wr_req->r_base_oid, &rd_req->r_base_oid);
2509
2510	err = ceph_osdc_alloc_messages(wr_req, GFP_KERNEL);
2511	if (err)
2512	goto out_unlock;
2513
2514	/* one page should be large enough for STAT data */
2515	pages = ceph_alloc_page_vector(1, GFP_KERNEL);
2516	if (IS_ERR(pages)) {
2517	err = PTR_ERR(pages);
2518	goto out_unlock;
2519	}
2520
2521	osd_req_op_raw_data_in_pages(rd_req, 0, pages, PAGE_SIZE,
2522	0, false, true);
2523	ceph_osdc_start_request(&fsc->client->osdc, rd_req);
2524
2525	wr_req->r_mtime = inode_get_mtime(&ci->netfs.inode);
2526	ceph_osdc_start_request(&fsc->client->osdc, wr_req);
2527
2528	err = ceph_osdc_wait_request(&fsc->client->osdc, rd_req);
2529	err2 = ceph_osdc_wait_request(&fsc->client->osdc, wr_req);
2530
2531	if (err >= 0 \|\| err == -ENOENT)
2532	have \|= POOL_READ;
2533	else if (err != -EPERM) {
2534	if (err == -EBLOCKLISTED)
2535	fsc->blocklisted = true;
2536	goto out_unlock;
2537	}
2538
2539	if (err2 == 0 \|\| err2 == -EEXIST)
2540	have \|= POOL_WRITE;
2541	else if (err2 != -EPERM) {
2542	if (err2 == -EBLOCKLISTED)
2543	fsc->blocklisted = true;
2544	err = err2;
2545	goto out_unlock;
2546	}
2547
2548	pool_ns_len = pool_ns ? pool_ns->len : 0;
2549	perm = kmalloc_flex(*perm, pool_ns, pool_ns_len + 1);
2550	if (!perm) {
2551	err = -ENOMEM;
2552	goto out_unlock;
2553	}
2554
2555	perm->pool = pool;
2556	perm->perm = have;
2557	perm->pool_ns_len = pool_ns_len;
2558	if (pool_ns_len > 0)
2559	memcpy(perm->pool_ns, pool_ns->str, pool_ns_len);
2560	perm->pool_ns[pool_ns_len] = 0;
2561
2562	rb_link_node(&perm->node, parent, p);
2563	rb_insert_color(&perm->node, &mdsc->pool_perm_tree);
2564	err = 0;
2565	out_unlock:
2566	up_write(&mdsc->pool_perm_rwsem);
2567
2568	ceph_osdc_put_request(rd_req);
2569	ceph_osdc_put_request(wr_req);
2570	out:
2571	if (!err)
2572	err = have;
2573	if (pool_ns)
2574	doutc(cl, "pool %lld ns %.*s result = %d\n", pool,
2575	(int)pool_ns->len, pool_ns->str, err);
2576	else
2577	doutc(cl, "pool %lld result = %d\n", pool, err);
2578	return err;
2579	}
2580
2581	int ceph_pool_perm_check(struct inode *inode, int need)
2582	{
2583	struct ceph_client *cl = ceph_inode_to_client(inode);
2584	struct ceph_inode_info *ci = ceph_inode(inode);
2585	struct ceph_string *pool_ns;
2586	s64 pool;
2587	int ret;
2588	unsigned long flags;
2589
2590	/* Only need to do this for regular files */
2591	if (!S_ISREG(inode->i_mode))
2592	return 0;
2593
2594	if (ci->i_vino.snap != CEPH_NOSNAP) {
2595	/*
2596	* Pool permission check needs to write to the first object.
2597	* But for snapshot, head of the first object may have already
2598	* been deleted. Skip check to avoid creating orphan object.
2599	*/
2600	return 0;
2601	}
2602
2603	if (ceph_test_mount_opt(ceph_inode_to_fs_client(inode),
2604	NOPOOLPERM))
2605	return 0;
2606
2607	spin_lock(&ci->i_ceph_lock);
2608	flags = ci->i_ceph_flags;
2609	pool = ci->i_layout.pool_id;
2610	spin_unlock(&ci->i_ceph_lock);
2611	check:
2612	if (flags & CEPH_I_POOL_PERM) {
2613	if ((need & CEPH_CAP_FILE_RD) && !(flags & CEPH_I_POOL_RD)) {
2614	doutc(cl, "pool %lld no read perm\n", pool);
2615	return -EPERM;
2616	}
2617	if ((need & CEPH_CAP_FILE_WR) && !(flags & CEPH_I_POOL_WR)) {
2618	doutc(cl, "pool %lld no write perm\n", pool);
2619	return -EPERM;
2620	}
2621	return 0;
2622	}
2623
2624	pool_ns = ceph_try_get_string(ci->i_layout.pool_ns);
2625	ret = __ceph_pool_perm_get(ci, pool, pool_ns);
2626	ceph_put_string(pool_ns);
2627	if (ret < 0)
2628	return ret;
2629
2630	spin_lock(&ci->i_ceph_lock);
2631	if (pool == ci->i_layout.pool_id &&
2632	pool_ns == rcu_dereference_raw(ci->i_layout.pool_ns)) {
2633	set_bit(CEPH_I_POOL_PERM_BIT, &ci->i_ceph_flags);
2634	if (ret & POOL_READ)
2635	set_bit(CEPH_I_POOL_RD_BIT, &ci->i_ceph_flags);
2636	if (ret & POOL_WRITE)
2637	set_bit(CEPH_I_POOL_WR_BIT, &ci->i_ceph_flags);
2638	} else {
2639	pool = ci->i_layout.pool_id;
2640	}
2641	/* Re-read flags under the lock so check: sees the updated bits. */
2642	flags = ci->i_ceph_flags;
2643	spin_unlock(&ci->i_ceph_lock);
2644	goto check;
2645	}
2646
2647	void ceph_pool_perm_destroy(struct ceph_mds_client *mdsc)
2648	{
2649	struct ceph_pool_perm *perm;
2650	struct rb_node *n;
2651
2652	while (!RB_EMPTY_ROOT(&mdsc->pool_perm_tree)) {
2653	n = rb_first(&mdsc->pool_perm_tree);
2654	perm = rb_entry(n, struct ceph_pool_perm, node);
2655	rb_erase(n, &mdsc->pool_perm_tree);
2656	kfree(perm);
2657	}
2658	}