1	// SPDX-License-Identifier: GPL-2.0
2	#include <linux/ceph/ceph_debug.h>
3
4	#include <linux/fs.h>
5	#include <linux/kernel.h>
6	#include <linux/sched/signal.h>
7	#include <linux/slab.h>
8	#include <linux/vmalloc.h>
9	#include <linux/wait.h>
10	#include <linux/writeback.h>
11	#include <linux/iversion.h>
12	#include <linux/filelock.h>
13	#include <linux/jiffies.h>
14
15	#include "super.h"
16	#include "mds_client.h"
17	#include "cache.h"
18	#include "crypto.h"
19	#include <linux/ceph/decode.h>
20	#include <linux/ceph/messenger.h>
21	#include <trace/events/ceph.h>
22
23	/*
24	* Capability management
25	*
26	* The Ceph metadata servers control client access to inode metadata
27	* and file data by issuing capabilities, granting clients permission
28	* to read and/or write both inode field and file data to OSDs
29	* (storage nodes). Each capability consists of a set of bits
30	* indicating which operations are allowed.
31	*
32	* If the client holds a *_SHARED cap, the client has a coherent value
33	* that can be safely read from the cached inode.
34	*
35	* In the case of a *_EXCL (exclusive) or FILE_WR capabilities, the
36	* client is allowed to change inode attributes (e.g., file size,
37	* mtime), note its dirty state in the ceph_cap, and asynchronously
38	* flush that metadata change to the MDS.
39	*
40	* In the event of a conflicting operation (perhaps by another
41	* client), the MDS will revoke the conflicting client capabilities.
42	*
43	* In order for a client to cache an inode, it must hold a capability
44	* with at least one MDS server. When inodes are released, release
45	* notifications are batched and periodically sent en masse to the MDS
46	* cluster to release server state.
47	*/
48
49	static u64 __get_oldest_flush_tid(struct ceph_mds_client *mdsc);
50	static void __kick_flushing_caps(struct ceph_mds_client *mdsc,
51	struct ceph_mds_session *session,
52	struct ceph_inode_info *ci,
53	u64 oldest_flush_tid);
54
55	/*
56	* Generate readable cap strings for debugging output.
57	*/
58	#define MAX_CAP_STR 20
59	static char cap_str[MAX_CAP_STR][40];
60	static DEFINE_SPINLOCK(cap_str_lock);
61	static int last_cap_str;
62
63	static char *gcap_string(char *s, int c)
64	{
65	if (c & CEPH_CAP_GSHARED)
66	*s++ = 's';
67	if (c & CEPH_CAP_GEXCL)
68	*s++ = 'x';
69	if (c & CEPH_CAP_GCACHE)
70	*s++ = 'c';
71	if (c & CEPH_CAP_GRD)
72	*s++ = 'r';
73	if (c & CEPH_CAP_GWR)
74	*s++ = 'w';
75	if (c & CEPH_CAP_GBUFFER)
76	*s++ = 'b';
77	if (c & CEPH_CAP_GWREXTEND)
78	*s++ = 'a';
79	if (c & CEPH_CAP_GLAZYIO)
80	*s++ = 'l';
81	return s;
82	}
83
84	const char *ceph_cap_string(int caps)
85	{
86	int i;
87	char *s;
88	int c;
89
90	spin_lock(lock: &cap_str_lock);
91	i = last_cap_str++;
92	if (last_cap_str == MAX_CAP_STR)
93	last_cap_str = 0;
94	spin_unlock(lock: &cap_str_lock);
95
96	s = cap_str[i];
97
98	if (caps & CEPH_CAP_PIN)
99	*s++ = 'p';
100
101	c = (caps >> CEPH_CAP_SAUTH) & 3;
102	if (c) {
103	*s++ = 'A';
104	s = gcap_string(s, c);
105	}
106
107	c = (caps >> CEPH_CAP_SLINK) & 3;
108	if (c) {
109	*s++ = 'L';
110	s = gcap_string(s, c);
111	}
112
113	c = (caps >> CEPH_CAP_SXATTR) & 3;
114	if (c) {
115	*s++ = 'X';
116	s = gcap_string(s, c);
117	}
118
119	c = caps >> CEPH_CAP_SFILE;
120	if (c) {
121	*s++ = 'F';
122	s = gcap_string(s, c);
123	}
124
125	if (s == cap_str[i])
126	*s++ = '-';
127	*s = 0;
128	return cap_str[i];
129	}
130
131	void ceph_caps_init(struct ceph_mds_client *mdsc)
132	{
133	INIT_LIST_HEAD(list: &mdsc->caps_list);
134	spin_lock_init(&mdsc->caps_list_lock);
135	}
136
137	void ceph_caps_finalize(struct ceph_mds_client *mdsc)
138	{
139	struct ceph_cap *cap;
140
141	spin_lock(lock: &mdsc->caps_list_lock);
142	while (!list_empty(head: &mdsc->caps_list)) {
143	cap = list_first_entry(&mdsc->caps_list,
144	struct ceph_cap, caps_item);
145	list_del(entry: &cap->caps_item);
146	kmem_cache_free(s: ceph_cap_cachep, objp: cap);
147	}
148	mdsc->caps_total_count = 0;
149	mdsc->caps_avail_count = 0;
150	mdsc->caps_use_count = 0;
151	mdsc->caps_reserve_count = 0;
152	mdsc->caps_min_count = 0;
153	spin_unlock(lock: &mdsc->caps_list_lock);
154	}
155
156	void ceph_adjust_caps_max_min(struct ceph_mds_client *mdsc,
157	struct ceph_mount_options *fsopt)
158	{
159	spin_lock(lock: &mdsc->caps_list_lock);
160	mdsc->caps_min_count = fsopt->max_readdir;
161	if (mdsc->caps_min_count < 1024)
162	mdsc->caps_min_count = 1024;
163	mdsc->caps_use_max = fsopt->caps_max;
164	if (mdsc->caps_use_max > 0 &&
165	mdsc->caps_use_max < mdsc->caps_min_count)
166	mdsc->caps_use_max = mdsc->caps_min_count;
167	spin_unlock(lock: &mdsc->caps_list_lock);
168	}
169
170	static void __ceph_unreserve_caps(struct ceph_mds_client *mdsc, int nr_caps)
171	{
172	struct ceph_cap *cap;
173	int i;
174
175	if (nr_caps) {
176	BUG_ON(mdsc->caps_reserve_count < nr_caps);
177	mdsc->caps_reserve_count -= nr_caps;
178	if (mdsc->caps_avail_count >=
179	mdsc->caps_reserve_count + mdsc->caps_min_count) {
180	mdsc->caps_total_count -= nr_caps;
181	for (i = 0; i < nr_caps; i++) {
182	cap = list_first_entry(&mdsc->caps_list,
183	struct ceph_cap, caps_item);
184	list_del(entry: &cap->caps_item);
185	kmem_cache_free(s: ceph_cap_cachep, objp: cap);
186	}
187	} else {
188	mdsc->caps_avail_count += nr_caps;
189	}
190
191	doutc(mdsc->fsc->client,
192	"caps %d = %d used + %d resv + %d avail\n",
193	mdsc->caps_total_count, mdsc->caps_use_count,
194	mdsc->caps_reserve_count, mdsc->caps_avail_count);
195	BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
196	mdsc->caps_reserve_count +
197	mdsc->caps_avail_count);
198	}
199	}
200
201	/*
202	* Called under mdsc->mutex.
203	*/
204	int ceph_reserve_caps(struct ceph_mds_client *mdsc,
205	struct ceph_cap_reservation *ctx, int need)
206	{
207	struct ceph_client *cl = mdsc->fsc->client;
208	int i, j;
209	struct ceph_cap *cap;
210	int have;
211	int alloc = 0;
212	int max_caps;
213	int err = 0;
214	bool trimmed = false;
215	struct ceph_mds_session *s;
216	LIST_HEAD(newcaps);
217
218	doutc(cl, "ctx=%p need=%d\n", ctx, need);
219
220	/* first reserve any caps that are already allocated */
221	spin_lock(lock: &mdsc->caps_list_lock);
222	if (mdsc->caps_avail_count >= need)
223	have = need;
224	else
225	have = mdsc->caps_avail_count;
226	mdsc->caps_avail_count -= have;
227	mdsc->caps_reserve_count += have;
228	BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
229	mdsc->caps_reserve_count +
230	mdsc->caps_avail_count);
231	spin_unlock(lock: &mdsc->caps_list_lock);
232
233	for (i = have; i < need; ) {
234	cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
235	if (cap) {
236	list_add(new: &cap->caps_item, head: &newcaps);
237	alloc++;
238	i++;
239	continue;
240	}
241
242	if (!trimmed) {
243	for (j = 0; j < mdsc->max_sessions; j++) {
244	s = __ceph_lookup_mds_session(mdsc, mds: j);
245	if (!s)
246	continue;
247	mutex_unlock(lock: &mdsc->mutex);
248
249	mutex_lock(&s->s_mutex);
250	max_caps = s->s_nr_caps - (need - i);
251	ceph_trim_caps(mdsc, session: s, max_caps);
252	mutex_unlock(lock: &s->s_mutex);
253
254	ceph_put_mds_session(s);
255	mutex_lock(&mdsc->mutex);
256	}
257	trimmed = true;
258
259	spin_lock(lock: &mdsc->caps_list_lock);
260	if (mdsc->caps_avail_count) {
261	int more_have;
262	if (mdsc->caps_avail_count >= need - i)
263	more_have = need - i;
264	else
265	more_have = mdsc->caps_avail_count;
266
267	i += more_have;
268	have += more_have;
269	mdsc->caps_avail_count -= more_have;
270	mdsc->caps_reserve_count += more_have;
271
272	}
273	spin_unlock(lock: &mdsc->caps_list_lock);
274
275	continue;
276	}
277
278	pr_warn_client(cl, "ctx=%p ENOMEM need=%d got=%d\n", ctx, need,
279	have + alloc);
280	err = -ENOMEM;
281	break;
282	}
283
284	if (!err) {
285	BUG_ON(have + alloc != need);
286	ctx->count = need;
287	ctx->used = 0;
288	}
289
290	spin_lock(lock: &mdsc->caps_list_lock);
291	mdsc->caps_total_count += alloc;
292	mdsc->caps_reserve_count += alloc;
293	list_splice(list: &newcaps, head: &mdsc->caps_list);
294
295	BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
296	mdsc->caps_reserve_count +
297	mdsc->caps_avail_count);
298
299	if (err)
300	__ceph_unreserve_caps(mdsc, nr_caps: have + alloc);
301
302	spin_unlock(lock: &mdsc->caps_list_lock);
303
304	doutc(cl, "ctx=%p %d = %d used + %d resv + %d avail\n", ctx,
305	mdsc->caps_total_count, mdsc->caps_use_count,
306	mdsc->caps_reserve_count, mdsc->caps_avail_count);
307	return err;
308	}
309
310	void ceph_unreserve_caps(struct ceph_mds_client *mdsc,
311	struct ceph_cap_reservation *ctx)
312	{
313	struct ceph_client *cl = mdsc->fsc->client;
314	bool reclaim = false;
315	if (!ctx->count)
316	return;
317
318	doutc(cl, "ctx=%p count=%d\n", ctx, ctx->count);
319	spin_lock(lock: &mdsc->caps_list_lock);
320	__ceph_unreserve_caps(mdsc, nr_caps: ctx->count);
321	ctx->count = 0;
322
323	if (mdsc->caps_use_max > 0 &&
324	mdsc->caps_use_count > mdsc->caps_use_max)
325	reclaim = true;
326	spin_unlock(lock: &mdsc->caps_list_lock);
327
328	if (reclaim)
329	ceph_reclaim_caps_nr(mdsc, nr: ctx->used);
330	}
331
332	struct ceph_cap *ceph_get_cap(struct ceph_mds_client *mdsc,
333	struct ceph_cap_reservation *ctx)
334	{
335	struct ceph_client *cl = mdsc->fsc->client;
336	struct ceph_cap *cap = NULL;
337
338	/* temporary, until we do something about cap import/export */
339	if (!ctx) {
340	cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
341	if (cap) {
342	spin_lock(lock: &mdsc->caps_list_lock);
343	mdsc->caps_use_count++;
344	mdsc->caps_total_count++;
345	spin_unlock(lock: &mdsc->caps_list_lock);
346	} else {
347	spin_lock(lock: &mdsc->caps_list_lock);
348	if (mdsc->caps_avail_count) {
349	BUG_ON(list_empty(&mdsc->caps_list));
350
351	mdsc->caps_avail_count--;
352	mdsc->caps_use_count++;
353	cap = list_first_entry(&mdsc->caps_list,
354	struct ceph_cap, caps_item);
355	list_del(entry: &cap->caps_item);
356
357	BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
358	mdsc->caps_reserve_count + mdsc->caps_avail_count);
359	}
360	spin_unlock(lock: &mdsc->caps_list_lock);
361	}
362
363	return cap;
364	}
365
366	spin_lock(lock: &mdsc->caps_list_lock);
367	doutc(cl, "ctx=%p (%d) %d = %d used + %d resv + %d avail\n", ctx,
368	ctx->count, mdsc->caps_total_count, mdsc->caps_use_count,
369	mdsc->caps_reserve_count, mdsc->caps_avail_count);
370	BUG_ON(!ctx->count);
371	BUG_ON(ctx->count > mdsc->caps_reserve_count);
372	BUG_ON(list_empty(&mdsc->caps_list));
373
374	ctx->count--;
375	ctx->used++;
376	mdsc->caps_reserve_count--;
377	mdsc->caps_use_count++;
378
379	cap = list_first_entry(&mdsc->caps_list, struct ceph_cap, caps_item);
380	list_del(entry: &cap->caps_item);
381
382	BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
383	mdsc->caps_reserve_count + mdsc->caps_avail_count);
384	spin_unlock(lock: &mdsc->caps_list_lock);
385	return cap;
386	}
387
388	void ceph_put_cap(struct ceph_mds_client *mdsc, struct ceph_cap *cap)
389	{
390	struct ceph_client *cl = mdsc->fsc->client;
391
392	spin_lock(lock: &mdsc->caps_list_lock);
393	doutc(cl, "%p %d = %d used + %d resv + %d avail\n", cap,
394	mdsc->caps_total_count, mdsc->caps_use_count,
395	mdsc->caps_reserve_count, mdsc->caps_avail_count);
396	mdsc->caps_use_count--;
397	/*
398	* Keep some preallocated caps around (ceph_min_count), to
399	* avoid lots of free/alloc churn.
400	*/
401	if (mdsc->caps_avail_count >= mdsc->caps_reserve_count +
402	mdsc->caps_min_count) {
403	mdsc->caps_total_count--;
404	kmem_cache_free(s: ceph_cap_cachep, objp: cap);
405	} else {
406	mdsc->caps_avail_count++;
407	list_add(new: &cap->caps_item, head: &mdsc->caps_list);
408	}
409
410	BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
411	mdsc->caps_reserve_count + mdsc->caps_avail_count);
412	spin_unlock(lock: &mdsc->caps_list_lock);
413	}
414
415	void ceph_reservation_status(struct ceph_fs_client *fsc,
416	int *total, int *avail, int *used, int *reserved,
417	int *min)
418	{
419	struct ceph_mds_client *mdsc = fsc->mdsc;
420
421	spin_lock(lock: &mdsc->caps_list_lock);
422
423	if (total)
424	*total = mdsc->caps_total_count;
425	if (avail)
426	*avail = mdsc->caps_avail_count;
427	if (used)
428	*used = mdsc->caps_use_count;
429	if (reserved)
430	*reserved = mdsc->caps_reserve_count;
431	if (min)
432	*min = mdsc->caps_min_count;
433
434	spin_unlock(lock: &mdsc->caps_list_lock);
435	}
436
437	/*
438	* Find ceph_cap for given mds, if any.
439	*
440	* Called with i_ceph_lock held.
441	*/
442	struct ceph_cap *__get_cap_for_mds(struct ceph_inode_info *ci, int mds)
443	{
444	struct ceph_cap *cap;
445	struct rb_node *n = ci->i_caps.rb_node;
446
447	while (n) {
448	cap = rb_entry(n, struct ceph_cap, ci_node);
449	if (mds < cap->mds)
450	n = n->rb_left;
451	else if (mds > cap->mds)
452	n = n->rb_right;
453	else
454	return cap;
455	}
456	return NULL;
457	}
458
459	struct ceph_cap *ceph_get_cap_for_mds(struct ceph_inode_info *ci, int mds)
460	{
461	struct ceph_cap *cap;
462
463	spin_lock(lock: &ci->i_ceph_lock);
464	cap = __get_cap_for_mds(ci, mds);
465	spin_unlock(lock: &ci->i_ceph_lock);
466	return cap;
467	}
468
469	/*
470	* Called under i_ceph_lock.
471	*/
472	static void __insert_cap_node(struct ceph_inode_info *ci,
473	struct ceph_cap *new)
474	{
475	struct rb_node **p = &ci->i_caps.rb_node;
476	struct rb_node *parent = NULL;
477	struct ceph_cap *cap = NULL;
478
479	while (*p) {
480	parent = *p;
481	cap = rb_entry(parent, struct ceph_cap, ci_node);
482	if (new->mds < cap->mds)
483	p = &(*p)->rb_left;
484	else if (new->mds > cap->mds)
485	p = &(*p)->rb_right;
486	else
487	BUG();
488	}
489
490	rb_link_node(node: &new->ci_node, parent, rb_link: p);
491	rb_insert_color(&new->ci_node, &ci->i_caps);
492	}
493
494	/*
495	* (re)set cap hold timeouts, which control the delayed release
496	* of unused caps back to the MDS. Should be called on cap use.
497	*/
498	static void __cap_set_timeouts(struct ceph_mds_client *mdsc,
499	struct ceph_inode_info *ci)
500	{
501	struct inode *inode = &ci->netfs.inode;
502	struct ceph_mount_options *opt = mdsc->fsc->mount_options;
503
504	ci->i_hold_caps_max = round_jiffies(j: jiffies +
505	opt->caps_wanted_delay_max * HZ);
506	doutc(mdsc->fsc->client, "%p %llx.%llx %lu\n", inode,
507	ceph_vinop(inode), ci->i_hold_caps_max - jiffies);
508	}
509
510	/*
511	* (Re)queue cap at the end of the delayed cap release list.
512	*
513	* If I_FLUSH is set, leave the inode at the front of the list.
514	*
515	* Caller holds i_ceph_lock
516	* -> we take mdsc->cap_delay_lock
517	*/
518	static void __cap_delay_requeue(struct ceph_mds_client *mdsc,
519	struct ceph_inode_info *ci)
520	{
521	struct inode *inode = &ci->netfs.inode;
522
523	doutc(mdsc->fsc->client, "%p %llx.%llx flags 0x%lx at %lu\n",
524	inode, ceph_vinop(inode), ci->i_ceph_flags,
525	ci->i_hold_caps_max);
526	if (!mdsc->stopping) {
527	spin_lock(lock: &mdsc->cap_delay_lock);
528	if (!list_empty(head: &ci->i_cap_delay_list)) {
529	if (ci->i_ceph_flags & CEPH_I_FLUSH)
530	goto no_change;
531	list_del_init(entry: &ci->i_cap_delay_list);
532	}
533	__cap_set_timeouts(mdsc, ci);
534	list_add_tail(new: &ci->i_cap_delay_list, head: &mdsc->cap_delay_list);
535	no_change:
536	spin_unlock(lock: &mdsc->cap_delay_lock);
537	}
538	}
539
540	/*
541	* Queue an inode for immediate writeback. Mark inode with I_FLUSH,
542	* indicating we should send a cap message to flush dirty metadata
543	* asap, and move to the front of the delayed cap list.
544	*/
545	static void __cap_delay_requeue_front(struct ceph_mds_client *mdsc,
546	struct ceph_inode_info *ci)
547	{
548	struct inode *inode = &ci->netfs.inode;
549
550	doutc(mdsc->fsc->client, "%p %llx.%llx\n", inode, ceph_vinop(inode));
551	spin_lock(lock: &mdsc->cap_delay_lock);
552	ci->i_ceph_flags |= CEPH_I_FLUSH;
553	if (!list_empty(head: &ci->i_cap_delay_list))
554	list_del_init(entry: &ci->i_cap_delay_list);
555	list_add(new: &ci->i_cap_delay_list, head: &mdsc->cap_delay_list);
556	spin_unlock(lock: &mdsc->cap_delay_lock);
557	}
558
559	/*
560	* Cancel delayed work on cap.
561	*
562	* Caller must hold i_ceph_lock.
563	*/
564	static void __cap_delay_cancel(struct ceph_mds_client *mdsc,
565	struct ceph_inode_info *ci)
566	{
567	struct inode *inode = &ci->netfs.inode;
568
569	doutc(mdsc->fsc->client, "%p %llx.%llx\n", inode, ceph_vinop(inode));
570	if (list_empty(head: &ci->i_cap_delay_list))
571	return;
572	spin_lock(lock: &mdsc->cap_delay_lock);
573	list_del_init(entry: &ci->i_cap_delay_list);
574	spin_unlock(lock: &mdsc->cap_delay_lock);
575	}
576
577	/* Common issue checks for add_cap, handle_cap_grant. */
578	static void __check_cap_issue(struct ceph_inode_info *ci, struct ceph_cap *cap,
579	unsigned issued)
580	{
581	struct inode *inode = &ci->netfs.inode;
582	struct ceph_client *cl = ceph_inode_to_client(inode);
583
584	unsigned had = __ceph_caps_issued(ci, NULL);
585
586	lockdep_assert_held(&ci->i_ceph_lock);
587
588	/*
589	* Each time we receive FILE_CACHE anew, we increment
590	* i_rdcache_gen.
591	*/
592	if (S_ISREG(ci->netfs.inode.i_mode) &&
593	(issued & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) &&
594	(had & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0) {
595	ci->i_rdcache_gen++;
596	}
597
598	/*
599	* If FILE_SHARED is newly issued, mark dir not complete. We don't
600	* know what happened to this directory while we didn't have the cap.
601	* If FILE_SHARED is being revoked, also mark dir not complete. It
602	* stops on-going cached readdir.
603	*/
604	if ((issued & CEPH_CAP_FILE_SHARED) != (had & CEPH_CAP_FILE_SHARED)) {
605	if (issued & CEPH_CAP_FILE_SHARED)
606	atomic_inc(v: &ci->i_shared_gen);
607	if (S_ISDIR(ci->netfs.inode.i_mode)) {
608	doutc(cl, " marking %p NOT complete\n", inode);
609	__ceph_dir_clear_complete(ci);
610	}
611	}
612
613	/* Wipe saved layout if we're losing DIR_CREATE caps */
614	if (S_ISDIR(ci->netfs.inode.i_mode) && (had & CEPH_CAP_DIR_CREATE) &&
615	!(issued & CEPH_CAP_DIR_CREATE)) {
616	ceph_put_string(rcu_dereference_raw(ci->i_cached_layout.pool_ns));
617	memset(&ci->i_cached_layout, 0, sizeof(ci->i_cached_layout));
618	}
619	}
620
621	/**
622	* change_auth_cap_ses - move inode to appropriate lists when auth caps change
623	* @ci: inode to be moved
624	* @session: new auth caps session
625	*/
626	void change_auth_cap_ses(struct ceph_inode_info *ci,
627	struct ceph_mds_session *session)
628	{
629	lockdep_assert_held(&ci->i_ceph_lock);
630
631	if (list_empty(head: &ci->i_dirty_item) && list_empty(head: &ci->i_flushing_item))
632	return;
633
634	spin_lock(lock: &session->s_mdsc->cap_dirty_lock);
635	if (!list_empty(head: &ci->i_dirty_item))
636	list_move(list: &ci->i_dirty_item, head: &session->s_cap_dirty);
637	if (!list_empty(head: &ci->i_flushing_item))
638	list_move_tail(list: &ci->i_flushing_item, head: &session->s_cap_flushing);
639	spin_unlock(lock: &session->s_mdsc->cap_dirty_lock);
640	}
641
642	/*
643	* Add a capability under the given MDS session.
644	*
645	* Caller should hold session snap_rwsem (read) and ci->i_ceph_lock
646	*
647	* @fmode is the open file mode, if we are opening a file, otherwise
648	* it is < 0. (This is so we can atomically add the cap and add an
649	* open file reference to it.)
650	*/
651	void ceph_add_cap(struct inode *inode,
652	struct ceph_mds_session *session, u64 cap_id,
653	unsigned issued, unsigned wanted,
654	unsigned seq, unsigned mseq, u64 realmino, int flags,
655	struct ceph_cap **new_cap)
656	{
657	struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
658	struct ceph_client *cl = ceph_inode_to_client(inode);
659	struct ceph_inode_info *ci = ceph_inode(inode);
660	struct ceph_cap *cap;
661	int mds = session->s_mds;
662	int actual_wanted;
663	u32 gen;
664
665	lockdep_assert_held(&ci->i_ceph_lock);
666
667	doutc(cl, "%p %llx.%llx mds%d cap %llx %s seq %d\n", inode,
668	ceph_vinop(inode), session->s_mds, cap_id,
669	ceph_cap_string(issued), seq);
670
671	gen = atomic_read(v: &session->s_cap_gen);
672
673	cap = __get_cap_for_mds(ci, mds);
674	if (!cap) {
675	cap = *new_cap;
676	*new_cap = NULL;
677
678	cap->issued = 0;
679	cap->implemented = 0;
680	cap->mds = mds;
681	cap->mds_wanted = 0;
682	cap->mseq = 0;
683
684	cap->ci = ci;
685	__insert_cap_node(ci, new: cap);
686
687	/* add to session cap list */
688	cap->session = session;
689	spin_lock(lock: &session->s_cap_lock);
690	list_add_tail(new: &cap->session_caps, head: &session->s_caps);
691	session->s_nr_caps++;
692	atomic64_inc(v: &mdsc->metric.total_caps);
693	spin_unlock(lock: &session->s_cap_lock);
694	} else {
695	spin_lock(lock: &session->s_cap_lock);
696	list_move_tail(list: &cap->session_caps, head: &session->s_caps);
697	spin_unlock(lock: &session->s_cap_lock);
698
699	if (cap->cap_gen < gen)
700	cap->issued = cap->implemented = CEPH_CAP_PIN;
701
702	/*
703	* auth mds of the inode changed. we received the cap export
704	* message, but still haven't received the cap import message.
705	* handle_cap_export() updated the new auth MDS' cap.
706	*
707	* "ceph_seq_cmp(seq, cap->seq) <= 0" means we are processing
708	* a message that was send before the cap import message. So
709	* don't remove caps.
710	*/
711	if (ceph_seq_cmp(a: seq, b: cap->seq) <= 0) {
712	WARN_ON(cap != ci->i_auth_cap);
713	WARN_ON(cap->cap_id != cap_id);
714	seq = cap->seq;
715	mseq = cap->mseq;
716	issued |= cap->issued;
717	flags |= CEPH_CAP_FLAG_AUTH;
718	}
719	}
720
721	if (!ci->i_snap_realm ||
722	((flags & CEPH_CAP_FLAG_AUTH) &&
723	realmino != (u64)-1 && ci->i_snap_realm->ino != realmino)) {
724	/*
725	* add this inode to the appropriate snap realm
726	*/
727	struct ceph_snap_realm *realm = ceph_lookup_snap_realm(mdsc,
728	ino: realmino);
729	if (realm)
730	ceph_change_snap_realm(inode, realm);
731	else
732	WARN(1, "%s: couldn't find snap realm 0x%llx (ino 0x%llx oldrealm 0x%llx)\n",
733	__func__, realmino, ci->i_vino.ino,
734	ci->i_snap_realm ? ci->i_snap_realm->ino : 0);
735	}
736
737	__check_cap_issue(ci, cap, issued);
738
739	/*
740	* If we are issued caps we don't want, or the mds' wanted
741	* value appears to be off, queue a check so we'll release
742	* later and/or update the mds wanted value.
743	*/
744	actual_wanted = __ceph_caps_wanted(ci);
745	if ((wanted & ~actual_wanted) ||
746	(issued & ~actual_wanted & CEPH_CAP_ANY_WR)) {
747	doutc(cl, "issued %s, mds wanted %s, actual %s, queueing\n",
748	ceph_cap_string(issued), ceph_cap_string(wanted),
749	ceph_cap_string(actual_wanted));
750	__cap_delay_requeue(mdsc, ci);
751	}
752
753	if (flags & CEPH_CAP_FLAG_AUTH) {
754	if (!ci->i_auth_cap ||
755	ceph_seq_cmp(a: ci->i_auth_cap->mseq, b: mseq) < 0) {
756	if (ci->i_auth_cap &&
757	ci->i_auth_cap->session != cap->session)
758	change_auth_cap_ses(ci, session: cap->session);
759	ci->i_auth_cap = cap;
760	cap->mds_wanted = wanted;
761	}
762	} else {
763	WARN_ON(ci->i_auth_cap == cap);
764	}
765
766	doutc(cl, "inode %p %llx.%llx cap %p %s now %s seq %d mds%d\n",
767	inode, ceph_vinop(inode), cap, ceph_cap_string(issued),
768	ceph_cap_string(issued|cap->issued), seq, mds);
769	cap->cap_id = cap_id;
770	cap->issued = issued;
771	cap->implemented |= issued;
772	if (ceph_seq_cmp(a: mseq, b: cap->mseq) > 0)
773	cap->mds_wanted = wanted;
774	else
775	cap->mds_wanted |= wanted;
776	cap->seq = seq;
777	cap->issue_seq = seq;
778	cap->mseq = mseq;
779	cap->cap_gen = gen;
780	wake_up_all(&ci->i_cap_wq);
781	}
782
783	/*
784	* Return true if cap has not timed out and belongs to the current
785	* generation of the MDS session (i.e. has not gone 'stale' due to
786	* us losing touch with the mds).
787	*/
788	static int __cap_is_valid(struct ceph_cap *cap)
789	{
790	struct inode *inode = &cap->ci->netfs.inode;
791	struct ceph_client *cl = cap->session->s_mdsc->fsc->client;
792	unsigned long ttl;
793	u32 gen;
794
795	gen = atomic_read(v: &cap->session->s_cap_gen);
796	ttl = cap->session->s_cap_ttl;
797
798	if (cap->cap_gen < gen || time_after_eq(jiffies, ttl)) {
799	doutc(cl, "%p %llx.%llx cap %p issued %s but STALE (gen %u vs %u)\n",
800	inode, ceph_vinop(inode), cap,
801	ceph_cap_string(cap->issued), cap->cap_gen, gen);
802	return 0;
803	}
804
805	return 1;
806	}
807
808	/*
809	* Return set of valid cap bits issued to us. Note that caps time
810	* out, and may be invalidated in bulk if the client session times out
811	* and session->s_cap_gen is bumped.
812	*/
813	int __ceph_caps_issued(struct ceph_inode_info *ci, int *implemented)
814	{
815	struct inode *inode = &ci->netfs.inode;
816	struct ceph_client *cl = ceph_inode_to_client(inode);
817	int have = ci->i_snap_caps;
818	struct ceph_cap *cap;
819	struct rb_node *p;
820
821	if (implemented)
822	*implemented = 0;
823	for (p = rb_first(root: &ci->i_caps); p; p = rb_next(p)) {
824	cap = rb_entry(p, struct ceph_cap, ci_node);
825	if (!__cap_is_valid(cap))
826	continue;
827	doutc(cl, "%p %llx.%llx cap %p issued %s\n", inode,
828	ceph_vinop(inode), cap, ceph_cap_string(cap->issued));
829	have |= cap->issued;
830	if (implemented)
831	*implemented |= cap->implemented;
832	}
833	/*
834	* exclude caps issued by non-auth MDS, but are been revoking
835	* by the auth MDS. The non-auth MDS should be revoking/exporting
836	* these caps, but the message is delayed.
837	*/
838	if (ci->i_auth_cap) {
839	cap = ci->i_auth_cap;
840	have &= ~cap->implemented | cap->issued;
841	}
842	return have;
843	}
844
845	/*
846	* Get cap bits issued by caps other than @ocap
847	*/
848	int __ceph_caps_issued_other(struct ceph_inode_info *ci, struct ceph_cap *ocap)
849	{
850	int have = ci->i_snap_caps;
851	struct ceph_cap *cap;
852	struct rb_node *p;
853
854	for (p = rb_first(root: &ci->i_caps); p; p = rb_next(p)) {
855	cap = rb_entry(p, struct ceph_cap, ci_node);
856	if (cap == ocap)
857	continue;
858	if (!__cap_is_valid(cap))
859	continue;
860	have |= cap->issued;
861	}
862	return have;
863	}
864
865	/*
866	* Move a cap to the end of the LRU (oldest caps at list head, newest
867	* at list tail).
868	*/
869	static void __touch_cap(struct ceph_cap *cap)
870	{
871	struct inode *inode = &cap->ci->netfs.inode;
872	struct ceph_mds_session *s = cap->session;
873	struct ceph_client *cl = s->s_mdsc->fsc->client;
874
875	spin_lock(lock: &s->s_cap_lock);
876	if (!s->s_cap_iterator) {
877	doutc(cl, "%p %llx.%llx cap %p mds%d\n", inode,
878	ceph_vinop(inode), cap, s->s_mds);
879	list_move_tail(list: &cap->session_caps, head: &s->s_caps);
880	} else {
881	doutc(cl, "%p %llx.%llx cap %p mds%d NOP, iterating over caps\n",
882	inode, ceph_vinop(inode), cap, s->s_mds);
883	}
884	spin_unlock(lock: &s->s_cap_lock);
885	}
886
887	/*
888	* Check if we hold the given mask. If so, move the cap(s) to the
889	* front of their respective LRUs. (This is the preferred way for
890	* callers to check for caps they want.)
891	*/
892	int __ceph_caps_issued_mask(struct ceph_inode_info *ci, int mask, int touch)
893	{
894	struct inode *inode = &ci->netfs.inode;
895	struct ceph_client *cl = ceph_inode_to_client(inode);
896	struct ceph_cap *cap;
897	struct rb_node *p;
898	int have = ci->i_snap_caps;
899
900	if ((have & mask) == mask) {
901	doutc(cl, "mask %p %llx.%llx snap issued %s (mask %s)\n",
902	inode, ceph_vinop(inode), ceph_cap_string(have),
903	ceph_cap_string(mask));
904	return 1;
905	}
906
907	for (p = rb_first(root: &ci->i_caps); p; p = rb_next(p)) {
908	cap = rb_entry(p, struct ceph_cap, ci_node);
909	if (!__cap_is_valid(cap))
910	continue;
911	if ((cap->issued & mask) == mask) {
912	doutc(cl, "mask %p %llx.%llx cap %p issued %s (mask %s)\n",
913	inode, ceph_vinop(inode), cap,
914	ceph_cap_string(cap->issued),
915	ceph_cap_string(mask));
916	if (touch)
917	__touch_cap(cap);
918	return 1;
919	}
920
921	/* does a combination of caps satisfy mask? */
922	have |= cap->issued;
923	if ((have & mask) == mask) {
924	doutc(cl, "mask %p %llx.%llx combo issued %s (mask %s)\n",
925	inode, ceph_vinop(inode),
926	ceph_cap_string(cap->issued),
927	ceph_cap_string(mask));
928	if (touch) {
929	struct rb_node *q;
930
931	/* touch this + preceding caps */
932	__touch_cap(cap);
933	for (q = rb_first(root: &ci->i_caps); q != p;
934	q = rb_next(q)) {
935	cap = rb_entry(q, struct ceph_cap,
936	ci_node);
937	if (!__cap_is_valid(cap))
938	continue;
939	if (cap->issued & mask)
940	__touch_cap(cap);
941	}
942	}
943	return 1;
944	}
945	}
946
947	return 0;
948	}
949
950	int __ceph_caps_issued_mask_metric(struct ceph_inode_info *ci, int mask,
951	int touch)
952	{
953	struct ceph_fs_client *fsc = ceph_sb_to_fs_client(sb: ci->netfs.inode.i_sb);
954	int r;
955
956	r = __ceph_caps_issued_mask(ci, mask, touch);
957	if (r)
958	ceph_update_cap_hit(m: &fsc->mdsc->metric);
959	else
960	ceph_update_cap_mis(m: &fsc->mdsc->metric);
961	return r;
962	}
963
964	/*
965	* Return true if mask caps are currently being revoked by an MDS.
966	*/
967	int __ceph_caps_revoking_other(struct ceph_inode_info *ci,
968	struct ceph_cap *ocap, int mask)
969	{
970	struct ceph_cap *cap;
971	struct rb_node *p;
972
973	for (p = rb_first(root: &ci->i_caps); p; p = rb_next(p)) {
974	cap = rb_entry(p, struct ceph_cap, ci_node);
975	if (cap != ocap &&
976	(cap->implemented & ~cap->issued & mask))
977	return 1;
978	}
979	return 0;
980	}
981
982	int __ceph_caps_used(struct ceph_inode_info *ci)
983	{
984	int used = 0;
985	if (ci->i_pin_ref)
986	used |= CEPH_CAP_PIN;
987	if (ci->i_rd_ref)
988	used |= CEPH_CAP_FILE_RD;
989	if (ci->i_rdcache_ref ||
990	(S_ISREG(ci->netfs.inode.i_mode) &&
991	ci->netfs.inode.i_data.nrpages))
992	used |= CEPH_CAP_FILE_CACHE;
993	if (ci->i_wr_ref)
994	used |= CEPH_CAP_FILE_WR;
995	if (ci->i_wb_ref || ci->i_wrbuffer_ref)
996	used |= CEPH_CAP_FILE_BUFFER;
997	if (ci->i_fx_ref)
998	used |= CEPH_CAP_FILE_EXCL;
999	return used;
1000	}
1001
1002	#define FMODE_WAIT_BIAS 1000
1003
1004	/*
1005	* wanted, by virtue of open file modes
1006	*/
1007	int __ceph_caps_file_wanted(struct ceph_inode_info *ci)
1008	{
1009	const int PIN_SHIFT = ffs(CEPH_FILE_MODE_PIN);
1010	const int RD_SHIFT = ffs(CEPH_FILE_MODE_RD);
1011	const int WR_SHIFT = ffs(CEPH_FILE_MODE_WR);
1012	const int LAZY_SHIFT = ffs(CEPH_FILE_MODE_LAZY);
1013	struct ceph_mount_options *opt =
1014	ceph_inode_to_fs_client(inode: &ci->netfs.inode)->mount_options;
1015	unsigned long used_cutoff = jiffies - opt->caps_wanted_delay_max * HZ;
1016	unsigned long idle_cutoff = jiffies - opt->caps_wanted_delay_min * HZ;
1017
1018	if (S_ISDIR(ci->netfs.inode.i_mode)) {
1019	int want = 0;
1020
1021	/* use used_cutoff here, to keep dir's wanted caps longer */
1022	if (ci->i_nr_by_mode[RD_SHIFT] > 0 ||
1023	time_after(ci->i_last_rd, used_cutoff))
1024	want |= CEPH_CAP_ANY_SHARED;
1025
1026	if (ci->i_nr_by_mode[WR_SHIFT] > 0 ||
1027	time_after(ci->i_last_wr, used_cutoff)) {
1028	want |= CEPH_CAP_ANY_SHARED | CEPH_CAP_FILE_EXCL;
1029	if (opt->flags & CEPH_MOUNT_OPT_ASYNC_DIROPS)
1030	want |= CEPH_CAP_ANY_DIR_OPS;
1031	}
1032
1033	if (want || ci->i_nr_by_mode[PIN_SHIFT] > 0)
1034	want |= CEPH_CAP_PIN;
1035
1036	return want;
1037	} else {
1038	int bits = 0;
1039
1040	if (ci->i_nr_by_mode[RD_SHIFT] > 0) {
1041	if (ci->i_nr_by_mode[RD_SHIFT] >= FMODE_WAIT_BIAS ||
1042	time_after(ci->i_last_rd, used_cutoff))
1043	bits |= 1 << RD_SHIFT;
1044	} else if (time_after(ci->i_last_rd, idle_cutoff)) {
1045	bits |= 1 << RD_SHIFT;
1046	}
1047
1048	if (ci->i_nr_by_mode[WR_SHIFT] > 0) {
1049	if (ci->i_nr_by_mode[WR_SHIFT] >= FMODE_WAIT_BIAS ||
1050	time_after(ci->i_last_wr, used_cutoff))
1051	bits |= 1 << WR_SHIFT;
1052	} else if (time_after(ci->i_last_wr, idle_cutoff)) {
1053	bits |= 1 << WR_SHIFT;
1054	}
1055
1056	/* check lazyio only when read/write is wanted */
1057	if ((bits & (CEPH_FILE_MODE_RDWR << 1)) &&
1058	ci->i_nr_by_mode[LAZY_SHIFT] > 0)
1059	bits |= 1 << LAZY_SHIFT;
1060
1061	return bits ? ceph_caps_for_mode(mode: bits >> 1) : 0;
1062	}
1063	}
1064
1065	/*
1066	* wanted, by virtue of open file modes AND cap refs (buffered/cached data)
1067	*/
1068	int __ceph_caps_wanted(struct ceph_inode_info *ci)
1069	{
1070	int w = __ceph_caps_file_wanted(ci) | __ceph_caps_used(ci);
1071	if (S_ISDIR(ci->netfs.inode.i_mode)) {
1072	/* we want EXCL if holding caps of dir ops */
1073	if (w & CEPH_CAP_ANY_DIR_OPS)
1074	w |= CEPH_CAP_FILE_EXCL;
1075	} else {
1076	/* we want EXCL if dirty data */
1077	if (w & CEPH_CAP_FILE_BUFFER)
1078	w |= CEPH_CAP_FILE_EXCL;
1079	}
1080	return w;
1081	}
1082
1083	/*
1084	* Return caps we have registered with the MDS(s) as 'wanted'.
1085	*/
1086	int __ceph_caps_mds_wanted(struct ceph_inode_info *ci, bool check)
1087	{
1088	struct ceph_cap *cap;
1089	struct rb_node *p;
1090	int mds_wanted = 0;
1091
1092	for (p = rb_first(root: &ci->i_caps); p; p = rb_next(p)) {
1093	cap = rb_entry(p, struct ceph_cap, ci_node);
1094	if (check && !__cap_is_valid(cap))
1095	continue;
1096	if (cap == ci->i_auth_cap)
1097	mds_wanted |= cap->mds_wanted;
1098	else
1099	mds_wanted |= (cap->mds_wanted & ~CEPH_CAP_ANY_FILE_WR);
1100	}
1101	return mds_wanted;
1102	}
1103
1104	int ceph_is_any_caps(struct inode *inode)
1105	{
1106	struct ceph_inode_info *ci = ceph_inode(inode);
1107	int ret;
1108
1109	spin_lock(lock: &ci->i_ceph_lock);
1110	ret = __ceph_is_any_real_caps(ci);
1111	spin_unlock(lock: &ci->i_ceph_lock);
1112
1113	return ret;
1114	}
1115
1116	/*
1117	* Remove a cap. Take steps to deal with a racing iterate_session_caps.
1118	*
1119	* caller should hold i_ceph_lock.
1120	* caller will not hold session s_mutex if called from destroy_inode.
1121	*/
1122	void __ceph_remove_cap(struct ceph_cap *cap, bool queue_release)
1123	{
1124	struct ceph_mds_session *session = cap->session;
1125	struct ceph_client *cl = session->s_mdsc->fsc->client;
1126	struct ceph_inode_info *ci = cap->ci;
1127	struct inode *inode = &ci->netfs.inode;
1128	struct ceph_mds_client *mdsc;
1129	int removed = 0;
1130
1131	/* 'ci' being NULL means the remove have already occurred */
1132	if (!ci) {
1133	doutc(cl, "inode is NULL\n");
1134	return;
1135	}
1136
1137	lockdep_assert_held(&ci->i_ceph_lock);
1138
1139	doutc(cl, "%p from %p %llx.%llx\n", cap, inode, ceph_vinop(inode));
1140
1141	mdsc = ceph_inode_to_fs_client(inode: &ci->netfs.inode)->mdsc;
1142
1143	/* remove from inode's cap rbtree, and clear auth cap */
1144	rb_erase(&cap->ci_node, &ci->i_caps);
1145	if (ci->i_auth_cap == cap)
1146	ci->i_auth_cap = NULL;
1147
1148	/* remove from session list */
1149	spin_lock(lock: &session->s_cap_lock);
1150	if (session->s_cap_iterator == cap) {
1151	/* not yet, we are iterating over this very cap */
1152	doutc(cl, "delaying %p removal from session %p\n", cap,
1153	cap->session);
1154	} else {
1155	list_del_init(entry: &cap->session_caps);
1156	session->s_nr_caps--;
1157	atomic64_dec(v: &mdsc->metric.total_caps);
1158	cap->session = NULL;
1159	removed = 1;
1160	}
1161	/* protect backpointer with s_cap_lock: see iterate_session_caps */
1162	cap->ci = NULL;
1163
1164	/*
1165	* s_cap_reconnect is protected by s_cap_lock. no one changes
1166	* s_cap_gen while session is in the reconnect state.
1167	*/
1168	if (queue_release &&
1169	(!session->s_cap_reconnect ||
1170	cap->cap_gen == atomic_read(v: &session->s_cap_gen))) {
1171	cap->queue_release = 1;
1172	if (removed) {
1173	__ceph_queue_cap_release(session, cap);
1174	removed = 0;
1175	}
1176	} else {
1177	cap->queue_release = 0;
1178	}
1179	cap->cap_ino = ci->i_vino.ino;
1180
1181	spin_unlock(lock: &session->s_cap_lock);
1182
1183	if (removed)
1184	ceph_put_cap(mdsc, cap);
1185
1186	if (!__ceph_is_any_real_caps(ci)) {
1187	/* when reconnect denied, we remove session caps forcibly,
1188	* i_wr_ref can be non-zero. If there are ongoing write,
1189	* keep i_snap_realm.
1190	*/
1191	if (ci->i_wr_ref == 0 && ci->i_snap_realm)
1192	ceph_change_snap_realm(inode: &ci->netfs.inode, NULL);
1193
1194	__cap_delay_cancel(mdsc, ci);
1195	}
1196	}
1197
1198	void ceph_remove_cap(struct ceph_mds_client *mdsc, struct ceph_cap *cap,
1199	bool queue_release)
1200	{
1201	struct ceph_inode_info *ci = cap->ci;
1202	struct ceph_fs_client *fsc;
1203
1204	/* 'ci' being NULL means the remove have already occurred */
1205	if (!ci) {
1206	doutc(mdsc->fsc->client, "inode is NULL\n");
1207	return;
1208	}
1209
1210	lockdep_assert_held(&ci->i_ceph_lock);
1211
1212	fsc = ceph_inode_to_fs_client(inode: &ci->netfs.inode);
1213	WARN_ON_ONCE(ci->i_auth_cap == cap &&
1214	!list_empty(&ci->i_dirty_item) &&
1215	!fsc->blocklisted &&
1216	!ceph_inode_is_shutdown(&ci->netfs.inode));
1217
1218	__ceph_remove_cap(cap, queue_release);
1219	}
1220
1221	struct cap_msg_args {
1222	struct ceph_mds_session *session;
1223	u64 ino, cid, follows;
1224	u64 flush_tid, oldest_flush_tid, size, max_size;
1225	u64 xattr_version;
1226	u64 change_attr;
1227	struct ceph_buffer *xattr_buf;
1228	struct ceph_buffer *old_xattr_buf;
1229	struct timespec64 atime, mtime, ctime, btime;
1230	int op, caps, wanted, dirty;
1231	u32 seq, issue_seq, mseq, time_warp_seq;
1232	u32 flags;
1233	kuid_t uid;
1234	kgid_t gid;
1235	umode_t mode;
1236	bool inline_data;
1237	bool wake;
1238	bool encrypted;
1239	u32 fscrypt_auth_len;
1240	u8 fscrypt_auth[sizeof(struct ceph_fscrypt_auth)]; // for context
1241	};
1242
1243	/* Marshal up the cap msg to the MDS */
1244	static void encode_cap_msg(struct ceph_msg *msg, struct cap_msg_args *arg)
1245	{
1246	struct ceph_mds_caps *fc;
1247	void *p;
1248	struct ceph_mds_client *mdsc = arg->session->s_mdsc;
1249	struct ceph_osd_client *osdc = &mdsc->fsc->client->osdc;
1250
1251	doutc(mdsc->fsc->client,
1252	"%s %llx %llx caps %s wanted %s dirty %s seq %u/%u"
1253	" tid %llu/%llu mseq %u follows %lld size %llu/%llu"
1254	" xattr_ver %llu xattr_len %d\n",
1255	ceph_cap_op_name(arg->op), arg->cid, arg->ino,
1256	ceph_cap_string(arg->caps), ceph_cap_string(arg->wanted),
1257	ceph_cap_string(arg->dirty), arg->seq, arg->issue_seq,
1258	arg->flush_tid, arg->oldest_flush_tid, arg->mseq, arg->follows,
1259	arg->size, arg->max_size, arg->xattr_version,
1260	arg->xattr_buf ? (int)arg->xattr_buf->vec.iov_len : 0);
1261
1262	msg->hdr.version = cpu_to_le16(12);
1263	msg->hdr.tid = cpu_to_le64(arg->flush_tid);
1264
1265	fc = msg->front.iov_base;
1266	memset(fc, 0, sizeof(*fc));
1267
1268	fc->cap_id = cpu_to_le64(arg->cid);
1269	fc->op = cpu_to_le32(arg->op);
1270	fc->seq = cpu_to_le32(arg->seq);
1271	fc->issue_seq = cpu_to_le32(arg->issue_seq);
1272	fc->migrate_seq = cpu_to_le32(arg->mseq);
1273	fc->caps = cpu_to_le32(arg->caps);
1274	fc->wanted = cpu_to_le32(arg->wanted);
1275	fc->dirty = cpu_to_le32(arg->dirty);
1276	fc->ino = cpu_to_le64(arg->ino);
1277	fc->snap_follows = cpu_to_le64(arg->follows);
1278
1279	#if IS_ENABLED(CONFIG_FS_ENCRYPTION)
1280	if (arg->encrypted)
1281	fc->size = cpu_to_le64(round_up(arg->size,
1282	CEPH_FSCRYPT_BLOCK_SIZE));
1283	else
1284	#endif
1285	fc->size = cpu_to_le64(arg->size);
1286	fc->max_size = cpu_to_le64(arg->max_size);
1287	ceph_encode_timespec64(tv: &fc->mtime, ts: &arg->mtime);
1288	ceph_encode_timespec64(tv: &fc->atime, ts: &arg->atime);
1289	ceph_encode_timespec64(tv: &fc->ctime, ts: &arg->ctime);
1290	fc->time_warp_seq = cpu_to_le32(arg->time_warp_seq);
1291
1292	fc->uid = cpu_to_le32(from_kuid(&init_user_ns, arg->uid));
1293	fc->gid = cpu_to_le32(from_kgid(&init_user_ns, arg->gid));
1294	fc->mode = cpu_to_le32(arg->mode);
1295
1296	fc->xattr_version = cpu_to_le64(arg->xattr_version);
1297	if (arg->xattr_buf) {
1298	msg->middle = ceph_buffer_get(b: arg->xattr_buf);
1299	fc->xattr_len = cpu_to_le32(arg->xattr_buf->vec.iov_len);
1300	msg->hdr.middle_len = cpu_to_le32(arg->xattr_buf->vec.iov_len);
1301	}
1302
1303	p = fc + 1;
1304	/* flock buffer size (version 2) */
1305	ceph_encode_32(p: &p, v: 0);
1306	/* inline version (version 4) */
1307	ceph_encode_64(p: &p, v: arg->inline_data ? 0 : CEPH_INLINE_NONE);
1308	/* inline data size */
1309	ceph_encode_32(p: &p, v: 0);
1310	/*
1311	* osd_epoch_barrier (version 5)
1312	* The epoch_barrier is protected osdc->lock, so READ_ONCE here in
1313	* case it was recently changed
1314	*/
1315	ceph_encode_32(p: &p, READ_ONCE(osdc->epoch_barrier));
1316	/* oldest_flush_tid (version 6) */
1317	ceph_encode_64(p: &p, v: arg->oldest_flush_tid);
1318
1319	/*
1320	* caller_uid/caller_gid (version 7)
1321	*
1322	* Currently, we don't properly track which caller dirtied the caps
1323	* last, and force a flush of them when there is a conflict. For now,
1324	* just set this to 0:0, to emulate how the MDS has worked up to now.
1325	*/
1326	ceph_encode_32(p: &p, v: 0);
1327	ceph_encode_32(p: &p, v: 0);
1328
1329	/* pool namespace (version 8) (mds always ignores this) */
1330	ceph_encode_32(p: &p, v: 0);
1331
1332	/* btime and change_attr (version 9) */
1333	ceph_encode_timespec64(tv: p, ts: &arg->btime);
1334	p += sizeof(struct ceph_timespec);
1335	ceph_encode_64(p: &p, v: arg->change_attr);
1336
1337	/* Advisory flags (version 10) */
1338	ceph_encode_32(p: &p, v: arg->flags);
1339
1340	/* dirstats (version 11) - these are r/o on the client */
1341	ceph_encode_64(p: &p, v: 0);
1342	ceph_encode_64(p: &p, v: 0);
1343
1344	#if IS_ENABLED(CONFIG_FS_ENCRYPTION)
1345	/*
1346	* fscrypt_auth and fscrypt_file (version 12)
1347	*
1348	* fscrypt_auth holds the crypto context (if any). fscrypt_file
1349	* tracks the real i_size as an __le64 field (and we use a rounded-up
1350	* i_size in the traditional size field).
1351	*/
1352	ceph_encode_32(p: &p, v: arg->fscrypt_auth_len);
1353	ceph_encode_copy(p: &p, s: arg->fscrypt_auth, len: arg->fscrypt_auth_len);
1354	ceph_encode_32(p: &p, v: sizeof(__le64));
1355	ceph_encode_64(p: &p, v: arg->size);
1356	#else /* CONFIG_FS_ENCRYPTION */
1357	ceph_encode_32(&p, 0);
1358	ceph_encode_32(&p, 0);
1359	#endif /* CONFIG_FS_ENCRYPTION */
1360	}
1361
1362	/*
1363	* Queue cap releases when an inode is dropped from our cache.
1364	*/
1365	void __ceph_remove_caps(struct ceph_inode_info *ci)
1366	{
1367	struct inode *inode = &ci->netfs.inode;
1368	struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
1369	struct rb_node *p;
1370
1371	/* lock i_ceph_lock, because ceph_d_revalidate(..., LOOKUP_RCU)
1372	* may call __ceph_caps_issued_mask() on a freeing inode. */
1373	spin_lock(lock: &ci->i_ceph_lock);
1374	p = rb_first(root: &ci->i_caps);
1375	while (p) {
1376	struct ceph_cap *cap = rb_entry(p, struct ceph_cap, ci_node);
1377	p = rb_next(p);
1378	ceph_remove_cap(mdsc, cap, queue_release: true);
1379	}
1380	spin_unlock(lock: &ci->i_ceph_lock);
1381	}
1382
1383	/*
1384	* Prepare to send a cap message to an MDS. Update the cap state, and populate
1385	* the arg struct with the parameters that will need to be sent. This should
1386	* be done under the i_ceph_lock to guard against changes to cap state.
1387	*
1388	* Make note of max_size reported/requested from mds, revoked caps
1389	* that have now been implemented.
1390	*/
1391	static void __prep_cap(struct cap_msg_args *arg, struct ceph_cap *cap,
1392	int op, int flags, int used, int want, int retain,
1393	int flushing, u64 flush_tid, u64 oldest_flush_tid)
1394	{
1395	struct ceph_inode_info *ci = cap->ci;
1396	struct inode *inode = &ci->netfs.inode;
1397	struct ceph_client *cl = ceph_inode_to_client(inode);
1398	int held, revoking;
1399
1400	lockdep_assert_held(&ci->i_ceph_lock);
1401
1402	held = cap->issued | cap->implemented;
1403	revoking = cap->implemented & ~cap->issued;
1404	retain &= ~revoking;
1405
1406	doutc(cl, "%p %llx.%llx cap %p session %p %s -> %s (revoking %s)\n",
1407	inode, ceph_vinop(inode), cap, cap->session,
1408	ceph_cap_string(held), ceph_cap_string(held & retain),
1409	ceph_cap_string(revoking));
1410	BUG_ON((retain & CEPH_CAP_PIN) == 0);
1411
1412	ci->i_ceph_flags &= ~CEPH_I_FLUSH;
1413
1414	cap->issued &= retain; /* drop bits we don't want */
1415	/*
1416	* Wake up any waiters on wanted -> needed transition. This is due to
1417	* the weird transition from buffered to sync IO... we need to flush
1418	* dirty pages _before_ allowing sync writes to avoid reordering.
1419	*/
1420	arg->wake = cap->implemented & ~cap->issued;
1421	cap->implemented &= cap->issued | used;
1422	cap->mds_wanted = want;
1423
1424	arg->session = cap->session;
1425	arg->ino = ceph_vino(inode).ino;
1426	arg->cid = cap->cap_id;
1427	arg->follows = flushing ? ci->i_head_snapc->seq : 0;
1428	arg->flush_tid = flush_tid;
1429	arg->oldest_flush_tid = oldest_flush_tid;
1430	arg->size = i_size_read(inode);
1431	ci->i_reported_size = arg->size;
1432	arg->max_size = ci->i_wanted_max_size;
1433	if (cap == ci->i_auth_cap) {
1434	if (want & CEPH_CAP_ANY_FILE_WR)
1435	ci->i_requested_max_size = arg->max_size;
1436	else
1437	ci->i_requested_max_size = 0;
1438	}
1439
1440	if (flushing & CEPH_CAP_XATTR_EXCL) {
1441	arg->old_xattr_buf = __ceph_build_xattrs_blob(ci);
1442	arg->xattr_version = ci->i_xattrs.version;
1443	arg->xattr_buf = ceph_buffer_get(b: ci->i_xattrs.blob);
1444	} else {
1445	arg->xattr_buf = NULL;
1446	arg->old_xattr_buf = NULL;
1447	}
1448
1449	arg->mtime = inode_get_mtime(inode);
1450	arg->atime = inode_get_atime(inode);
1451	arg->ctime = inode_get_ctime(inode);
1452	arg->btime = ci->i_btime;
1453	arg->change_attr = inode_peek_iversion_raw(inode);
1454
1455	arg->op = op;
1456	arg->caps = cap->implemented;
1457	arg->wanted = want;
1458	arg->dirty = flushing;
1459
1460	arg->seq = cap->seq;
1461	arg->issue_seq = cap->issue_seq;
1462	arg->mseq = cap->mseq;
1463	arg->time_warp_seq = ci->i_time_warp_seq;
1464
1465	arg->uid = inode->i_uid;
1466	arg->gid = inode->i_gid;
1467	arg->mode = inode->i_mode;
1468
1469	arg->inline_data = ci->i_inline_version != CEPH_INLINE_NONE;
1470	if (!(flags & CEPH_CLIENT_CAPS_PENDING_CAPSNAP) &&
1471	!list_empty(head: &ci->i_cap_snaps)) {
1472	struct ceph_cap_snap *capsnap;
1473	list_for_each_entry_reverse(capsnap, &ci->i_cap_snaps, ci_item) {
1474	if (capsnap->cap_flush.tid)
1475	break;
1476	if (capsnap->need_flush) {
1477	flags |= CEPH_CLIENT_CAPS_PENDING_CAPSNAP;
1478	break;
1479	}
1480	}
1481	}
1482	arg->flags = flags;
1483	arg->encrypted = IS_ENCRYPTED(inode);
1484	#if IS_ENABLED(CONFIG_FS_ENCRYPTION)
1485	if (ci->fscrypt_auth_len &&
1486	WARN_ON_ONCE(ci->fscrypt_auth_len > sizeof(struct ceph_fscrypt_auth))) {
1487	/* Don't set this if it's too big */
1488	arg->fscrypt_auth_len = 0;
1489	} else {
1490	arg->fscrypt_auth_len = ci->fscrypt_auth_len;
1491	memcpy(arg->fscrypt_auth, ci->fscrypt_auth,
1492	min_t(size_t, ci->fscrypt_auth_len,
1493	sizeof(arg->fscrypt_auth)));
1494	}
1495	#endif /* CONFIG_FS_ENCRYPTION */
1496	}
1497
1498	#if IS_ENABLED(CONFIG_FS_ENCRYPTION)
1499	#define CAP_MSG_FIXED_FIELDS (sizeof(struct ceph_mds_caps) + \
1500	4 + 8 + 4 + 4 + 8 + 4 + 4 + 4 + 8 + 8 + 4 + 8 + 8 + 4 + 4 + 8)
1501
1502	static inline int cap_msg_size(struct cap_msg_args *arg)
1503	{
1504	return CAP_MSG_FIXED_FIELDS + arg->fscrypt_auth_len;
1505	}
1506	#else
1507	#define CAP_MSG_FIXED_FIELDS (sizeof(struct ceph_mds_caps) + \
1508	4 + 8 + 4 + 4 + 8 + 4 + 4 + 4 + 8 + 8 + 4 + 8 + 8 + 4 + 4)
1509
1510	static inline int cap_msg_size(struct cap_msg_args *arg)
1511	{
1512	return CAP_MSG_FIXED_FIELDS;
1513	}
1514	#endif /* CONFIG_FS_ENCRYPTION */
1515
1516	/*
1517	* Send a cap msg on the given inode.
1518	*
1519	* Caller should hold snap_rwsem (read), s_mutex.
1520	*/
1521	static void __send_cap(struct cap_msg_args *arg, struct ceph_inode_info *ci)
1522	{
1523	struct ceph_msg *msg;
1524	struct inode *inode = &ci->netfs.inode;
1525	struct ceph_client *cl = ceph_inode_to_client(inode);
1526
1527	msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPS, front_len: cap_msg_size(arg), GFP_NOFS,
1528	can_fail: false);
1529	if (!msg) {
1530	pr_err_client(cl,
1531	"error allocating cap msg: ino (%llx.%llx)"
1532	" flushing %s tid %llu, requeuing cap.\n",
1533	ceph_vinop(inode), ceph_cap_string(arg->dirty),
1534	arg->flush_tid);
1535	spin_lock(lock: &ci->i_ceph_lock);
1536	__cap_delay_requeue(mdsc: arg->session->s_mdsc, ci);
1537	spin_unlock(lock: &ci->i_ceph_lock);
1538	return;
1539	}
1540
1541	encode_cap_msg(msg, arg);
1542	ceph_con_send(con: &arg->session->s_con, msg);
1543	ceph_buffer_put(b: arg->old_xattr_buf);
1544	ceph_buffer_put(b: arg->xattr_buf);
1545	if (arg->wake)
1546	wake_up_all(&ci->i_cap_wq);
1547	}
1548
1549	static inline int __send_flush_snap(struct inode *inode,
1550	struct ceph_mds_session *session,
1551	struct ceph_cap_snap *capsnap,
1552	u32 mseq, u64 oldest_flush_tid)
1553	{
1554	struct cap_msg_args arg;
1555	struct ceph_msg *msg;
1556
1557	arg.session = session;
1558	arg.ino = ceph_vino(inode).ino;
1559	arg.cid = 0;
1560	arg.follows = capsnap->follows;
1561	arg.flush_tid = capsnap->cap_flush.tid;
1562	arg.oldest_flush_tid = oldest_flush_tid;
1563
1564	arg.size = capsnap->size;
1565	arg.max_size = 0;
1566	arg.xattr_version = capsnap->xattr_version;
1567	arg.xattr_buf = capsnap->xattr_blob;
1568	arg.old_xattr_buf = NULL;
1569
1570	arg.atime = capsnap->atime;
1571	arg.mtime = capsnap->mtime;
1572	arg.ctime = capsnap->ctime;
1573	arg.btime = capsnap->btime;
1574	arg.change_attr = capsnap->change_attr;
1575
1576	arg.op = CEPH_CAP_OP_FLUSHSNAP;
1577	arg.caps = capsnap->issued;
1578	arg.wanted = 0;
1579	arg.dirty = capsnap->dirty;
1580
1581	arg.seq = 0;
1582	arg.issue_seq = 0;
1583	arg.mseq = mseq;
1584	arg.time_warp_seq = capsnap->time_warp_seq;
1585
1586	arg.uid = capsnap->uid;
1587	arg.gid = capsnap->gid;
1588	arg.mode = capsnap->mode;
1589
1590	arg.inline_data = capsnap->inline_data;
1591	arg.flags = 0;
1592	arg.wake = false;
1593	arg.encrypted = IS_ENCRYPTED(inode);
1594
1595	/* No fscrypt_auth changes from a capsnap.*/
1596	arg.fscrypt_auth_len = 0;
1597
1598	msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPS, front_len: cap_msg_size(arg: &arg),
1599	GFP_NOFS, can_fail: false);
1600	if (!msg)
1601	return -ENOMEM;
1602
1603	encode_cap_msg(msg, arg: &arg);
1604	ceph_con_send(con: &arg.session->s_con, msg);
1605	return 0;
1606	}
1607
1608	/*
1609	* When a snapshot is taken, clients accumulate dirty metadata on
1610	* inodes with capabilities in ceph_cap_snaps to describe the file
1611	* state at the time the snapshot was taken. This must be flushed
1612	* asynchronously back to the MDS once sync writes complete and dirty
1613	* data is written out.
1614	*
1615	* Called under i_ceph_lock.
1616	*/
1617	static void __ceph_flush_snaps(struct ceph_inode_info *ci,
1618	struct ceph_mds_session *session)
1619	__releases(ci->i_ceph_lock)
1620	__acquires(ci->i_ceph_lock)
1621	{
1622	struct inode *inode = &ci->netfs.inode;
1623	struct ceph_mds_client *mdsc = session->s_mdsc;
1624	struct ceph_client *cl = mdsc->fsc->client;
1625	struct ceph_cap_snap *capsnap;
1626	u64 oldest_flush_tid = 0;
1627	u64 first_tid = 1, last_tid = 0;
1628
1629	doutc(cl, "%p %llx.%llx session %p\n", inode, ceph_vinop(inode),
1630	session);
1631
1632	list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
1633	/*
1634	* we need to wait for sync writes to complete and for dirty
1635	* pages to be written out.
1636	*/
1637	if (capsnap->dirty_pages || capsnap->writing)
1638	break;
1639
1640	/* should be removed by ceph_try_drop_cap_snap() */
1641	BUG_ON(!capsnap->need_flush);
1642
1643	/* only flush each capsnap once */
1644	if (capsnap->cap_flush.tid > 0) {
1645	doutc(cl, "already flushed %p, skipping\n", capsnap);
1646	continue;
1647	}
1648
1649	spin_lock(lock: &mdsc->cap_dirty_lock);
1650	capsnap->cap_flush.tid = ++mdsc->last_cap_flush_tid;
1651	list_add_tail(new: &capsnap->cap_flush.g_list,
1652	head: &mdsc->cap_flush_list);
1653	if (oldest_flush_tid == 0)
1654	oldest_flush_tid = __get_oldest_flush_tid(mdsc);
1655	if (list_empty(head: &ci->i_flushing_item)) {
1656	list_add_tail(new: &ci->i_flushing_item,
1657	head: &session->s_cap_flushing);
1658	}
1659	spin_unlock(lock: &mdsc->cap_dirty_lock);
1660
1661	list_add_tail(new: &capsnap->cap_flush.i_list,
1662	head: &ci->i_cap_flush_list);
1663
1664	if (first_tid == 1)
1665	first_tid = capsnap->cap_flush.tid;
1666	last_tid = capsnap->cap_flush.tid;
1667	}
1668
1669	ci->i_ceph_flags &= ~CEPH_I_FLUSH_SNAPS;
1670
1671	while (first_tid <= last_tid) {
1672	struct ceph_cap *cap = ci->i_auth_cap;
1673	struct ceph_cap_flush *cf = NULL, *iter;
1674	int ret;
1675
1676	if (!(cap && cap->session == session)) {
1677	doutc(cl, "%p %llx.%llx auth cap %p not mds%d, stop\n",
1678	inode, ceph_vinop(inode), cap, session->s_mds);
1679	break;
1680	}
1681
1682	ret = -ENOENT;
1683	list_for_each_entry(iter, &ci->i_cap_flush_list, i_list) {
1684	if (iter->tid >= first_tid) {
1685	cf = iter;
1686	ret = 0;
1687	break;
1688	}
1689	}
1690	if (ret < 0)
1691	break;
1692
1693	first_tid = cf->tid + 1;
1694
1695	capsnap = container_of(cf, struct ceph_cap_snap, cap_flush);
1696	refcount_inc(r: &capsnap->nref);
1697	spin_unlock(lock: &ci->i_ceph_lock);
1698
1699	doutc(cl, "%p %llx.%llx capsnap %p tid %llu %s\n", inode,
1700	ceph_vinop(inode), capsnap, cf->tid,
1701	ceph_cap_string(capsnap->dirty));
1702
1703	ret = __send_flush_snap(inode, session, capsnap, mseq: cap->mseq,
1704	oldest_flush_tid);
1705	if (ret < 0) {
1706	pr_err_client(cl, "error sending cap flushsnap, "
1707	"ino (%llx.%llx) tid %llu follows %llu\n",
1708	ceph_vinop(inode), cf->tid,
1709	capsnap->follows);
1710	}
1711
1712	ceph_put_cap_snap(capsnap);
1713	spin_lock(lock: &ci->i_ceph_lock);
1714	}
1715	}
1716
1717	void ceph_flush_snaps(struct ceph_inode_info *ci,
1718	struct ceph_mds_session **psession)
1719	{
1720	struct inode *inode = &ci->netfs.inode;
1721	struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
1722	struct ceph_client *cl = ceph_inode_to_client(inode);
1723	struct ceph_mds_session *session = NULL;
1724	bool need_put = false;
1725	int mds;
1726
1727	doutc(cl, "%p %llx.%llx\n", inode, ceph_vinop(inode));
1728	if (psession)
1729	session = *psession;
1730	retry:
1731	spin_lock(lock: &ci->i_ceph_lock);
1732	if (!(ci->i_ceph_flags & CEPH_I_FLUSH_SNAPS)) {
1733	doutc(cl, " no capsnap needs flush, doing nothing\n");
1734	goto out;
1735	}
1736	if (!ci->i_auth_cap) {
1737	doutc(cl, " no auth cap (migrating?), doing nothing\n");
1738	goto out;
1739	}
1740
1741	mds = ci->i_auth_cap->session->s_mds;
1742	if (session && session->s_mds != mds) {
1743	doutc(cl, " oops, wrong session %p mutex\n", session);
1744	ceph_put_mds_session(s: session);
1745	session = NULL;
1746	}
1747	if (!session) {
1748	spin_unlock(lock: &ci->i_ceph_lock);
1749	mutex_lock(&mdsc->mutex);
1750	session = __ceph_lookup_mds_session(mdsc, mds);
1751	mutex_unlock(lock: &mdsc->mutex);
1752	goto retry;
1753	}
1754
1755	// make sure flushsnap messages are sent in proper order.
1756	if (ci->i_ceph_flags & CEPH_I_KICK_FLUSH)
1757	__kick_flushing_caps(mdsc, session, ci, oldest_flush_tid: 0);
1758
1759	__ceph_flush_snaps(ci, session);
1760	out:
1761	spin_unlock(lock: &ci->i_ceph_lock);
1762
1763	if (psession)
1764	*psession = session;
1765	else
1766	ceph_put_mds_session(s: session);
1767	/* we flushed them all; remove this inode from the queue */
1768	spin_lock(lock: &mdsc->snap_flush_lock);
1769	if (!list_empty(head: &ci->i_snap_flush_item))
1770	need_put = true;
1771	list_del_init(entry: &ci->i_snap_flush_item);
1772	spin_unlock(lock: &mdsc->snap_flush_lock);
1773
1774	if (need_put)
1775	iput(inode);
1776	}
1777
1778	/*
1779	* Mark caps dirty. If inode is newly dirty, return the dirty flags.
1780	* Caller is then responsible for calling __mark_inode_dirty with the
1781	* returned flags value.
1782	*/
1783	int __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask,
1784	struct ceph_cap_flush **pcf)
1785	{
1786	struct ceph_mds_client *mdsc =
1787	ceph_sb_to_fs_client(sb: ci->netfs.inode.i_sb)->mdsc;
1788	struct inode *inode = &ci->netfs.inode;
1789	struct ceph_client *cl = ceph_inode_to_client(inode);
1790	int was = ci->i_dirty_caps;
1791	int dirty = 0;
1792
1793	lockdep_assert_held(&ci->i_ceph_lock);
1794
1795	if (!ci->i_auth_cap) {
1796	pr_warn_client(cl, "%p %llx.%llx mask %s, "
1797	"but no auth cap (session was closed?)\n",
1798	inode, ceph_vinop(inode),
1799	ceph_cap_string(mask));
1800	return 0;
1801	}
1802
1803	doutc(cl, "%p %llx.%llx %s dirty %s -> %s\n", inode,
1804	ceph_vinop(inode), ceph_cap_string(mask),
1805	ceph_cap_string(was), ceph_cap_string(was | mask));
1806	ci->i_dirty_caps |= mask;
1807	if (was == 0) {
1808	struct ceph_mds_session *session = ci->i_auth_cap->session;
1809
1810	WARN_ON_ONCE(ci->i_prealloc_cap_flush);
1811	swap(ci->i_prealloc_cap_flush, *pcf);
1812
1813	if (!ci->i_head_snapc) {
1814	WARN_ON_ONCE(!rwsem_is_locked(&mdsc->snap_rwsem));
1815	ci->i_head_snapc = ceph_get_snap_context(
1816	sc: ci->i_snap_realm->cached_context);
1817	}
1818	doutc(cl, "%p %llx.%llx now dirty snapc %p auth cap %p\n",
1819	inode, ceph_vinop(inode), ci->i_head_snapc,
1820	ci->i_auth_cap);
1821	BUG_ON(!list_empty(&ci->i_dirty_item));
1822	spin_lock(lock: &mdsc->cap_dirty_lock);
1823	list_add(new: &ci->i_dirty_item, head: &session->s_cap_dirty);
1824	spin_unlock(lock: &mdsc->cap_dirty_lock);
1825	if (ci->i_flushing_caps == 0) {
1826	ihold(inode);
1827	dirty |= I_DIRTY_SYNC;
1828	}
1829	} else {
1830	WARN_ON_ONCE(!ci->i_prealloc_cap_flush);
1831	}
1832	BUG_ON(list_empty(&ci->i_dirty_item));
1833	if (((was | ci->i_flushing_caps) & CEPH_CAP_FILE_BUFFER) &&
1834	(mask & CEPH_CAP_FILE_BUFFER))
1835	dirty |= I_DIRTY_DATASYNC;
1836	__cap_delay_requeue(mdsc, ci);
1837	return dirty;
1838	}
1839
1840	struct ceph_cap_flush *ceph_alloc_cap_flush(void)
1841	{
1842	struct ceph_cap_flush *cf;
1843
1844	cf = kmem_cache_alloc(ceph_cap_flush_cachep, GFP_KERNEL);
1845	if (!cf)
1846	return NULL;
1847
1848	cf->is_capsnap = false;
1849	return cf;
1850	}
1851
1852	void ceph_free_cap_flush(struct ceph_cap_flush *cf)
1853	{
1854	if (cf)
1855	kmem_cache_free(s: ceph_cap_flush_cachep, objp: cf);
1856	}
1857
1858	static u64 __get_oldest_flush_tid(struct ceph_mds_client *mdsc)
1859	{
1860	if (!list_empty(head: &mdsc->cap_flush_list)) {
1861	struct ceph_cap_flush *cf =
1862	list_first_entry(&mdsc->cap_flush_list,
1863	struct ceph_cap_flush, g_list);
1864	return cf->tid;
1865	}
1866	return 0;
1867	}
1868
1869	/*
1870	* Remove cap_flush from the mdsc's or inode's flushing cap list.
1871	* Return true if caller needs to wake up flush waiters.
1872	*/
1873	static bool __detach_cap_flush_from_mdsc(struct ceph_mds_client *mdsc,
1874	struct ceph_cap_flush *cf)
1875	{
1876	struct ceph_cap_flush *prev;
1877	bool wake = cf->wake;
1878
1879	if (wake && cf->g_list.prev != &mdsc->cap_flush_list) {
1880	prev = list_prev_entry(cf, g_list);
1881	prev->wake = true;
1882	wake = false;
1883	}
1884	list_del_init(entry: &cf->g_list);
1885	return wake;
1886	}
1887
1888	static bool __detach_cap_flush_from_ci(struct ceph_inode_info *ci,
1889	struct ceph_cap_flush *cf)
1890	{
1891	struct ceph_cap_flush *prev;
1892	bool wake = cf->wake;
1893
1894	if (wake && cf->i_list.prev != &ci->i_cap_flush_list) {
1895	prev = list_prev_entry(cf, i_list);
1896	prev->wake = true;
1897	wake = false;
1898	}
1899	list_del_init(entry: &cf->i_list);
1900	return wake;
1901	}
1902
1903	/*
1904	* Add dirty inode to the flushing list. Assigned a seq number so we
1905	* can wait for caps to flush without starving.
1906	*
1907	* Called under i_ceph_lock. Returns the flush tid.
1908	*/
1909	static u64 __mark_caps_flushing(struct inode *inode,
1910	struct ceph_mds_session *session, bool wake,
1911	u64 *oldest_flush_tid)
1912	{
1913	struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(sb: inode->i_sb)->mdsc;
1914	struct ceph_client *cl = ceph_inode_to_client(inode);
1915	struct ceph_inode_info *ci = ceph_inode(inode);
1916	struct ceph_cap_flush *cf = NULL;
1917	int flushing;
1918
1919	lockdep_assert_held(&ci->i_ceph_lock);
1920	BUG_ON(ci->i_dirty_caps == 0);
1921	BUG_ON(list_empty(&ci->i_dirty_item));
1922	BUG_ON(!ci->i_prealloc_cap_flush);
1923
1924	flushing = ci->i_dirty_caps;
1925	doutc(cl, "flushing %s, flushing_caps %s -> %s\n",
1926	ceph_cap_string(flushing),
1927	ceph_cap_string(ci->i_flushing_caps),
1928	ceph_cap_string(ci->i_flushing_caps | flushing));
1929	ci->i_flushing_caps |= flushing;
1930	ci->i_dirty_caps = 0;
1931	doutc(cl, "%p %llx.%llx now !dirty\n", inode, ceph_vinop(inode));
1932
1933	swap(cf, ci->i_prealloc_cap_flush);
1934	cf->caps = flushing;
1935	cf->wake = wake;
1936
1937	spin_lock(lock: &mdsc->cap_dirty_lock);
1938	list_del_init(entry: &ci->i_dirty_item);
1939
1940	cf->tid = ++mdsc->last_cap_flush_tid;
1941	list_add_tail(new: &cf->g_list, head: &mdsc->cap_flush_list);
1942	*oldest_flush_tid = __get_oldest_flush_tid(mdsc);
1943
1944	if (list_empty(head: &ci->i_flushing_item)) {
1945	list_add_tail(new: &ci->i_flushing_item, head: &session->s_cap_flushing);
1946	mdsc->num_cap_flushing++;
1947	}
1948	spin_unlock(lock: &mdsc->cap_dirty_lock);
1949
1950	list_add_tail(new: &cf->i_list, head: &ci->i_cap_flush_list);
1951
1952	return cf->tid;
1953	}
1954
1955	/*
1956	* try to invalidate mapping pages without blocking.
1957	*/
1958	static int try_nonblocking_invalidate(struct inode *inode)
1959	__releases(ci->i_ceph_lock)
1960	__acquires(ci->i_ceph_lock)
1961	{
1962	struct ceph_client *cl = ceph_inode_to_client(inode);
1963	struct ceph_inode_info *ci = ceph_inode(inode);
1964	u32 invalidating_gen = ci->i_rdcache_gen;
1965
1966	spin_unlock(lock: &ci->i_ceph_lock);
1967	ceph_fscache_invalidate(inode, dio_write: false);
1968	invalidate_mapping_pages(mapping: &inode->i_data, start: 0, end: -1);
1969	spin_lock(lock: &ci->i_ceph_lock);
1970
1971	if (inode->i_data.nrpages == 0 &&
1972	invalidating_gen == ci->i_rdcache_gen) {
1973	/* success. */
1974	doutc(cl, "%p %llx.%llx success\n", inode,
1975	ceph_vinop(inode));
1976	/* save any racing async invalidate some trouble */
1977	ci->i_rdcache_revoking = ci->i_rdcache_gen - 1;
1978	return 0;
1979	}
1980	doutc(cl, "%p %llx.%llx failed\n", inode, ceph_vinop(inode));
1981	return -1;
1982	}
1983
1984	bool __ceph_should_report_size(struct ceph_inode_info *ci)
1985	{
1986	loff_t size = i_size_read(inode: &ci->netfs.inode);
1987	/* mds will adjust max size according to the reported size */
1988	if (ci->i_flushing_caps & CEPH_CAP_FILE_WR)
1989	return false;
1990	if (size >= ci->i_max_size)
1991	return true;
1992	/* half of previous max_size increment has been used */
1993	if (ci->i_max_size > ci->i_reported_size &&
1994	(size << 1) >= ci->i_max_size + ci->i_reported_size)
1995	return true;
1996	return false;
1997	}
1998
1999	/*
2000	* Swiss army knife function to examine currently used and wanted
2001	* versus held caps. Release, flush, ack revoked caps to mds as
2002	* appropriate.
2003	*
2004	* CHECK_CAPS_AUTHONLY - we should only check the auth cap
2005	* CHECK_CAPS_FLUSH - we should flush any dirty caps immediately, without
2006	* further delay.
2007	* CHECK_CAPS_FLUSH_FORCE - we should flush any caps immediately, without
2008	* further delay.
2009	*/
2010	void ceph_check_caps(struct ceph_inode_info *ci, int flags)
2011	{
2012	struct inode *inode = &ci->netfs.inode;
2013	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(sb: inode->i_sb);
2014	struct ceph_client *cl = ceph_inode_to_client(inode);
2015	struct ceph_cap *cap;
2016	u64 flush_tid, oldest_flush_tid;
2017	int file_wanted, used, cap_used;
2018	int issued, implemented, want, retain, revoking, flushing = 0;
2019	int mds = -1; /* keep track of how far we've gone through i_caps list
2020	to avoid an infinite loop on retry */
2021	struct rb_node *p;
2022	bool queue_invalidate = false;
2023	bool tried_invalidate = false;
2024	bool queue_writeback = false;
2025	struct ceph_mds_session *session = NULL;
2026
2027	spin_lock(lock: &ci->i_ceph_lock);
2028	if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE) {
2029	ci->i_ceph_flags |= CEPH_I_ASYNC_CHECK_CAPS;
2030
2031	/* Don't send messages until we get async create reply */
2032	spin_unlock(lock: &ci->i_ceph_lock);
2033	return;
2034	}
2035
2036	if (ci->i_ceph_flags & CEPH_I_FLUSH)
2037	flags |= CHECK_CAPS_FLUSH;
2038	retry:
2039	/* Caps wanted by virtue of active open files. */
2040	file_wanted = __ceph_caps_file_wanted(ci);
2041
2042	/* Caps which have active references against them */
2043	used = __ceph_caps_used(ci);
2044
2045	/*
2046	* "issued" represents the current caps that the MDS wants us to have.
2047	* "implemented" is the set that we have been granted, and includes the
2048	* ones that have not yet been returned to the MDS (the "revoking" set,
2049	* usually because they have outstanding references).
2050	*/
2051	issued = __ceph_caps_issued(ci, implemented: &implemented);
2052	revoking = implemented & ~issued;
2053
2054	want = file_wanted;
2055
2056	/* The ones we currently want to retain (may be adjusted below) */
2057	retain = file_wanted | used | CEPH_CAP_PIN;
2058	if (!mdsc->stopping && inode->i_nlink > 0) {
2059	if (file_wanted) {
2060	retain |= CEPH_CAP_ANY; /* be greedy */
2061	} else if (S_ISDIR(inode->i_mode) &&
2062	(issued & CEPH_CAP_FILE_SHARED) &&
2063	__ceph_dir_is_complete(ci)) {
2064	/*
2065	* If a directory is complete, we want to keep
2066	* the exclusive cap. So that MDS does not end up
2067	* revoking the shared cap on every create/unlink
2068	* operation.
2069	*/
2070	if (IS_RDONLY(inode)) {
2071	want = CEPH_CAP_ANY_SHARED;
2072	} else {
2073	want |= CEPH_CAP_ANY_SHARED | CEPH_CAP_FILE_EXCL;
2074	}
2075	retain |= want;
2076	} else {
2077
2078	retain |= CEPH_CAP_ANY_SHARED;
2079	/*
2080	* keep RD only if we didn't have the file open RW,
2081	* because then the mds would revoke it anyway to
2082	* journal max_size=0.
2083	*/
2084	if (ci->i_max_size == 0)
2085	retain |= CEPH_CAP_ANY_RD;
2086	}
2087	}
2088
2089	doutc(cl, "%p %llx.%llx file_want %s used %s dirty %s "
2090	"flushing %s issued %s revoking %s retain %s %s%s%s%s\n",
2091	inode, ceph_vinop(inode), ceph_cap_string(file_wanted),
2092	ceph_cap_string(used), ceph_cap_string(ci->i_dirty_caps),
2093	ceph_cap_string(ci->i_flushing_caps),
2094	ceph_cap_string(issued), ceph_cap_string(revoking),
2095	ceph_cap_string(retain),
2096	(flags & CHECK_CAPS_AUTHONLY) ? " AUTHONLY" : "",
2097	(flags & CHECK_CAPS_FLUSH) ? " FLUSH" : "",
2098	(flags & CHECK_CAPS_NOINVAL) ? " NOINVAL" : "",
2099	(flags & CHECK_CAPS_FLUSH_FORCE) ? " FLUSH_FORCE" : "");
2100
2101	/*
2102	* If we no longer need to hold onto old our caps, and we may
2103	* have cached pages, but don't want them, then try to invalidate.
2104	* If we fail, it's because pages are locked.... try again later.
2105	*/
2106	if ((!(flags & CHECK_CAPS_NOINVAL) || mdsc->stopping) &&
2107	S_ISREG(inode->i_mode) &&
2108	!(ci->i_wb_ref || ci->i_wrbuffer_ref) && /* no dirty pages... */
2109	inode->i_data.nrpages && /* have cached pages */
2110	(revoking & (CEPH_CAP_FILE_CACHE|
2111	CEPH_CAP_FILE_LAZYIO)) && /* or revoking cache */
2112	!tried_invalidate) {
2113	doutc(cl, "trying to invalidate on %p %llx.%llx\n",
2114	inode, ceph_vinop(inode));
2115	if (try_nonblocking_invalidate(inode) < 0) {
2116	doutc(cl, "queuing invalidate\n");
2117	queue_invalidate = true;
2118	ci->i_rdcache_revoking = ci->i_rdcache_gen;
2119	}
2120	tried_invalidate = true;
2121	goto retry;
2122	}
2123
2124	for (p = rb_first(root: &ci->i_caps); p; p = rb_next(p)) {
2125	int mflags = 0;
2126	struct cap_msg_args arg;
2127
2128	cap = rb_entry(p, struct ceph_cap, ci_node);
2129
2130	/* avoid looping forever */
2131	if (mds >= cap->mds ||
2132	((flags & CHECK_CAPS_AUTHONLY) && cap != ci->i_auth_cap))
2133	continue;
2134
2135	/*
2136	* If we have an auth cap, we don't need to consider any
2137	* overlapping caps as used.
2138	*/
2139	cap_used = used;
2140	if (ci->i_auth_cap && cap != ci->i_auth_cap)
2141	cap_used &= ~ci->i_auth_cap->issued;
2142
2143	revoking = cap->implemented & ~cap->issued;
2144	doutc(cl, " mds%d cap %p used %s issued %s implemented %s revoking %s\n",
2145	cap->mds, cap, ceph_cap_string(cap_used),
2146	ceph_cap_string(cap->issued),
2147	ceph_cap_string(cap->implemented),
2148	ceph_cap_string(revoking));
2149
2150	/* completed revocation? going down and there are no caps? */
2151	if (revoking) {
2152	if ((revoking & cap_used) == 0) {
2153	doutc(cl, "completed revocation of %s\n",
2154	ceph_cap_string(cap->implemented & ~cap->issued));
2155	goto ack;
2156	}
2157
2158	/*
2159	* If the "i_wrbuffer_ref" was increased by mmap or generic
2160	* cache write just before the ceph_check_caps() is called,
2161	* the Fb capability revoking will fail this time. Then we
2162	* must wait for the BDI's delayed work to flush the dirty
2163	* pages and to release the "i_wrbuffer_ref", which will cost
2164	* at most 5 seconds. That means the MDS needs to wait at
2165	* most 5 seconds to finished the Fb capability's revocation.
2166	*
2167	* Let's queue a writeback for it.
2168	*/
2169	if (S_ISREG(inode->i_mode) && ci->i_wrbuffer_ref &&
2170	(revoking & CEPH_CAP_FILE_BUFFER))
2171	queue_writeback = true;
2172	}
2173
2174	if (flags & CHECK_CAPS_FLUSH_FORCE) {
2175	doutc(cl, "force to flush caps\n");
2176	goto ack;
2177	}
2178
2179	if (cap == ci->i_auth_cap &&
2180	(cap->issued & CEPH_CAP_FILE_WR)) {
2181	/* request larger max_size from MDS? */
2182	if (ci->i_wanted_max_size > ci->i_max_size &&
2183	ci->i_wanted_max_size > ci->i_requested_max_size) {
2184	doutc(cl, "requesting new max_size\n");
2185	goto ack;
2186	}
2187
2188	/* approaching file_max? */
2189	if (__ceph_should_report_size(ci)) {
2190	doutc(cl, "i_size approaching max_size\n");
2191	goto ack;
2192	}
2193	}
2194	/* flush anything dirty? */
2195	if (cap == ci->i_auth_cap) {
2196	if ((flags & CHECK_CAPS_FLUSH) && ci->i_dirty_caps) {
2197	doutc(cl, "flushing dirty caps\n");
2198	goto ack;
2199	}
2200	if (ci->i_ceph_flags & CEPH_I_FLUSH_SNAPS) {
2201	doutc(cl, "flushing snap caps\n");
2202	goto ack;
2203	}
2204	}
2205
2206	/* want more caps from mds? */
2207	if (want & ~cap->mds_wanted) {
2208	if (want & ~(cap->mds_wanted | cap->issued))
2209	goto ack;
2210	if (!__cap_is_valid(cap))
2211	goto ack;
2212	}
2213
2214	/* things we might delay */
2215	if ((cap->issued & ~retain) == 0)
2216	continue; /* nope, all good */
2217
2218	ack:
2219	ceph_put_mds_session(s: session);
2220	session = ceph_get_mds_session(s: cap->session);
2221
2222	/* kick flushing and flush snaps before sending normal
2223	* cap message */
2224	if (cap == ci->i_auth_cap &&
2225	(ci->i_ceph_flags &
2226	(CEPH_I_KICK_FLUSH | CEPH_I_FLUSH_SNAPS))) {
2227	if (ci->i_ceph_flags & CEPH_I_KICK_FLUSH)
2228	__kick_flushing_caps(mdsc, session, ci, oldest_flush_tid: 0);
2229	if (ci->i_ceph_flags & CEPH_I_FLUSH_SNAPS)
2230	__ceph_flush_snaps(ci, session);
2231
2232	goto retry;
2233	}
2234
2235	if (cap == ci->i_auth_cap && ci->i_dirty_caps) {
2236	flushing = ci->i_dirty_caps;
2237	flush_tid = __mark_caps_flushing(inode, session, wake: false,
2238	oldest_flush_tid: &oldest_flush_tid);
2239	if (flags & CHECK_CAPS_FLUSH &&
2240	list_empty(head: &session->s_cap_dirty))
2241	mflags |= CEPH_CLIENT_CAPS_SYNC;
2242	} else {
2243	flushing = 0;
2244	flush_tid = 0;
2245	spin_lock(lock: &mdsc->cap_dirty_lock);
2246	oldest_flush_tid = __get_oldest_flush_tid(mdsc);
2247	spin_unlock(lock: &mdsc->cap_dirty_lock);
2248	}
2249
2250	mds = cap->mds; /* remember mds, so we don't repeat */
2251
2252	__prep_cap(arg: &arg, cap, op: CEPH_CAP_OP_UPDATE, flags: mflags, used: cap_used,
2253	want, retain, flushing, flush_tid, oldest_flush_tid);
2254
2255	spin_unlock(lock: &ci->i_ceph_lock);
2256	__send_cap(arg: &arg, ci);
2257	spin_lock(lock: &ci->i_ceph_lock);
2258
2259	goto retry; /* retake i_ceph_lock and restart our cap scan. */
2260	}
2261
2262	/* periodically re-calculate caps wanted by open files */
2263	if (__ceph_is_any_real_caps(ci) &&
2264	list_empty(head: &ci->i_cap_delay_list) &&
2265	(file_wanted & ~CEPH_CAP_PIN) &&
2266	!(used & (CEPH_CAP_FILE_RD | CEPH_CAP_ANY_FILE_WR))) {
2267	__cap_delay_requeue(mdsc, ci);
2268	}
2269
2270	spin_unlock(lock: &ci->i_ceph_lock);
2271
2272	ceph_put_mds_session(s: session);
2273	if (queue_writeback)
2274	ceph_queue_writeback(inode);
2275	if (queue_invalidate)
2276	ceph_queue_invalidate(inode);
2277	}
2278
2279	/*
2280	* Try to flush dirty caps back to the auth mds.
2281	*/
2282	static int try_flush_caps(struct inode *inode, u64 *ptid)
2283	{
2284	struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(sb: inode->i_sb)->mdsc;
2285	struct ceph_inode_info *ci = ceph_inode(inode);
2286	int flushing = 0;
2287	u64 flush_tid = 0, oldest_flush_tid = 0;
2288
2289	spin_lock(lock: &ci->i_ceph_lock);
2290	retry_locked:
2291	if (ci->i_dirty_caps && ci->i_auth_cap) {
2292	struct ceph_cap *cap = ci->i_auth_cap;
2293	struct cap_msg_args arg;
2294	struct ceph_mds_session *session = cap->session;
2295
2296	if (session->s_state < CEPH_MDS_SESSION_OPEN) {
2297	spin_unlock(lock: &ci->i_ceph_lock);
2298	goto out;
2299	}
2300
2301	if (ci->i_ceph_flags &
2302	(CEPH_I_KICK_FLUSH | CEPH_I_FLUSH_SNAPS)) {
2303	if (ci->i_ceph_flags & CEPH_I_KICK_FLUSH)
2304	__kick_flushing_caps(mdsc, session, ci, oldest_flush_tid: 0);
2305	if (ci->i_ceph_flags & CEPH_I_FLUSH_SNAPS)
2306	__ceph_flush_snaps(ci, session);
2307	goto retry_locked;
2308	}
2309
2310	flushing = ci->i_dirty_caps;
2311	flush_tid = __mark_caps_flushing(inode, session, wake: true,
2312	oldest_flush_tid: &oldest_flush_tid);
2313
2314	__prep_cap(arg: &arg, cap, op: CEPH_CAP_OP_FLUSH, CEPH_CLIENT_CAPS_SYNC,
2315	used: __ceph_caps_used(ci), want: __ceph_caps_wanted(ci),
2316	retain: (cap->issued | cap->implemented),
2317	flushing, flush_tid, oldest_flush_tid);
2318	spin_unlock(lock: &ci->i_ceph_lock);
2319
2320	__send_cap(arg: &arg, ci);
2321	} else {
2322	if (!list_empty(head: &ci->i_cap_flush_list)) {
2323	struct ceph_cap_flush *cf =
2324	list_last_entry(&ci->i_cap_flush_list,
2325	struct ceph_cap_flush, i_list);
2326	cf->wake = true;
2327	flush_tid = cf->tid;
2328	}
2329	flushing = ci->i_flushing_caps;
2330	spin_unlock(lock: &ci->i_ceph_lock);
2331	}
2332	out:
2333	*ptid = flush_tid;
2334	return flushing;
2335	}
2336
2337	/*
2338	* Return true if we've flushed caps through the given flush_tid.
2339	*/
2340	static int caps_are_flushed(struct inode *inode, u64 flush_tid)
2341	{
2342	struct ceph_inode_info *ci = ceph_inode(inode);
2343	int ret = 1;
2344
2345	spin_lock(lock: &ci->i_ceph_lock);
2346	if (!list_empty(head: &ci->i_cap_flush_list)) {
2347	struct ceph_cap_flush * cf =
2348	list_first_entry(&ci->i_cap_flush_list,
2349	struct ceph_cap_flush, i_list);
2350	if (cf->tid <= flush_tid)
2351	ret = 0;
2352	}
2353	spin_unlock(lock: &ci->i_ceph_lock);
2354	return ret;
2355	}
2356
2357	/*
2358	* flush the mdlog and wait for any unsafe requests to complete.
2359	*/
2360	static int flush_mdlog_and_wait_inode_unsafe_requests(struct inode *inode)
2361	{
2362	struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(sb: inode->i_sb)->mdsc;
2363	struct ceph_client *cl = ceph_inode_to_client(inode);
2364	struct ceph_inode_info *ci = ceph_inode(inode);
2365	struct ceph_mds_request *req1 = NULL, *req2 = NULL;
2366	int ret, err = 0;
2367
2368	spin_lock(lock: &ci->i_unsafe_lock);
2369	if (S_ISDIR(inode->i_mode) && !list_empty(head: &ci->i_unsafe_dirops)) {
2370	req1 = list_last_entry(&ci->i_unsafe_dirops,
2371	struct ceph_mds_request,
2372	r_unsafe_dir_item);
2373	ceph_mdsc_get_request(req: req1);
2374	}
2375	if (!list_empty(head: &ci->i_unsafe_iops)) {
2376	req2 = list_last_entry(&ci->i_unsafe_iops,
2377	struct ceph_mds_request,
2378	r_unsafe_target_item);
2379	ceph_mdsc_get_request(req: req2);
2380	}
2381	spin_unlock(lock: &ci->i_unsafe_lock);
2382
2383	/*
2384	* Trigger to flush the journal logs in all the relevant MDSes
2385	* manually, or in the worst case we must wait at most 5 seconds
2386	* to wait the journal logs to be flushed by the MDSes periodically.
2387	*/
2388	if (req1 || req2) {
2389	struct ceph_mds_request *req;
2390	struct ceph_mds_session **sessions;
2391	struct ceph_mds_session *s;
2392	unsigned int max_sessions;
2393	int i;
2394
2395	mutex_lock(&mdsc->mutex);
2396	max_sessions = mdsc->max_sessions;
2397
2398	sessions = kcalloc(max_sessions, sizeof(s), GFP_KERNEL);
2399	if (!sessions) {
2400	mutex_unlock(lock: &mdsc->mutex);
2401	err = -ENOMEM;
2402	goto out;
2403	}
2404
2405	spin_lock(lock: &ci->i_unsafe_lock);
2406	if (req1) {
2407	list_for_each_entry(req, &ci->i_unsafe_dirops,
2408	r_unsafe_dir_item) {
2409	s = req->r_session;
2410	if (!s)
2411	continue;
2412	if (!sessions[s->s_mds]) {
2413	s = ceph_get_mds_session(s);
2414	sessions[s->s_mds] = s;
2415	}
2416	}
2417	}
2418	if (req2) {
2419	list_for_each_entry(req, &ci->i_unsafe_iops,
2420	r_unsafe_target_item) {
2421	s = req->r_session;
2422	if (!s)
2423	continue;
2424	if (!sessions[s->s_mds]) {
2425	s = ceph_get_mds_session(s);
2426	sessions[s->s_mds] = s;
2427	}
2428	}
2429	}
2430	spin_unlock(lock: &ci->i_unsafe_lock);
2431
2432	/* the auth MDS */
2433	spin_lock(lock: &ci->i_ceph_lock);
2434	if (ci->i_auth_cap) {
2435	s = ci->i_auth_cap->session;
2436	if (!sessions[s->s_mds])
2437	sessions[s->s_mds] = ceph_get_mds_session(s);
2438	}
2439	spin_unlock(lock: &ci->i_ceph_lock);
2440	mutex_unlock(lock: &mdsc->mutex);
2441
2442	/* send flush mdlog request to MDSes */
2443	for (i = 0; i < max_sessions; i++) {
2444	s = sessions[i];
2445	if (s) {
2446	send_flush_mdlog(s);
2447	ceph_put_mds_session(s);
2448	}
2449	}
2450	kfree(objp: sessions);
2451	}
2452
2453	doutc(cl, "%p %llx.%llx wait on tid %llu %llu\n", inode,
2454	ceph_vinop(inode), req1 ? req1->r_tid : 0ULL,
2455	req2 ? req2->r_tid : 0ULL);
2456	if (req1) {
2457	ret = !wait_for_completion_timeout(x: &req1->r_safe_completion,
2458	timeout: ceph_timeout_jiffies(timeout: req1->r_timeout));
2459	if (ret)
2460	err = -EIO;
2461	}
2462	if (req2) {
2463	ret = !wait_for_completion_timeout(x: &req2->r_safe_completion,
2464	timeout: ceph_timeout_jiffies(timeout: req2->r_timeout));
2465	if (ret)
2466	err = -EIO;
2467	}
2468
2469	out:
2470	if (req1)
2471	ceph_mdsc_put_request(req: req1);
2472	if (req2)
2473	ceph_mdsc_put_request(req: req2);
2474	return err;
2475	}
2476
2477	int ceph_fsync(struct file *file, loff_t start, loff_t end, int datasync)
2478	{
2479	struct inode *inode = file->f_mapping->host;
2480	struct ceph_inode_info *ci = ceph_inode(inode);
2481	struct ceph_client *cl = ceph_inode_to_client(inode);
2482	u64 flush_tid;
2483	int ret, err;
2484	int dirty;
2485
2486	doutc(cl, "%p %llx.%llx%s\n", inode, ceph_vinop(inode),
2487	datasync ? " datasync" : "");
2488
2489	ret = file_write_and_wait_range(file, start, end);
2490	if (datasync)
2491	goto out;
2492
2493	ret = ceph_wait_on_async_create(inode);
2494	if (ret)
2495	goto out;
2496
2497	dirty = try_flush_caps(inode, ptid: &flush_tid);
2498	doutc(cl, "dirty caps are %s\n", ceph_cap_string(dirty));
2499
2500	err = flush_mdlog_and_wait_inode_unsafe_requests(inode);
2501
2502	/*
2503	* only wait on non-file metadata writeback (the mds
2504	* can recover size and mtime, so we don't need to
2505	* wait for that)
2506	*/
2507	if (!err && (dirty & ~CEPH_CAP_ANY_FILE_WR)) {
2508	err = wait_event_interruptible(ci->i_cap_wq,
2509	caps_are_flushed(inode, flush_tid));
2510	}
2511
2512	if (err < 0)
2513	ret = err;
2514
2515	err = file_check_and_advance_wb_err(file);
2516	if (err < 0)
2517	ret = err;
2518	out:
2519	doutc(cl, "%p %llx.%llx%s result=%d\n", inode, ceph_vinop(inode),
2520	datasync ? " datasync" : "", ret);
2521	return ret;
2522	}
2523
2524	/*
2525	* Flush any dirty caps back to the mds. If we aren't asked to wait,
2526	* queue inode for flush but don't do so immediately, because we can
2527	* get by with fewer MDS messages if we wait for data writeback to
2528	* complete first.
2529	*/
2530	int ceph_write_inode(struct inode *inode, struct writeback_control *wbc)
2531	{
2532	struct ceph_inode_info *ci = ceph_inode(inode);
2533	struct ceph_client *cl = ceph_inode_to_client(inode);
2534	u64 flush_tid;
2535	int err = 0;
2536	int dirty;
2537	int wait = (wbc->sync_mode == WB_SYNC_ALL && !wbc->for_sync);
2538
2539	doutc(cl, "%p %llx.%llx wait=%d\n", inode, ceph_vinop(inode), wait);
2540	ceph_fscache_unpin_writeback(inode, wbc);
2541	if (wait) {
2542	err = ceph_wait_on_async_create(inode);
2543	if (err)
2544	return err;
2545	dirty = try_flush_caps(inode, ptid: &flush_tid);
2546	if (dirty)
2547	err = wait_event_interruptible(ci->i_cap_wq,
2548	caps_are_flushed(inode, flush_tid));
2549	} else {
2550	struct ceph_mds_client *mdsc =
2551	ceph_sb_to_fs_client(sb: inode->i_sb)->mdsc;
2552
2553	spin_lock(lock: &ci->i_ceph_lock);
2554	if (__ceph_caps_dirty(ci))
2555	__cap_delay_requeue_front(mdsc, ci);
2556	spin_unlock(lock: &ci->i_ceph_lock);
2557	}
2558	return err;
2559	}
2560
2561	static void __kick_flushing_caps(struct ceph_mds_client *mdsc,
2562	struct ceph_mds_session *session,
2563	struct ceph_inode_info *ci,
2564	u64 oldest_flush_tid)
2565	__releases(ci->i_ceph_lock)
2566	__acquires(ci->i_ceph_lock)
2567	{
2568	struct inode *inode = &ci->netfs.inode;
2569	struct ceph_client *cl = mdsc->fsc->client;
2570	struct ceph_cap *cap;
2571	struct ceph_cap_flush *cf;
2572	int ret;
2573	u64 first_tid = 0;
2574	u64 last_snap_flush = 0;
2575
2576	/* Don't do anything until create reply comes in */
2577	if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE)
2578	return;
2579
2580	ci->i_ceph_flags &= ~CEPH_I_KICK_FLUSH;
2581
2582	list_for_each_entry_reverse(cf, &ci->i_cap_flush_list, i_list) {
2583	if (cf->is_capsnap) {
2584	last_snap_flush = cf->tid;
2585	break;
2586	}
2587	}
2588
2589	list_for_each_entry(cf, &ci->i_cap_flush_list, i_list) {
2590	if (cf->tid < first_tid)
2591	continue;
2592
2593	cap = ci->i_auth_cap;
2594	if (!(cap && cap->session == session)) {
2595	pr_err_client(cl, "%p auth cap %p not mds%d ???\n",
2596	inode, cap, session->s_mds);
2597	break;
2598	}
2599
2600	first_tid = cf->tid + 1;
2601
2602	if (!cf->is_capsnap) {
2603	struct cap_msg_args arg;
2604
2605	doutc(cl, "%p %llx.%llx cap %p tid %llu %s\n",
2606	inode, ceph_vinop(inode), cap, cf->tid,
2607	ceph_cap_string(cf->caps));
2608	__prep_cap(arg: &arg, cap, op: CEPH_CAP_OP_FLUSH,
2609	flags: (cf->tid < last_snap_flush ?
2610	CEPH_CLIENT_CAPS_PENDING_CAPSNAP : 0),
2611	used: __ceph_caps_used(ci),
2612	want: __ceph_caps_wanted(ci),
2613	retain: (cap->issued | cap->implemented),
2614	flushing: cf->caps, flush_tid: cf->tid, oldest_flush_tid);
2615	spin_unlock(lock: &ci->i_ceph_lock);
2616	__send_cap(arg: &arg, ci);
2617	} else {
2618	struct ceph_cap_snap *capsnap =
2619	container_of(cf, struct ceph_cap_snap,
2620	cap_flush);
2621	doutc(cl, "%p %llx.%llx capsnap %p tid %llu %s\n",
2622	inode, ceph_vinop(inode), capsnap, cf->tid,
2623	ceph_cap_string(capsnap->dirty));
2624
2625	refcount_inc(r: &capsnap->nref);
2626	spin_unlock(lock: &ci->i_ceph_lock);
2627
2628	ret = __send_flush_snap(inode, session, capsnap, mseq: cap->mseq,
2629	oldest_flush_tid);
2630	if (ret < 0) {
2631	pr_err_client(cl, "error sending cap flushsnap,"
2632	" %p %llx.%llx tid %llu follows %llu\n",
2633	inode, ceph_vinop(inode), cf->tid,
2634	capsnap->follows);
2635	}
2636
2637	ceph_put_cap_snap(capsnap);
2638	}
2639
2640	spin_lock(lock: &ci->i_ceph_lock);
2641	}
2642	}
2643
2644	void ceph_early_kick_flushing_caps(struct ceph_mds_client *mdsc,
2645	struct ceph_mds_session *session)
2646	{
2647	struct ceph_client *cl = mdsc->fsc->client;
2648	struct ceph_inode_info *ci;
2649	struct ceph_cap *cap;
2650	u64 oldest_flush_tid;
2651
2652	doutc(cl, "mds%d\n", session->s_mds);
2653
2654	spin_lock(lock: &mdsc->cap_dirty_lock);
2655	oldest_flush_tid = __get_oldest_flush_tid(mdsc);
2656	spin_unlock(lock: &mdsc->cap_dirty_lock);
2657
2658	list_for_each_entry(ci, &session->s_cap_flushing, i_flushing_item) {
2659	struct inode *inode = &ci->netfs.inode;
2660
2661	spin_lock(lock: &ci->i_ceph_lock);
2662	cap = ci->i_auth_cap;
2663	if (!(cap && cap->session == session)) {
2664	pr_err_client(cl, "%p %llx.%llx auth cap %p not mds%d ???\n",
2665	inode, ceph_vinop(inode), cap,
2666	session->s_mds);
2667	spin_unlock(lock: &ci->i_ceph_lock);
2668	continue;
2669	}
2670
2671
2672	/*
2673	* if flushing caps were revoked, we re-send the cap flush
2674	* in client reconnect stage. This guarantees MDS * processes
2675	* the cap flush message before issuing the flushing caps to
2676	* other client.
2677	*/
2678	if ((cap->issued & ci->i_flushing_caps) !=
2679	ci->i_flushing_caps) {
2680	/* encode_caps_cb() also will reset these sequence
2681	* numbers. make sure sequence numbers in cap flush
2682	* message match later reconnect message */
2683	cap->seq = 0;
2684	cap->issue_seq = 0;
2685	cap->mseq = 0;
2686	__kick_flushing_caps(mdsc, session, ci,
2687	oldest_flush_tid);
2688	} else {
2689	ci->i_ceph_flags |= CEPH_I_KICK_FLUSH;
2690	}
2691
2692	spin_unlock(lock: &ci->i_ceph_lock);
2693	}
2694	}
2695
2696	void ceph_kick_flushing_caps(struct ceph_mds_client *mdsc,
2697	struct ceph_mds_session *session)
2698	{
2699	struct ceph_client *cl = mdsc->fsc->client;
2700	struct ceph_inode_info *ci;
2701	struct ceph_cap *cap;
2702	u64 oldest_flush_tid;
2703
2704	lockdep_assert_held(&session->s_mutex);
2705
2706	doutc(cl, "mds%d\n", session->s_mds);
2707
2708	spin_lock(lock: &mdsc->cap_dirty_lock);
2709	oldest_flush_tid = __get_oldest_flush_tid(mdsc);
2710	spin_unlock(lock: &mdsc->cap_dirty_lock);
2711
2712	list_for_each_entry(ci, &session->s_cap_flushing, i_flushing_item) {
2713	struct inode *inode = &ci->netfs.inode;
2714
2715	spin_lock(lock: &ci->i_ceph_lock);
2716	cap = ci->i_auth_cap;
2717	if (!(cap && cap->session == session)) {
2718	pr_err_client(cl, "%p %llx.%llx auth cap %p not mds%d ???\n",
2719	inode, ceph_vinop(inode), cap,
2720	session->s_mds);
2721	spin_unlock(lock: &ci->i_ceph_lock);
2722	continue;
2723	}
2724	if (ci->i_ceph_flags & CEPH_I_KICK_FLUSH) {
2725	__kick_flushing_caps(mdsc, session, ci,
2726	oldest_flush_tid);
2727	}
2728	spin_unlock(lock: &ci->i_ceph_lock);
2729	}
2730	}
2731
2732	void ceph_kick_flushing_inode_caps(struct ceph_mds_session *session,
2733	struct ceph_inode_info *ci)
2734	{
2735	struct ceph_mds_client *mdsc = session->s_mdsc;
2736	struct ceph_cap *cap = ci->i_auth_cap;
2737	struct inode *inode = &ci->netfs.inode;
2738
2739	lockdep_assert_held(&ci->i_ceph_lock);
2740
2741	doutc(mdsc->fsc->client, "%p %llx.%llx flushing %s\n",
2742	inode, ceph_vinop(inode),
2743	ceph_cap_string(ci->i_flushing_caps));
2744
2745	if (!list_empty(head: &ci->i_cap_flush_list)) {
2746	u64 oldest_flush_tid;
2747	spin_lock(lock: &mdsc->cap_dirty_lock);
2748	list_move_tail(list: &ci->i_flushing_item,
2749	head: &cap->session->s_cap_flushing);
2750	oldest_flush_tid = __get_oldest_flush_tid(mdsc);
2751	spin_unlock(lock: &mdsc->cap_dirty_lock);
2752
2753	__kick_flushing_caps(mdsc, session, ci, oldest_flush_tid);
2754	}
2755	}
2756
2757
2758	/*
2759	* Take references to capabilities we hold, so that we don't release
2760	* them to the MDS prematurely.
2761	*/
2762	void ceph_take_cap_refs(struct ceph_inode_info *ci, int got,
2763	bool snap_rwsem_locked)
2764	{
2765	struct inode *inode = &ci->netfs.inode;
2766	struct ceph_client *cl = ceph_inode_to_client(inode);
2767
2768	lockdep_assert_held(&ci->i_ceph_lock);
2769
2770	if (got & CEPH_CAP_PIN)
2771	ci->i_pin_ref++;
2772	if (got & CEPH_CAP_FILE_RD)
2773	ci->i_rd_ref++;
2774	if (got & CEPH_CAP_FILE_CACHE)
2775	ci->i_rdcache_ref++;
2776	if (got & CEPH_CAP_FILE_EXCL)
2777	ci->i_fx_ref++;
2778	if (got & CEPH_CAP_FILE_WR) {
2779	if (ci->i_wr_ref == 0 && !ci->i_head_snapc) {
2780	BUG_ON(!snap_rwsem_locked);
2781	ci->i_head_snapc = ceph_get_snap_context(
2782	sc: ci->i_snap_realm->cached_context);
2783	}
2784	ci->i_wr_ref++;
2785	}
2786	if (got & CEPH_CAP_FILE_BUFFER) {
2787	if (ci->i_wb_ref == 0)
2788	ihold(inode);
2789	ci->i_wb_ref++;
2790	doutc(cl, "%p %llx.%llx wb %d -> %d (?)\n", inode,
2791	ceph_vinop(inode), ci->i_wb_ref-1, ci->i_wb_ref);
2792	}
2793	}
2794
2795	/*
2796	* Try to grab cap references. Specify those refs we @want, and the
2797	* minimal set we @need. Also include the larger offset we are writing
2798	* to (when applicable), and check against max_size here as well.
2799	* Note that caller is responsible for ensuring max_size increases are
2800	* requested from the MDS.
2801	*
2802	* Returns 0 if caps were not able to be acquired (yet), 1 if succeed,
2803	* or a negative error code. There are 3 special error codes:
2804	* -EAGAIN: need to sleep but non-blocking is specified
2805	* -EFBIG: ask caller to call check_max_size() and try again.
2806	* -EUCLEAN: ask caller to call ceph_renew_caps() and try again.
2807	*/
2808	enum {
2809	/* first 8 bits are reserved for CEPH_FILE_MODE_FOO */
2810	NON_BLOCKING = (1 << 8),
2811	CHECK_FILELOCK = (1 << 9),
2812	};
2813
2814	static int try_get_cap_refs(struct inode *inode, int need, int want,
2815	loff_t endoff, int flags, int *got)
2816	{
2817	struct ceph_inode_info *ci = ceph_inode(inode);
2818	struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
2819	struct ceph_client *cl = ceph_inode_to_client(inode);
2820	int ret = 0;
2821	int have, implemented;
2822	bool snap_rwsem_locked = false;
2823
2824	doutc(cl, "%p %llx.%llx need %s want %s\n", inode,
2825	ceph_vinop(inode), ceph_cap_string(need),
2826	ceph_cap_string(want));
2827
2828	again:
2829	spin_lock(lock: &ci->i_ceph_lock);
2830
2831	if ((flags & CHECK_FILELOCK) &&
2832	(ci->i_ceph_flags & CEPH_I_ERROR_FILELOCK)) {
2833	doutc(cl, "%p %llx.%llx error filelock\n", inode,
2834	ceph_vinop(inode));
2835	ret = -EIO;
2836	goto out_unlock;
2837	}
2838
2839	/* finish pending truncate */
2840	while (ci->i_truncate_pending) {
2841	spin_unlock(lock: &ci->i_ceph_lock);
2842	if (snap_rwsem_locked) {
2843	up_read(sem: &mdsc->snap_rwsem);
2844	snap_rwsem_locked = false;
2845	}
2846	__ceph_do_pending_vmtruncate(inode);
2847	spin_lock(lock: &ci->i_ceph_lock);
2848	}
2849
2850	have = __ceph_caps_issued(ci, implemented: &implemented);
2851
2852	if (have & need & CEPH_CAP_FILE_WR) {
2853	if (endoff >= 0 && endoff > (loff_t)ci->i_max_size) {
2854	doutc(cl, "%p %llx.%llx endoff %llu > maxsize %llu\n",
2855	inode, ceph_vinop(inode), endoff, ci->i_max_size);
2856	if (endoff > ci->i_requested_max_size)
2857	ret = ci->i_auth_cap ? -EFBIG : -EUCLEAN;
2858	goto out_unlock;
2859	}
2860	/*
2861	* If a sync write is in progress, we must wait, so that we
2862	* can get a final snapshot value for size+mtime.
2863	*/
2864	if (__ceph_have_pending_cap_snap(ci)) {
2865	doutc(cl, "%p %llx.%llx cap_snap_pending\n", inode,
2866	ceph_vinop(inode));
2867	goto out_unlock;
2868	}
2869	}
2870
2871	if ((have & need) == need) {
2872	/*
2873	* Look at (implemented & ~have & not) so that we keep waiting
2874	* on transition from wanted -> needed caps. This is needed
2875	* for WRBUFFER|WR -> WR to avoid a new WR sync write from
2876	* going before a prior buffered writeback happens.
2877	*
2878	* For RDCACHE|RD -> RD, there is not need to wait and we can
2879	* just exclude the revoking caps and force to sync read.
2880	*/
2881	int not = want & ~(have & need);
2882	int revoking = implemented & ~have;
2883	int exclude = revoking & not;
2884	doutc(cl, "%p %llx.%llx have %s but not %s (revoking %s)\n",
2885	inode, ceph_vinop(inode), ceph_cap_string(have),
2886	ceph_cap_string(not), ceph_cap_string(revoking));
2887	if (!exclude || !(exclude & CEPH_CAP_FILE_BUFFER)) {
2888	if (!snap_rwsem_locked &&
2889	!ci->i_head_snapc &&
2890	(need & CEPH_CAP_FILE_WR)) {
2891	if (!down_read_trylock(sem: &mdsc->snap_rwsem)) {
2892	/*
2893	* we can not call down_read() when
2894	* task isn't in TASK_RUNNING state
2895	*/
2896	if (flags & NON_BLOCKING) {
2897	ret = -EAGAIN;
2898	goto out_unlock;
2899	}
2900
2901	spin_unlock(lock: &ci->i_ceph_lock);
2902	down_read(sem: &mdsc->snap_rwsem);
2903	snap_rwsem_locked = true;
2904	goto again;
2905	}
2906	snap_rwsem_locked = true;
2907	}
2908	if ((have & want) == want)
2909	*got = need | (want & ~exclude);
2910	else
2911	*got = need;
2912	ceph_take_cap_refs(ci, got: *got, snap_rwsem_locked: true);
2913	ret = 1;
2914	}
2915	} else {
2916	int session_readonly = false;
2917	int mds_wanted;
2918	if (ci->i_auth_cap &&
2919	(need & (CEPH_CAP_FILE_WR | CEPH_CAP_FILE_EXCL))) {
2920	struct ceph_mds_session *s = ci->i_auth_cap->session;
2921	spin_lock(lock: &s->s_cap_lock);
2922	session_readonly = s->s_readonly;
2923	spin_unlock(lock: &s->s_cap_lock);
2924	}
2925	if (session_readonly) {
2926	doutc(cl, "%p %llx.%llx need %s but mds%d readonly\n",
2927	inode, ceph_vinop(inode), ceph_cap_string(need),
2928	ci->i_auth_cap->mds);
2929	ret = -EROFS;
2930	goto out_unlock;
2931	}
2932
2933	if (ceph_inode_is_shutdown(inode)) {
2934	doutc(cl, "%p %llx.%llx inode is shutdown\n",
2935	inode, ceph_vinop(inode));
2936	ret = -ESTALE;
2937	goto out_unlock;
2938	}
2939	mds_wanted = __ceph_caps_mds_wanted(ci, check: false);
2940	if (need & ~mds_wanted) {
2941	doutc(cl, "%p %llx.%llx need %s > mds_wanted %s\n",
2942	inode, ceph_vinop(inode), ceph_cap_string(need),
2943	ceph_cap_string(mds_wanted));
2944	ret = -EUCLEAN;
2945	goto out_unlock;
2946	}
2947
2948	doutc(cl, "%p %llx.%llx have %s need %s\n", inode,
2949	ceph_vinop(inode), ceph_cap_string(have),
2950	ceph_cap_string(need));
2951	}
2952	out_unlock:
2953
2954	__ceph_touch_fmode(ci, mdsc, fmode: flags);
2955
2956	spin_unlock(lock: &ci->i_ceph_lock);
2957	if (snap_rwsem_locked)
2958	up_read(sem: &mdsc->snap_rwsem);
2959
2960	if (!ret)
2961	ceph_update_cap_mis(m: &mdsc->metric);
2962	else if (ret == 1)
2963	ceph_update_cap_hit(m: &mdsc->metric);
2964
2965	doutc(cl, "%p %llx.%llx ret %d got %s\n", inode,
2966	ceph_vinop(inode), ret, ceph_cap_string(*got));
2967	return ret;
2968	}
2969
2970	/*
2971	* Check the offset we are writing up to against our current
2972	* max_size. If necessary, tell the MDS we want to write to
2973	* a larger offset.
2974	*/
2975	static void check_max_size(struct inode *inode, loff_t endoff)
2976	{
2977	struct ceph_inode_info *ci = ceph_inode(inode);
2978	struct ceph_client *cl = ceph_inode_to_client(inode);
2979	int check = 0;
2980
2981	/* do we need to explicitly request a larger max_size? */
2982	spin_lock(lock: &ci->i_ceph_lock);
2983	if (endoff >= ci->i_max_size && endoff > ci->i_wanted_max_size) {
2984	doutc(cl, "write %p %llx.%llx at large endoff %llu, req max_size\n",
2985	inode, ceph_vinop(inode), endoff);
2986	ci->i_wanted_max_size = endoff;
2987	}
2988	/* duplicate ceph_check_caps()'s logic */
2989	if (ci->i_auth_cap &&
2990	(ci->i_auth_cap->issued & CEPH_CAP_FILE_WR) &&
2991	ci->i_wanted_max_size > ci->i_max_size &&
2992	ci->i_wanted_max_size > ci->i_requested_max_size)
2993	check = 1;
2994	spin_unlock(lock: &ci->i_ceph_lock);
2995	if (check)
2996	ceph_check_caps(ci, CHECK_CAPS_AUTHONLY);
2997	}
2998
2999	static inline int get_used_fmode(int caps)
3000	{
3001	int fmode = 0;
3002	if (caps & CEPH_CAP_FILE_RD)
3003	fmode |= CEPH_FILE_MODE_RD;
3004	if (caps & CEPH_CAP_FILE_WR)
3005	fmode |= CEPH_FILE_MODE_WR;
3006	return fmode;
3007	}
3008
3009	int ceph_try_get_caps(struct inode *inode, int need, int want,
3010	bool nonblock, int *got)
3011	{
3012	int ret, flags;
3013
3014	BUG_ON(need & ~CEPH_CAP_FILE_RD);
3015	BUG_ON(want & ~(CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO |
3016	CEPH_CAP_FILE_SHARED | CEPH_CAP_FILE_EXCL |
3017	CEPH_CAP_ANY_DIR_OPS));
3018	if (need) {
3019	ret = ceph_pool_perm_check(inode, need);
3020	if (ret < 0)
3021	return ret;
3022	}
3023
3024	flags = get_used_fmode(caps: need | want);
3025	if (nonblock)
3026	flags |= NON_BLOCKING;
3027
3028	ret = try_get_cap_refs(inode, need, want, endoff: 0, flags, got);
3029	/* three special error codes */
3030	if (ret == -EAGAIN || ret == -EFBIG || ret == -EUCLEAN)
3031	ret = 0;
3032	return ret;
3033	}
3034
3035	/*
3036	* Wait for caps, and take cap references. If we can't get a WR cap
3037	* due to a small max_size, make sure we check_max_size (and possibly
3038	* ask the mds) so we don't get hung up indefinitely.
3039	*/
3040	int __ceph_get_caps(struct inode *inode, struct ceph_file_info *fi, int need,
3041	int want, loff_t endoff, int *got)
3042	{
3043	struct ceph_inode_info *ci = ceph_inode(inode);
3044	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
3045	int ret, _got, flags;
3046
3047	ret = ceph_pool_perm_check(inode, need);
3048	if (ret < 0)
3049	return ret;
3050
3051	if (fi && (fi->fmode & CEPH_FILE_MODE_WR) &&
3052	fi->filp_gen != READ_ONCE(fsc->filp_gen))
3053	return -EBADF;
3054
3055	flags = get_used_fmode(caps: need | want);
3056
3057	while (true) {
3058	flags &= CEPH_FILE_MODE_MASK;
3059	if (vfs_inode_has_locks(inode))
3060	flags |= CHECK_FILELOCK;
3061	_got = 0;
3062	ret = try_get_cap_refs(inode, need, want, endoff,
3063	flags, got: &_got);
3064	WARN_ON_ONCE(ret == -EAGAIN);
3065	if (!ret) {
3066	#ifdef CONFIG_DEBUG_FS
3067	struct ceph_mds_client *mdsc = fsc->mdsc;
3068	struct cap_wait cw;
3069	#endif
3070	DEFINE_WAIT_FUNC(wait, woken_wake_function);
3071
3072	#ifdef CONFIG_DEBUG_FS
3073	cw.ino = ceph_ino(inode);
3074	cw.tgid = current->tgid;
3075	cw.need = need;
3076	cw.want = want;
3077
3078	spin_lock(lock: &mdsc->caps_list_lock);
3079	list_add(new: &cw.list, head: &mdsc->cap_wait_list);
3080	spin_unlock(lock: &mdsc->caps_list_lock);
3081	#endif
3082
3083	/* make sure used fmode not timeout */
3084	ceph_get_fmode(ci, mode: flags, FMODE_WAIT_BIAS);
3085	add_wait_queue(wq_head: &ci->i_cap_wq, wq_entry: &wait);
3086
3087	flags |= NON_BLOCKING;
3088	while (!(ret = try_get_cap_refs(inode, need, want,
3089	endoff, flags, got: &_got))) {
3090	if (signal_pending(current)) {
3091	ret = -ERESTARTSYS;
3092	break;
3093	}
3094	wait_woken(wq_entry: &wait, TASK_INTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
3095	}
3096
3097	remove_wait_queue(wq_head: &ci->i_cap_wq, wq_entry: &wait);
3098	ceph_put_fmode(ci, mode: flags, FMODE_WAIT_BIAS);
3099
3100	#ifdef CONFIG_DEBUG_FS
3101	spin_lock(lock: &mdsc->caps_list_lock);
3102	list_del(entry: &cw.list);
3103	spin_unlock(lock: &mdsc->caps_list_lock);
3104	#endif
3105
3106	if (ret == -EAGAIN)
3107	continue;
3108	}
3109
3110	if (fi && (fi->fmode & CEPH_FILE_MODE_WR) &&
3111	fi->filp_gen != READ_ONCE(fsc->filp_gen)) {
3112	if (ret >= 0 && _got)
3113	ceph_put_cap_refs(ci, had: _got);
3114	return -EBADF;
3115	}
3116
3117	if (ret < 0) {
3118	if (ret == -EFBIG || ret == -EUCLEAN) {
3119	int ret2 = ceph_wait_on_async_create(inode);
3120	if (ret2 < 0)
3121	return ret2;
3122	}
3123	if (ret == -EFBIG) {
3124	check_max_size(inode, endoff);
3125	continue;
3126	}
3127	if (ret == -EUCLEAN) {
3128	/* session was killed, try renew caps */
3129	ret = ceph_renew_caps(inode, fmode: flags);
3130	if (ret == 0)
3131	continue;
3132	}
3133	return ret;
3134	}
3135
3136	if (S_ISREG(ci->netfs.inode.i_mode) &&
3137	ceph_has_inline_data(ci) &&
3138	(_got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) &&
3139	i_size_read(inode) > 0) {
3140	struct page *page =
3141	find_get_page(mapping: inode->i_mapping, offset: 0);
3142	if (page) {
3143	bool uptodate = PageUptodate(page);
3144
3145	put_page(page);
3146	if (uptodate)
3147	break;
3148	}
3149	/*
3150	* drop cap refs first because getattr while
3151	* holding * caps refs can cause deadlock.
3152	*/
3153	ceph_put_cap_refs(ci, had: _got);
3154	_got = 0;
3155
3156	/*
3157	* getattr request will bring inline data into
3158	* page cache
3159	*/
3160	ret = __ceph_do_getattr(inode, NULL,
3161	CEPH_STAT_CAP_INLINE_DATA,
3162	force: true);
3163	if (ret < 0)
3164	return ret;
3165	continue;
3166	}
3167	break;
3168	}
3169	*got = _got;
3170	return 0;
3171	}
3172
3173	int ceph_get_caps(struct file *filp, int need, int want, loff_t endoff,
3174	int *got)
3175	{
3176	struct ceph_file_info *fi = filp->private_data;
3177	struct inode *inode = file_inode(f: filp);
3178
3179	return __ceph_get_caps(inode, fi, need, want, endoff, got);
3180	}
3181
3182	/*
3183	* Take cap refs. Caller must already know we hold at least one ref
3184	* on the caps in question or we don't know this is safe.
3185	*/
3186	void ceph_get_cap_refs(struct ceph_inode_info *ci, int caps)
3187	{
3188	spin_lock(lock: &ci->i_ceph_lock);
3189	ceph_take_cap_refs(ci, got: caps, snap_rwsem_locked: false);
3190	spin_unlock(lock: &ci->i_ceph_lock);
3191	}
3192
3193
3194	/*
3195	* drop cap_snap that is not associated with any snapshot.
3196	* we don't need to send FLUSHSNAP message for it.
3197	*/
3198	static int ceph_try_drop_cap_snap(struct ceph_inode_info *ci,
3199	struct ceph_cap_snap *capsnap)
3200	{
3201	struct inode *inode = &ci->netfs.inode;
3202	struct ceph_client *cl = ceph_inode_to_client(inode);
3203
3204	if (!capsnap->need_flush &&
3205	!capsnap->writing && !capsnap->dirty_pages) {
3206	doutc(cl, "%p follows %llu\n", capsnap, capsnap->follows);
3207	BUG_ON(capsnap->cap_flush.tid > 0);
3208	ceph_put_snap_context(sc: capsnap->context);
3209	if (!list_is_last(list: &capsnap->ci_item, head: &ci->i_cap_snaps))
3210	ci->i_ceph_flags |= CEPH_I_FLUSH_SNAPS;
3211
3212	list_del(entry: &capsnap->ci_item);
3213	ceph_put_cap_snap(capsnap);
3214	return 1;
3215	}
3216	return 0;
3217	}
3218
3219	enum put_cap_refs_mode {
3220	PUT_CAP_REFS_SYNC = 0,
3221	PUT_CAP_REFS_ASYNC,
3222	};
3223
3224	/*
3225	* Release cap refs.
3226	*
3227	* If we released the last ref on any given cap, call ceph_check_caps
3228	* to release (or schedule a release).
3229	*
3230	* If we are releasing a WR cap (from a sync write), finalize any affected
3231	* cap_snap, and wake up any waiters.
3232	*/
3233	static void __ceph_put_cap_refs(struct ceph_inode_info *ci, int had,
3234	enum put_cap_refs_mode mode)
3235	{
3236	struct inode *inode = &ci->netfs.inode;
3237	struct ceph_client *cl = ceph_inode_to_client(inode);
3238	int last = 0, put = 0, flushsnaps = 0, wake = 0;
3239	bool check_flushsnaps = false;
3240
3241	spin_lock(lock: &ci->i_ceph_lock);
3242	if (had & CEPH_CAP_PIN)
3243	--ci->i_pin_ref;
3244	if (had & CEPH_CAP_FILE_RD)
3245	if (--ci->i_rd_ref == 0)
3246	last++;
3247	if (had & CEPH_CAP_FILE_CACHE)
3248	if (--ci->i_rdcache_ref == 0)
3249	last++;
3250	if (had & CEPH_CAP_FILE_EXCL)
3251	if (--ci->i_fx_ref == 0)
3252	last++;
3253	if (had & CEPH_CAP_FILE_BUFFER) {
3254	if (--ci->i_wb_ref == 0) {
3255	last++;
3256	/* put the ref held by ceph_take_cap_refs() */
3257	put++;
3258	check_flushsnaps = true;
3259	}
3260	doutc(cl, "%p %llx.%llx wb %d -> %d (?)\n", inode,
3261	ceph_vinop(inode), ci->i_wb_ref+1, ci->i_wb_ref);
3262	}
3263	if (had & CEPH_CAP_FILE_WR) {
3264	if (--ci->i_wr_ref == 0) {
3265	/*
3266	* The Fb caps will always be took and released
3267	* together with the Fw caps.
3268	*/
3269	WARN_ON_ONCE(ci->i_wb_ref);
3270
3271	last++;
3272	check_flushsnaps = true;
3273	if (ci->i_wrbuffer_ref_head == 0 &&
3274	ci->i_dirty_caps == 0 &&
3275	ci->i_flushing_caps == 0) {
3276	BUG_ON(!ci->i_head_snapc);
3277	ceph_put_snap_context(sc: ci->i_head_snapc);
3278	ci->i_head_snapc = NULL;
3279	}
3280	/* see comment in __ceph_remove_cap() */
3281	if (!__ceph_is_any_real_caps(ci) && ci->i_snap_realm)
3282	ceph_change_snap_realm(inode, NULL);
3283	}
3284	}
3285	if (check_flushsnaps && __ceph_have_pending_cap_snap(ci)) {
3286	struct ceph_cap_snap *capsnap =
3287	list_last_entry(&ci->i_cap_snaps,
3288	struct ceph_cap_snap,
3289	ci_item);
3290
3291	capsnap->writing = 0;
3292	if (ceph_try_drop_cap_snap(ci, capsnap))
3293	/* put the ref held by ceph_queue_cap_snap() */
3294	put++;
3295	else if (__ceph_finish_cap_snap(ci, capsnap))
3296	flushsnaps = 1;
3297	wake = 1;
3298	}
3299	spin_unlock(lock: &ci->i_ceph_lock);
3300
3301	doutc(cl, "%p %llx.%llx had %s%s%s\n", inode, ceph_vinop(inode),
3302	ceph_cap_string(had), last ? " last" : "", put ? " put" : "");
3303
3304	switch (mode) {
3305	case PUT_CAP_REFS_SYNC:
3306	if (last)
3307	ceph_check_caps(ci, flags: 0);
3308	else if (flushsnaps)
3309	ceph_flush_snaps(ci, NULL);
3310	break;
3311	case PUT_CAP_REFS_ASYNC:
3312	if (last)
3313	ceph_queue_check_caps(inode);
3314	else if (flushsnaps)
3315	ceph_queue_flush_snaps(inode);
3316	break;
3317	default:
3318	break;
3319	}
3320	if (wake)
3321	wake_up_all(&ci->i_cap_wq);
3322	while (put-- > 0)
3323	iput(inode);
3324	}
3325
3326	void ceph_put_cap_refs(struct ceph_inode_info *ci, int had)
3327	{
3328	__ceph_put_cap_refs(ci, had, mode: PUT_CAP_REFS_SYNC);
3329	}
3330
3331	void ceph_put_cap_refs_async(struct ceph_inode_info *ci, int had)
3332	{
3333	__ceph_put_cap_refs(ci, had, mode: PUT_CAP_REFS_ASYNC);
3334	}
3335
3336	/*
3337	* Release @nr WRBUFFER refs on dirty pages for the given @snapc snap
3338	* context. Adjust per-snap dirty page accounting as appropriate.
3339	* Once all dirty data for a cap_snap is flushed, flush snapped file
3340	* metadata back to the MDS. If we dropped the last ref, call
3341	* ceph_check_caps.
3342	*/
3343	void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
3344	struct ceph_snap_context *snapc)
3345	{
3346	struct inode *inode = &ci->netfs.inode;
3347	struct ceph_client *cl = ceph_inode_to_client(inode);
3348	struct ceph_cap_snap *capsnap = NULL, *iter;
3349	int put = 0;
3350	bool last = false;
3351	bool flush_snaps = false;
3352	bool complete_capsnap = false;
3353
3354	spin_lock(lock: &ci->i_ceph_lock);
3355	ci->i_wrbuffer_ref -= nr;
3356	if (ci->i_wrbuffer_ref == 0) {
3357	last = true;
3358	put++;
3359	}
3360
3361	if (ci->i_head_snapc == snapc) {
3362	ci->i_wrbuffer_ref_head -= nr;
3363	if (ci->i_wrbuffer_ref_head == 0 &&
3364	ci->i_wr_ref == 0 &&
3365	ci->i_dirty_caps == 0 &&
3366	ci->i_flushing_caps == 0) {
3367	BUG_ON(!ci->i_head_snapc);
3368	ceph_put_snap_context(sc: ci->i_head_snapc);
3369	ci->i_head_snapc = NULL;
3370	}
3371	doutc(cl, "on %p %llx.%llx head %d/%d -> %d/%d %s\n",
3372	inode, ceph_vinop(inode), ci->i_wrbuffer_ref+nr,
3373	ci->i_wrbuffer_ref_head+nr, ci->i_wrbuffer_ref,
3374	ci->i_wrbuffer_ref_head, last ? " LAST" : "");
3375	} else {
3376	list_for_each_entry(iter, &ci->i_cap_snaps, ci_item) {
3377	if (iter->context == snapc) {
3378	capsnap = iter;
3379	break;
3380	}
3381	}
3382
3383	if (!capsnap) {
3384	/*
3385	* The capsnap should already be removed when removing
3386	* auth cap in the case of a forced unmount.
3387	*/
3388	WARN_ON_ONCE(ci->i_auth_cap);
3389	goto unlock;
3390	}
3391
3392	capsnap->dirty_pages -= nr;
3393	if (capsnap->dirty_pages == 0) {
3394	complete_capsnap = true;
3395	if (!capsnap->writing) {
3396	if (ceph_try_drop_cap_snap(ci, capsnap)) {
3397	put++;
3398	} else {
3399	ci->i_ceph_flags |= CEPH_I_FLUSH_SNAPS;
3400	flush_snaps = true;
3401	}
3402	}
3403	}
3404	doutc(cl, "%p %llx.%llx cap_snap %p snap %lld %d/%d -> %d/%d %s%s\n",
3405	inode, ceph_vinop(inode), capsnap, capsnap->context->seq,
3406	ci->i_wrbuffer_ref+nr, capsnap->dirty_pages + nr,
3407	ci->i_wrbuffer_ref, capsnap->dirty_pages,
3408	last ? " (wrbuffer last)" : "",
3409	complete_capsnap ? " (complete capsnap)" : "");
3410	}
3411
3412	unlock:
3413	spin_unlock(lock: &ci->i_ceph_lock);
3414
3415	if (last) {
3416	ceph_check_caps(ci, flags: 0);
3417	} else if (flush_snaps) {
3418	ceph_flush_snaps(ci, NULL);
3419	}
3420	if (complete_capsnap)
3421	wake_up_all(&ci->i_cap_wq);
3422	while (put-- > 0) {
3423	iput(inode);
3424	}
3425	}
3426
3427	/*
3428	* Invalidate unlinked inode's aliases, so we can drop the inode ASAP.
3429	*/
3430	static void invalidate_aliases(struct inode *inode)
3431	{
3432	struct ceph_client *cl = ceph_inode_to_client(inode);
3433	struct dentry *dn, *prev = NULL;
3434
3435	doutc(cl, "%p %llx.%llx\n", inode, ceph_vinop(inode));
3436	d_prune_aliases(inode);
3437	/*
3438	* For non-directory inode, d_find_alias() only returns
3439	* hashed dentry. After calling d_invalidate(), the
3440	* dentry becomes unhashed.
3441	*
3442	* For directory inode, d_find_alias() can return
3443	* unhashed dentry. But directory inode should have
3444	* one alias at most.
3445	*/
3446	while ((dn = d_find_alias(inode))) {
3447	if (dn == prev) {
3448	dput(dn);
3449	break;
3450	}
3451	d_invalidate(dn);
3452	if (prev)
3453	dput(prev);
3454	prev = dn;
3455	}
3456	if (prev)
3457	dput(prev);
3458	}
3459
3460	struct cap_extra_info {
3461	struct ceph_string *pool_ns;
3462	/* inline data */
3463	u64 inline_version;
3464	void *inline_data;
3465	u32 inline_len;
3466	/* dirstat */
3467	bool dirstat_valid;
3468	u64 nfiles;
3469	u64 nsubdirs;
3470	u64 change_attr;
3471	/* currently issued */
3472	int issued;
3473	struct timespec64 btime;
3474	u8 *fscrypt_auth;
3475	u32 fscrypt_auth_len;
3476	u64 fscrypt_file_size;
3477	};
3478
3479	/*
3480	* Handle a cap GRANT message from the MDS. (Note that a GRANT may
3481	* actually be a revocation if it specifies a smaller cap set.)
3482	*
3483	* caller holds s_mutex and i_ceph_lock, we drop both.
3484	*/
3485	static void handle_cap_grant(struct inode *inode,
3486	struct ceph_mds_session *session,
3487	struct ceph_cap *cap,
3488	struct ceph_mds_caps *grant,
3489	struct ceph_buffer *xattr_buf,
3490	struct cap_extra_info *extra_info)
3491	__releases(ci->i_ceph_lock)
3492	__releases(session->s_mdsc->snap_rwsem)
3493	{
3494	struct ceph_client *cl = ceph_inode_to_client(inode);
3495	struct ceph_inode_info *ci = ceph_inode(inode);
3496	int seq = le32_to_cpu(grant->seq);
3497	int newcaps = le32_to_cpu(grant->caps);
3498	int used, wanted, dirty;
3499	u64 size = le64_to_cpu(grant->size);
3500	u64 max_size = le64_to_cpu(grant->max_size);
3501	unsigned char check_caps = 0;
3502	bool was_stale = cap->cap_gen < atomic_read(v: &session->s_cap_gen);
3503	bool wake = false;
3504	bool writeback = false;
3505	bool queue_trunc = false;
3506	bool queue_invalidate = false;
3507	bool deleted_inode = false;
3508	bool fill_inline = false;
3509	bool revoke_wait = false;
3510	int flags = 0;
3511
3512	/*
3513	* If there is at least one crypto block then we'll trust
3514	* fscrypt_file_size. If the real length of the file is 0, then
3515	* ignore it (it has probably been truncated down to 0 by the MDS).
3516	*/
3517	if (IS_ENCRYPTED(inode) && size)
3518	size = extra_info->fscrypt_file_size;
3519
3520	doutc(cl, "%p %llx.%llx cap %p mds%d seq %d %s\n", inode,
3521	ceph_vinop(inode), cap, session->s_mds, seq,
3522	ceph_cap_string(newcaps));
3523	doutc(cl, " size %llu max_size %llu, i_size %llu\n", size,
3524	max_size, i_size_read(inode));
3525
3526
3527	/*
3528	* If CACHE is being revoked, and we have no dirty buffers,
3529	* try to invalidate (once). (If there are dirty buffers, we
3530	* will invalidate _after_ writeback.)
3531	*/
3532	if (S_ISREG(inode->i_mode) && /* don't invalidate readdir cache */
3533	((cap->issued & ~newcaps) & CEPH_CAP_FILE_CACHE) &&
3534	(newcaps & CEPH_CAP_FILE_LAZYIO) == 0 &&
3535	!(ci->i_wrbuffer_ref || ci->i_wb_ref)) {
3536	if (try_nonblocking_invalidate(inode)) {
3537	/* there were locked pages.. invalidate later
3538	in a separate thread. */
3539	if (ci->i_rdcache_revoking != ci->i_rdcache_gen) {
3540	queue_invalidate = true;
3541	ci->i_rdcache_revoking = ci->i_rdcache_gen;
3542	}
3543	}
3544	}
3545
3546	if (was_stale)
3547	cap->issued = cap->implemented = CEPH_CAP_PIN;
3548
3549	/*
3550	* auth mds of the inode changed. we received the cap export message,
3551	* but still haven't received the cap import message. handle_cap_export
3552	* updated the new auth MDS' cap.
3553	*
3554	* "ceph_seq_cmp(seq, cap->seq) <= 0" means we are processing a message
3555	* that was sent before the cap import message. So don't remove caps.
3556	*/
3557	if (ceph_seq_cmp(a: seq, b: cap->seq) <= 0) {
3558	WARN_ON(cap != ci->i_auth_cap);
3559	WARN_ON(cap->cap_id != le64_to_cpu(grant->cap_id));
3560	seq = cap->seq;
3561	newcaps |= cap->issued;
3562	}
3563
3564	/* side effects now are allowed */
3565	cap->cap_gen = atomic_read(v: &session->s_cap_gen);
3566	cap->seq = seq;
3567
3568	__check_cap_issue(ci, cap, issued: newcaps);
3569
3570	inode_set_max_iversion_raw(inode, val: extra_info->change_attr);
3571
3572	if ((newcaps & CEPH_CAP_AUTH_SHARED) &&
3573	(extra_info->issued & CEPH_CAP_AUTH_EXCL) == 0) {
3574	umode_t mode = le32_to_cpu(grant->mode);
3575
3576	if (inode_wrong_type(inode, mode))
3577	pr_warn_once("inode type changed! (ino %llx.%llx is 0%o, mds says 0%o)\n",
3578	ceph_vinop(inode), inode->i_mode, mode);
3579	else
3580	inode->i_mode = mode;
3581	inode->i_uid = make_kuid(from: &init_user_ns, le32_to_cpu(grant->uid));
3582	inode->i_gid = make_kgid(from: &init_user_ns, le32_to_cpu(grant->gid));
3583	ci->i_btime = extra_info->btime;
3584	doutc(cl, "%p %llx.%llx mode 0%o uid.gid %d.%d\n", inode,
3585	ceph_vinop(inode), inode->i_mode,
3586	from_kuid(&init_user_ns, inode->i_uid),
3587	from_kgid(&init_user_ns, inode->i_gid));
3588	#if IS_ENABLED(CONFIG_FS_ENCRYPTION)
3589	if (ci->fscrypt_auth_len != extra_info->fscrypt_auth_len ||
3590	memcmp(p: ci->fscrypt_auth, q: extra_info->fscrypt_auth,
3591	size: ci->fscrypt_auth_len))
3592	pr_warn_ratelimited_client(cl,
3593	"cap grant attempt to change fscrypt_auth on non-I_NEW inode (old len %d new len %d)\n",
3594	ci->fscrypt_auth_len,
3595	extra_info->fscrypt_auth_len);
3596	#endif
3597	}
3598
3599	if ((newcaps & CEPH_CAP_LINK_SHARED) &&
3600	(extra_info->issued & CEPH_CAP_LINK_EXCL) == 0) {
3601	set_nlink(inode, le32_to_cpu(grant->nlink));
3602	if (inode->i_nlink == 0)
3603	deleted_inode = true;
3604	}
3605
3606	if ((extra_info->issued & CEPH_CAP_XATTR_EXCL) == 0 &&
3607	grant->xattr_len) {
3608	int len = le32_to_cpu(grant->xattr_len);
3609	u64 version = le64_to_cpu(grant->xattr_version);
3610
3611	if (version > ci->i_xattrs.version) {
3612	doutc(cl, " got new xattrs v%llu on %p %llx.%llx len %d\n",
3613	version, inode, ceph_vinop(inode), len);
3614	if (ci->i_xattrs.blob)
3615	ceph_buffer_put(b: ci->i_xattrs.blob);
3616	ci->i_xattrs.blob = ceph_buffer_get(b: xattr_buf);
3617	ci->i_xattrs.version = version;
3618	ceph_forget_all_cached_acls(inode);
3619	ceph_security_invalidate_secctx(inode);
3620	}
3621	}
3622
3623	if (newcaps & CEPH_CAP_ANY_RD) {
3624	struct timespec64 mtime, atime, ctime;
3625	/* ctime/mtime/atime? */
3626	ceph_decode_timespec64(ts: &mtime, tv: &grant->mtime);
3627	ceph_decode_timespec64(ts: &atime, tv: &grant->atime);
3628	ceph_decode_timespec64(ts: &ctime, tv: &grant->ctime);
3629	ceph_fill_file_time(inode, issued: extra_info->issued,
3630	le32_to_cpu(grant->time_warp_seq),
3631	ctime: &ctime, mtime: &mtime, atime: &atime);
3632	}
3633
3634	if ((newcaps & CEPH_CAP_FILE_SHARED) && extra_info->dirstat_valid) {
3635	ci->i_files = extra_info->nfiles;
3636	ci->i_subdirs = extra_info->nsubdirs;
3637	}
3638
3639	if (newcaps & (CEPH_CAP_ANY_FILE_RD | CEPH_CAP_ANY_FILE_WR)) {
3640	/* file layout may have changed */
3641	s64 old_pool = ci->i_layout.pool_id;
3642	struct ceph_string *old_ns;
3643
3644	ceph_file_layout_from_legacy(fl: &ci->i_layout, legacy: &grant->layout);
3645	old_ns = rcu_dereference_protected(ci->i_layout.pool_ns,
3646	lockdep_is_held(&ci->i_ceph_lock));
3647	rcu_assign_pointer(ci->i_layout.pool_ns, extra_info->pool_ns);
3648
3649	if (ci->i_layout.pool_id != old_pool ||
3650	extra_info->pool_ns != old_ns)
3651	ci->i_ceph_flags &= ~CEPH_I_POOL_PERM;
3652
3653	extra_info->pool_ns = old_ns;
3654
3655	/* size/truncate_seq? */
3656	queue_trunc = ceph_fill_file_size(inode, issued: extra_info->issued,
3657	le32_to_cpu(grant->truncate_seq),
3658	le64_to_cpu(grant->truncate_size),
3659	size);
3660	}
3661
3662	if (ci->i_auth_cap == cap && (newcaps & CEPH_CAP_ANY_FILE_WR)) {
3663	if (max_size != ci->i_max_size) {
3664	doutc(cl, "max_size %lld -> %llu\n", ci->i_max_size,
3665	max_size);
3666	ci->i_max_size = max_size;
3667	if (max_size >= ci->i_wanted_max_size) {
3668	ci->i_wanted_max_size = 0; /* reset */
3669	ci->i_requested_max_size = 0;
3670	}
3671	wake = true;
3672	}
3673	}
3674
3675	/* check cap bits */
3676	wanted = __ceph_caps_wanted(ci);
3677	used = __ceph_caps_used(ci);
3678	dirty = __ceph_caps_dirty(ci);
3679	doutc(cl, " my wanted = %s, used = %s, dirty %s\n",
3680	ceph_cap_string(wanted), ceph_cap_string(used),
3681	ceph_cap_string(dirty));
3682
3683	if ((was_stale || le32_to_cpu(grant->op) == CEPH_CAP_OP_IMPORT) &&
3684	(wanted & ~(cap->mds_wanted | newcaps))) {
3685	/*
3686	* If mds is importing cap, prior cap messages that update
3687	* 'wanted' may get dropped by mds (migrate seq mismatch).
3688	*
3689	* We don't send cap message to update 'wanted' if what we
3690	* want are already issued. If mds revokes caps, cap message
3691	* that releases caps also tells mds what we want. But if
3692	* caps got revoked by mds forcedly (session stale). We may
3693	* haven't told mds what we want.
3694	*/
3695	check_caps = 1;
3696	}
3697
3698	/* revocation, grant, or no-op? */
3699	if (cap->issued & ~newcaps) {
3700	int revoking = cap->issued & ~newcaps;
3701
3702	doutc(cl, "revocation: %s -> %s (revoking %s)\n",
3703	ceph_cap_string(cap->issued), ceph_cap_string(newcaps),
3704	ceph_cap_string(revoking));
3705	if (S_ISREG(inode->i_mode) &&
3706	(revoking & used & CEPH_CAP_FILE_BUFFER)) {
3707	writeback = true; /* initiate writeback; will delay ack */
3708	revoke_wait = true;
3709	} else if (queue_invalidate &&
3710	revoking == CEPH_CAP_FILE_CACHE &&
3711	(newcaps & CEPH_CAP_FILE_LAZYIO) == 0) {
3712	revoke_wait = true; /* do nothing yet, invalidation will be queued */
3713	} else if (cap == ci->i_auth_cap) {
3714	check_caps = 1; /* check auth cap only */
3715	} else {
3716	check_caps = 2; /* check all caps */
3717	}
3718	/* If there is new caps, try to wake up the waiters */
3719	if (~cap->issued & newcaps)
3720	wake = true;
3721	cap->issued = newcaps;
3722	cap->implemented |= newcaps;
3723	} else if (cap->issued == newcaps) {
3724	doutc(cl, "caps unchanged: %s -> %s\n",
3725	ceph_cap_string(cap->issued),
3726	ceph_cap_string(newcaps));
3727	} else {
3728	doutc(cl, "grant: %s -> %s\n", ceph_cap_string(cap->issued),
3729	ceph_cap_string(newcaps));
3730	/* non-auth MDS is revoking the newly grant caps ? */
3731	if (cap == ci->i_auth_cap &&
3732	__ceph_caps_revoking_other(ci, ocap: cap, mask: newcaps))
3733	check_caps = 2;
3734
3735	cap->issued = newcaps;
3736	cap->implemented |= newcaps; /* add bits only, to
3737	* avoid stepping on a
3738	* pending revocation */
3739	wake = true;
3740	}
3741	BUG_ON(cap->issued & ~cap->implemented);
3742
3743	/* don't let check_caps skip sending a response to MDS for revoke msgs */
3744	if (!revoke_wait && le32_to_cpu(grant->op) == CEPH_CAP_OP_REVOKE) {
3745	cap->mds_wanted = 0;
3746	flags |= CHECK_CAPS_FLUSH_FORCE;
3747	if (cap == ci->i_auth_cap)
3748	check_caps = 1; /* check auth cap only */
3749	else
3750	check_caps = 2; /* check all caps */
3751	}
3752
3753	if (extra_info->inline_version > 0 &&
3754	extra_info->inline_version >= ci->i_inline_version) {
3755	ci->i_inline_version = extra_info->inline_version;
3756	if (ci->i_inline_version != CEPH_INLINE_NONE &&
3757	(newcaps & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)))
3758	fill_inline = true;
3759	}
3760
3761	if (le32_to_cpu(grant->op) == CEPH_CAP_OP_IMPORT) {
3762	if (ci->i_auth_cap == cap) {
3763	if (newcaps & ~extra_info->issued)
3764	wake = true;
3765
3766	if (ci->i_requested_max_size > max_size ||
3767	!(le32_to_cpu(grant->wanted) & CEPH_CAP_ANY_FILE_WR)) {
3768	/* re-request max_size if necessary */
3769	ci->i_requested_max_size = 0;
3770	wake = true;
3771	}
3772
3773	ceph_kick_flushing_inode_caps(session, ci);
3774	}
3775	up_read(sem: &session->s_mdsc->snap_rwsem);
3776	}
3777	spin_unlock(lock: &ci->i_ceph_lock);
3778
3779	if (fill_inline)
3780	ceph_fill_inline_data(inode, NULL, data: extra_info->inline_data,
3781	len: extra_info->inline_len);
3782
3783	if (queue_trunc)
3784	ceph_queue_vmtruncate(inode);
3785
3786	if (writeback)
3787	/*
3788	* queue inode for writeback: we can't actually call
3789	* filemap_write_and_wait, etc. from message handler
3790	* context.
3791	*/
3792	ceph_queue_writeback(inode);
3793	if (queue_invalidate)
3794	ceph_queue_invalidate(inode);
3795	if (deleted_inode)
3796	invalidate_aliases(inode);
3797	if (wake)
3798	wake_up_all(&ci->i_cap_wq);
3799
3800	mutex_unlock(lock: &session->s_mutex);
3801	if (check_caps == 1)
3802	ceph_check_caps(ci, flags: flags | CHECK_CAPS_AUTHONLY | CHECK_CAPS_NOINVAL);
3803	else if (check_caps == 2)
3804	ceph_check_caps(ci, flags: flags | CHECK_CAPS_NOINVAL);
3805	}
3806
3807	/*
3808	* Handle FLUSH_ACK from MDS, indicating that metadata we sent to the
3809	* MDS has been safely committed.
3810	*/
3811	static void handle_cap_flush_ack(struct inode *inode, u64 flush_tid,
3812	struct ceph_mds_caps *m,
3813	struct ceph_mds_session *session,
3814	struct ceph_cap *cap)
3815	__releases(ci->i_ceph_lock)
3816	{
3817	struct ceph_inode_info *ci = ceph_inode(inode);
3818	struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(sb: inode->i_sb)->mdsc;
3819	struct ceph_client *cl = mdsc->fsc->client;
3820	struct ceph_cap_flush *cf, *tmp_cf;
3821	LIST_HEAD(to_remove);
3822	unsigned seq = le32_to_cpu(m->seq);
3823	int dirty = le32_to_cpu(m->dirty);
3824	int cleaned = 0;
3825	bool drop = false;
3826	bool wake_ci = false;
3827	bool wake_mdsc = false;
3828
3829	list_for_each_entry_safe(cf, tmp_cf, &ci->i_cap_flush_list, i_list) {
3830	/* Is this the one that was flushed? */
3831	if (cf->tid == flush_tid)
3832	cleaned = cf->caps;
3833
3834	/* Is this a capsnap? */
3835	if (cf->is_capsnap)
3836	continue;
3837
3838	if (cf->tid <= flush_tid) {
3839	/*
3840	* An earlier or current tid. The FLUSH_ACK should
3841	* represent a superset of this flush's caps.
3842	*/
3843	wake_ci |= __detach_cap_flush_from_ci(ci, cf);
3844	list_add_tail(new: &cf->i_list, head: &to_remove);
3845	} else {
3846	/*
3847	* This is a later one. Any caps in it are still dirty
3848	* so don't count them as cleaned.
3849	*/
3850	cleaned &= ~cf->caps;
3851	if (!cleaned)
3852	break;
3853	}
3854	}
3855
3856	doutc(cl, "%p %llx.%llx mds%d seq %d on %s cleaned %s, flushing %s -> %s\n",
3857	inode, ceph_vinop(inode), session->s_mds, seq,
3858	ceph_cap_string(dirty), ceph_cap_string(cleaned),
3859	ceph_cap_string(ci->i_flushing_caps),
3860	ceph_cap_string(ci->i_flushing_caps & ~cleaned));
3861
3862	if (list_empty(head: &to_remove) && !cleaned)
3863	goto out;
3864
3865	ci->i_flushing_caps &= ~cleaned;
3866
3867	spin_lock(lock: &mdsc->cap_dirty_lock);
3868
3869	list_for_each_entry(cf, &to_remove, i_list)
3870	wake_mdsc |= __detach_cap_flush_from_mdsc(mdsc, cf);
3871
3872	if (ci->i_flushing_caps == 0) {
3873	if (list_empty(head: &ci->i_cap_flush_list)) {
3874	list_del_init(entry: &ci->i_flushing_item);
3875	if (!list_empty(head: &session->s_cap_flushing)) {
3876	struct inode *inode =
3877	&list_first_entry(&session->s_cap_flushing,
3878	struct ceph_inode_info,
3879	i_flushing_item)->netfs.inode;
3880	doutc(cl, " mds%d still flushing cap on %p %llx.%llx\n",
3881	session->s_mds, inode, ceph_vinop(inode));
3882	}
3883	}
3884	mdsc->num_cap_flushing--;
3885	doutc(cl, " %p %llx.%llx now !flushing\n", inode,
3886	ceph_vinop(inode));
3887
3888	if (ci->i_dirty_caps == 0) {
3889	doutc(cl, " %p %llx.%llx now clean\n", inode,
3890	ceph_vinop(inode));
3891	BUG_ON(!list_empty(&ci->i_dirty_item));
3892	drop = true;
3893	if (ci->i_wr_ref == 0 &&
3894	ci->i_wrbuffer_ref_head == 0) {
3895	BUG_ON(!ci->i_head_snapc);
3896	ceph_put_snap_context(sc: ci->i_head_snapc);
3897	ci->i_head_snapc = NULL;
3898	}
3899	} else {
3900	BUG_ON(list_empty(&ci->i_dirty_item));
3901	}
3902	}
3903	spin_unlock(lock: &mdsc->cap_dirty_lock);
3904
3905	out:
3906	spin_unlock(lock: &ci->i_ceph_lock);
3907
3908	while (!list_empty(head: &to_remove)) {
3909	cf = list_first_entry(&to_remove,
3910	struct ceph_cap_flush, i_list);
3911	list_del_init(entry: &cf->i_list);
3912	if (!cf->is_capsnap)
3913	ceph_free_cap_flush(cf);
3914	}
3915
3916	if (wake_ci)
3917	wake_up_all(&ci->i_cap_wq);
3918	if (wake_mdsc)
3919	wake_up_all(&mdsc->cap_flushing_wq);
3920	if (drop)
3921	iput(inode);
3922	}
3923
3924	void __ceph_remove_capsnap(struct inode *inode, struct ceph_cap_snap *capsnap,
3925	bool *wake_ci, bool *wake_mdsc)
3926	{
3927	struct ceph_inode_info *ci = ceph_inode(inode);
3928	struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(sb: inode->i_sb)->mdsc;
3929	struct ceph_client *cl = mdsc->fsc->client;
3930	bool ret;
3931
3932	lockdep_assert_held(&ci->i_ceph_lock);
3933
3934	doutc(cl, "removing capsnap %p, %p %llx.%llx ci %p\n", capsnap,
3935	inode, ceph_vinop(inode), ci);
3936
3937	list_del_init(entry: &capsnap->ci_item);
3938	ret = __detach_cap_flush_from_ci(ci, cf: &capsnap->cap_flush);
3939	if (wake_ci)
3940	*wake_ci = ret;
3941
3942	spin_lock(lock: &mdsc->cap_dirty_lock);
3943	if (list_empty(head: &ci->i_cap_flush_list))
3944	list_del_init(entry: &ci->i_flushing_item);
3945
3946	ret = __detach_cap_flush_from_mdsc(mdsc, cf: &capsnap->cap_flush);
3947	if (wake_mdsc)
3948	*wake_mdsc = ret;
3949	spin_unlock(lock: &mdsc->cap_dirty_lock);
3950	}
3951
3952	void ceph_remove_capsnap(struct inode *inode, struct ceph_cap_snap *capsnap,
3953	bool *wake_ci, bool *wake_mdsc)
3954	{
3955	struct ceph_inode_info *ci = ceph_inode(inode);
3956
3957	lockdep_assert_held(&ci->i_ceph_lock);
3958
3959	WARN_ON_ONCE(capsnap->dirty_pages || capsnap->writing);
3960	__ceph_remove_capsnap(inode, capsnap, wake_ci, wake_mdsc);
3961	}
3962
3963	/*
3964	* Handle FLUSHSNAP_ACK. MDS has flushed snap data to disk and we can
3965	* throw away our cap_snap.
3966	*
3967	* Caller hold s_mutex.
3968	*/
3969	static void handle_cap_flushsnap_ack(struct inode *inode, u64 flush_tid,
3970	struct ceph_mds_caps *m,
3971	struct ceph_mds_session *session)
3972	{
3973	struct ceph_inode_info *ci = ceph_inode(inode);
3974	struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(sb: inode->i_sb)->mdsc;
3975	struct ceph_client *cl = mdsc->fsc->client;
3976	u64 follows = le64_to_cpu(m->snap_follows);
3977	struct ceph_cap_snap *capsnap = NULL, *iter;
3978	bool wake_ci = false;
3979	bool wake_mdsc = false;
3980
3981	doutc(cl, "%p %llx.%llx ci %p mds%d follows %lld\n", inode,
3982	ceph_vinop(inode), ci, session->s_mds, follows);
3983
3984	spin_lock(lock: &ci->i_ceph_lock);
3985	list_for_each_entry(iter, &ci->i_cap_snaps, ci_item) {
3986	if (iter->follows == follows) {
3987	if (iter->cap_flush.tid != flush_tid) {
3988	doutc(cl, " cap_snap %p follows %lld "
3989	"tid %lld != %lld\n", iter,
3990	follows, flush_tid,
3991	iter->cap_flush.tid);
3992	break;
3993	}
3994	capsnap = iter;
3995	break;
3996	} else {
3997	doutc(cl, " skipping cap_snap %p follows %lld\n",
3998	iter, iter->follows);
3999	}
4000	}
4001	if (capsnap)
4002	ceph_remove_capsnap(inode, capsnap, wake_ci: &wake_ci, wake_mdsc: &wake_mdsc);
4003	spin_unlock(lock: &ci->i_ceph_lock);
4004
4005	if (capsnap) {
4006	ceph_put_snap_context(sc: capsnap->context);
4007	ceph_put_cap_snap(capsnap);
4008	if (wake_ci)
4009	wake_up_all(&ci->i_cap_wq);
4010	if (wake_mdsc)
4011	wake_up_all(&mdsc->cap_flushing_wq);
4012	iput(inode);
4013	}
4014	}
4015
4016	/*
4017	* Handle TRUNC from MDS, indicating file truncation.
4018	*
4019	* caller hold s_mutex.
4020	*/
4021	static bool handle_cap_trunc(struct inode *inode,
4022	struct ceph_mds_caps *trunc,
4023	struct ceph_mds_session *session,
4024	struct cap_extra_info *extra_info)
4025	{
4026	struct ceph_inode_info *ci = ceph_inode(inode);
4027	struct ceph_client *cl = ceph_inode_to_client(inode);
4028	int mds = session->s_mds;
4029	int seq = le32_to_cpu(trunc->seq);
4030	u32 truncate_seq = le32_to_cpu(trunc->truncate_seq);
4031	u64 truncate_size = le64_to_cpu(trunc->truncate_size);
4032	u64 size = le64_to_cpu(trunc->size);
4033	int implemented = 0;
4034	int dirty = __ceph_caps_dirty(ci);
4035	int issued = __ceph_caps_issued(ci: ceph_inode(inode), implemented: &implemented);
4036	bool queue_trunc = false;
4037
4038	lockdep_assert_held(&ci->i_ceph_lock);
4039
4040	issued |= implemented | dirty;
4041
4042	/*
4043	* If there is at least one crypto block then we'll trust
4044	* fscrypt_file_size. If the real length of the file is 0, then
4045	* ignore it (it has probably been truncated down to 0 by the MDS).
4046	*/
4047	if (IS_ENCRYPTED(inode) && size)
4048	size = extra_info->fscrypt_file_size;
4049
4050	doutc(cl, "%p %llx.%llx mds%d seq %d to %lld truncate seq %d\n",
4051	inode, ceph_vinop(inode), mds, seq, truncate_size, truncate_seq);
4052	queue_trunc = ceph_fill_file_size(inode, issued,
4053	truncate_seq, truncate_size, size);
4054	return queue_trunc;
4055	}
4056
4057	/*
4058	* Handle EXPORT from MDS. Cap is being migrated _from_ this mds to a
4059	* different one. If we are the most recent migration we've seen (as
4060	* indicated by mseq), make note of the migrating cap bits for the
4061	* duration (until we see the corresponding IMPORT).
4062	*
4063	* caller holds s_mutex
4064	*/
4065	static void handle_cap_export(struct inode *inode, struct ceph_mds_caps *ex,
4066	struct ceph_mds_cap_peer *ph,
4067	struct ceph_mds_session *session)
4068	{
4069	struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
4070	struct ceph_client *cl = mdsc->fsc->client;
4071	struct ceph_mds_session *tsession = NULL;
4072	struct ceph_cap *cap, *tcap, *new_cap = NULL;
4073	struct ceph_inode_info *ci = ceph_inode(inode);
4074	u64 t_cap_id;
4075	u32 t_issue_seq, t_mseq;
4076	int target, issued;
4077	int mds = session->s_mds;
4078
4079	if (ph) {
4080	t_cap_id = le64_to_cpu(ph->cap_id);
4081	t_issue_seq = le32_to_cpu(ph->issue_seq);
4082	t_mseq = le32_to_cpu(ph->mseq);
4083	target = le32_to_cpu(ph->mds);
4084	} else {
4085	t_cap_id = t_issue_seq = t_mseq = 0;
4086	target = -1;
4087	}
4088
4089	doutc(cl, " cap %llx.%llx export to peer %d piseq %u pmseq %u\n",
4090	ceph_vinop(inode), target, t_issue_seq, t_mseq);
4091	retry:
4092	down_read(sem: &mdsc->snap_rwsem);
4093	spin_lock(lock: &ci->i_ceph_lock);
4094	cap = __get_cap_for_mds(ci, mds);
4095	if (!cap || cap->cap_id != le64_to_cpu(ex->cap_id))
4096	goto out_unlock;
4097
4098	if (target < 0) {
4099	ceph_remove_cap(mdsc, cap, queue_release: false);
4100	goto out_unlock;
4101	}
4102
4103	/*
4104	* now we know we haven't received the cap import message yet
4105	* because the exported cap still exist.
4106	*/
4107
4108	issued = cap->issued;
4109	if (issued != cap->implemented)
4110	pr_err_ratelimited_client(cl, "issued != implemented: "
4111	"%p %llx.%llx mds%d seq %d mseq %d"
4112	" issued %s implemented %s\n",
4113	inode, ceph_vinop(inode), mds,
4114	cap->seq, cap->mseq,
4115	ceph_cap_string(issued),
4116	ceph_cap_string(cap->implemented));
4117
4118
4119	tcap = __get_cap_for_mds(ci, mds: target);
4120	if (tcap) {
4121	/* already have caps from the target */
4122	if (tcap->cap_id == t_cap_id &&
4123	ceph_seq_cmp(a: tcap->seq, b: t_issue_seq) < 0) {
4124	doutc(cl, " updating import cap %p mds%d\n", tcap,
4125	target);
4126	tcap->cap_id = t_cap_id;
4127	tcap->seq = t_issue_seq - 1;
4128	tcap->issue_seq = t_issue_seq - 1;
4129	tcap->issued |= issued;
4130	tcap->implemented |= issued;
4131	if (cap == ci->i_auth_cap) {
4132	ci->i_auth_cap = tcap;
4133	change_auth_cap_ses(ci, session: tcap->session);
4134	}
4135	}
4136	ceph_remove_cap(mdsc, cap, queue_release: false);
4137	goto out_unlock;
4138	} else if (tsession) {
4139	/* add placeholder for the export target */
4140	int flag = (cap == ci->i_auth_cap) ? CEPH_CAP_FLAG_AUTH : 0;
4141	tcap = new_cap;
4142	ceph_add_cap(inode, session: tsession, cap_id: t_cap_id, issued, wanted: 0,
4143	seq: t_issue_seq - 1, mseq: t_mseq, realmino: (u64)-1, flags: flag, new_cap: &new_cap);
4144
4145	if (!list_empty(head: &ci->i_cap_flush_list) &&
4146	ci->i_auth_cap == tcap) {
4147	spin_lock(lock: &mdsc->cap_dirty_lock);
4148	list_move_tail(list: &ci->i_flushing_item,
4149	head: &tcap->session->s_cap_flushing);
4150	spin_unlock(lock: &mdsc->cap_dirty_lock);
4151	}
4152
4153	ceph_remove_cap(mdsc, cap, queue_release: false);
4154	goto out_unlock;
4155	}
4156
4157	spin_unlock(lock: &ci->i_ceph_lock);
4158	up_read(sem: &mdsc->snap_rwsem);
4159	mutex_unlock(lock: &session->s_mutex);
4160
4161	/* open target session */
4162	tsession = ceph_mdsc_open_export_target_session(mdsc, target);
4163	if (!IS_ERR(ptr: tsession)) {
4164	if (mds > target) {
4165	mutex_lock(&session->s_mutex);
4166	mutex_lock_nested(lock: &tsession->s_mutex,
4167	SINGLE_DEPTH_NESTING);
4168	} else {
4169	mutex_lock(&tsession->s_mutex);
4170	mutex_lock_nested(lock: &session->s_mutex,
4171	SINGLE_DEPTH_NESTING);
4172	}
4173	new_cap = ceph_get_cap(mdsc, NULL);
4174	} else {
4175	WARN_ON(1);
4176	tsession = NULL;
4177	target = -1;
4178	mutex_lock(&session->s_mutex);
4179	}
4180	goto retry;
4181
4182	out_unlock:
4183	spin_unlock(lock: &ci->i_ceph_lock);
4184	up_read(sem: &mdsc->snap_rwsem);
4185	mutex_unlock(lock: &session->s_mutex);
4186	if (tsession) {
4187	mutex_unlock(lock: &tsession->s_mutex);
4188	ceph_put_mds_session(s: tsession);
4189	}
4190	if (new_cap)
4191	ceph_put_cap(mdsc, cap: new_cap);
4192	}
4193
4194	/*
4195	* Handle cap IMPORT.
4196	*
4197	* caller holds s_mutex. acquires i_ceph_lock
4198	*/
4199	static void handle_cap_import(struct ceph_mds_client *mdsc,
4200	struct inode *inode, struct ceph_mds_caps *im,
4201	struct ceph_mds_cap_peer *ph,
4202	struct ceph_mds_session *session,
4203	struct ceph_cap **target_cap, int *old_issued)
4204	{
4205	struct ceph_inode_info *ci = ceph_inode(inode);
4206	struct ceph_client *cl = mdsc->fsc->client;
4207	struct ceph_cap *cap, *ocap, *new_cap = NULL;
4208	int mds = session->s_mds;
4209	int issued;
4210	unsigned caps = le32_to_cpu(im->caps);
4211	unsigned wanted = le32_to_cpu(im->wanted);
4212	unsigned seq = le32_to_cpu(im->seq);
4213	unsigned mseq = le32_to_cpu(im->migrate_seq);
4214	u64 realmino = le64_to_cpu(im->realm);
4215	u64 cap_id = le64_to_cpu(im->cap_id);
4216	u64 p_cap_id;
4217	u32 piseq = 0;
4218	u32 pmseq = 0;
4219	int peer;
4220
4221	if (ph) {
4222	p_cap_id = le64_to_cpu(ph->cap_id);
4223	peer = le32_to_cpu(ph->mds);
4224	piseq = le32_to_cpu(ph->issue_seq);
4225	pmseq = le32_to_cpu(ph->mseq);
4226	} else {
4227	p_cap_id = 0;
4228	peer = -1;
4229	}
4230
4231	doutc(cl, " cap %llx.%llx import from peer %d piseq %u pmseq %u\n",
4232	ceph_vinop(inode), peer, piseq, pmseq);
4233	retry:
4234	cap = __get_cap_for_mds(ci, mds);
4235	if (!cap) {
4236	if (!new_cap) {
4237	spin_unlock(lock: &ci->i_ceph_lock);
4238	new_cap = ceph_get_cap(mdsc, NULL);
4239	spin_lock(lock: &ci->i_ceph_lock);
4240	goto retry;
4241	}
4242	cap = new_cap;
4243	} else {
4244	if (new_cap) {
4245	ceph_put_cap(mdsc, cap: new_cap);
4246	new_cap = NULL;
4247	}
4248	}
4249
4250	__ceph_caps_issued(ci, implemented: &issued);
4251	issued |= __ceph_caps_dirty(ci);
4252
4253	ceph_add_cap(inode, session, cap_id, issued: caps, wanted, seq, mseq,
4254	realmino, CEPH_CAP_FLAG_AUTH, new_cap: &new_cap);
4255
4256	ocap = peer >= 0 ? __get_cap_for_mds(ci, mds: peer) : NULL;
4257	if (ocap && ocap->cap_id == p_cap_id) {
4258	doutc(cl, " remove export cap %p mds%d flags %d\n",
4259	ocap, peer, ph->flags);
4260	if ((ph->flags & CEPH_CAP_FLAG_AUTH) &&
4261	(ocap->seq != piseq ||
4262	ocap->mseq != pmseq)) {
4263	pr_err_ratelimited_client(cl, "mismatched seq/mseq: "
4264	"%p %llx.%llx mds%d seq %d mseq %d"
4265	" importer mds%d has peer seq %d mseq %d\n",
4266	inode, ceph_vinop(inode), peer,
4267	ocap->seq, ocap->mseq, mds, piseq, pmseq);
4268	}
4269	ceph_remove_cap(mdsc, cap: ocap, queue_release: (ph->flags & CEPH_CAP_FLAG_RELEASE));
4270	}
4271
4272	*old_issued = issued;
4273	*target_cap = cap;
4274	}
4275
4276	#ifdef CONFIG_FS_ENCRYPTION
4277	static int parse_fscrypt_fields(void **p, void *end,
4278	struct cap_extra_info *extra)
4279	{
4280	u32 len;
4281
4282	ceph_decode_32_safe(p, end, extra->fscrypt_auth_len, bad);
4283	if (extra->fscrypt_auth_len) {
4284	ceph_decode_need(p, end, extra->fscrypt_auth_len, bad);
4285	extra->fscrypt_auth = kmalloc(extra->fscrypt_auth_len,
4286	GFP_KERNEL);
4287	if (!extra->fscrypt_auth)
4288	return -ENOMEM;
4289	ceph_decode_copy_safe(p, end, extra->fscrypt_auth,
4290	extra->fscrypt_auth_len, bad);
4291	}
4292
4293	ceph_decode_32_safe(p, end, len, bad);
4294	if (len >= sizeof(u64)) {
4295	ceph_decode_64_safe(p, end, extra->fscrypt_file_size, bad);
4296	len -= sizeof(u64);
4297	}
4298	ceph_decode_skip_n(p, end, len, bad);
4299	return 0;
4300	bad:
4301	return -EIO;
4302	}
4303	#else
4304	static int parse_fscrypt_fields(void **p, void *end,
4305	struct cap_extra_info *extra)
4306	{
4307	u32 len;
4308
4309	/* Don't care about these fields unless we're encryption-capable */
4310	ceph_decode_32_safe(p, end, len, bad);
4311	if (len)
4312	ceph_decode_skip_n(p, end, len, bad);
4313	ceph_decode_32_safe(p, end, len, bad);
4314	if (len)
4315	ceph_decode_skip_n(p, end, len, bad);
4316	return 0;
4317	bad:
4318	return -EIO;
4319	}
4320	#endif
4321
4322	/*
4323	* Handle a caps message from the MDS.
4324	*
4325	* Identify the appropriate session, inode, and call the right handler
4326	* based on the cap op.
4327	*/
4328	void ceph_handle_caps(struct ceph_mds_session *session,
4329	struct ceph_msg *msg)
4330	{
4331	struct ceph_mds_client *mdsc = session->s_mdsc;
4332	struct ceph_client *cl = mdsc->fsc->client;
4333	struct inode *inode;
4334	struct ceph_inode_info *ci;
4335	struct ceph_cap *cap;
4336	struct ceph_mds_caps *h;
4337	struct ceph_mds_cap_peer *peer = NULL;
4338	struct ceph_snap_realm *realm = NULL;
4339	int op;
4340	int msg_version = le16_to_cpu(msg->hdr.version);
4341	u32 seq, mseq, issue_seq;
4342	struct ceph_vino vino;
4343	void *snaptrace;
4344	size_t snaptrace_len;
4345	void *p, *end;
4346	struct cap_extra_info extra_info = {};
4347	bool queue_trunc;
4348	bool close_sessions = false;
4349	bool do_cap_release = false;
4350
4351	if (!ceph_inc_mds_stopping_blocker(mdsc, session))
4352	return;
4353
4354	/* decode */
4355	end = msg->front.iov_base + msg->front.iov_len;
4356	if (msg->front.iov_len < sizeof(*h))
4357	goto bad;
4358	h = msg->front.iov_base;
4359	op = le32_to_cpu(h->op);
4360	vino.ino = le64_to_cpu(h->ino);
4361	vino.snap = CEPH_NOSNAP;
4362	seq = le32_to_cpu(h->seq);
4363	mseq = le32_to_cpu(h->migrate_seq);
4364	issue_seq = le32_to_cpu(h->issue_seq);
4365
4366	snaptrace = h + 1;
4367	snaptrace_len = le32_to_cpu(h->snap_trace_len);
4368	p = snaptrace + snaptrace_len;
4369
4370	if (msg_version >= 2) {
4371	u32 flock_len;
4372	ceph_decode_32_safe(&p, end, flock_len, bad);
4373	if (p + flock_len > end)
4374	goto bad;
4375	p += flock_len;
4376	}
4377
4378	if (msg_version >= 3) {
4379	if (op == CEPH_CAP_OP_IMPORT) {
4380	if (p + sizeof(*peer) > end)
4381	goto bad;
4382	peer = p;
4383	p += sizeof(*peer);
4384	} else if (op == CEPH_CAP_OP_EXPORT) {
4385	/* recorded in unused fields */
4386	peer = (void *)&h->size;
4387	}
4388	}
4389
4390	if (msg_version >= 4) {
4391	ceph_decode_64_safe(&p, end, extra_info.inline_version, bad);
4392	ceph_decode_32_safe(&p, end, extra_info.inline_len, bad);
4393	if (p + extra_info.inline_len > end)
4394	goto bad;
4395	extra_info.inline_data = p;
4396	p += extra_info.inline_len;
4397	}
4398
4399	if (msg_version >= 5) {
4400	struct ceph_osd_client *osdc = &mdsc->fsc->client->osdc;
4401	u32 epoch_barrier;
4402
4403	ceph_decode_32_safe(&p, end, epoch_barrier, bad);
4404	ceph_osdc_update_epoch_barrier(osdc, eb: epoch_barrier);
4405	}
4406
4407	if (msg_version >= 8) {
4408	u32 pool_ns_len;
4409
4410	/* version >= 6 */
4411	ceph_decode_skip_64(&p, end, bad); // flush_tid
4412	/* version >= 7 */
4413	ceph_decode_skip_32(&p, end, bad); // caller_uid
4414	ceph_decode_skip_32(&p, end, bad); // caller_gid
4415	/* version >= 8 */
4416	ceph_decode_32_safe(&p, end, pool_ns_len, bad);
4417	if (pool_ns_len > 0) {
4418	ceph_decode_need(&p, end, pool_ns_len, bad);
4419	extra_info.pool_ns =
4420	ceph_find_or_create_string(str: p, len: pool_ns_len);
4421	p += pool_ns_len;
4422	}
4423	}
4424
4425	if (msg_version >= 9) {
4426	struct ceph_timespec *btime;
4427
4428	if (p + sizeof(*btime) > end)
4429	goto bad;
4430	btime = p;
4431	ceph_decode_timespec64(ts: &extra_info.btime, tv: btime);
4432	p += sizeof(*btime);
4433	ceph_decode_64_safe(&p, end, extra_info.change_attr, bad);
4434	}
4435
4436	if (msg_version >= 11) {
4437	/* version >= 10 */
4438	ceph_decode_skip_32(&p, end, bad); // flags
4439	/* version >= 11 */
4440	extra_info.dirstat_valid = true;
4441	ceph_decode_64_safe(&p, end, extra_info.nfiles, bad);
4442	ceph_decode_64_safe(&p, end, extra_info.nsubdirs, bad);
4443	}
4444
4445	if (msg_version >= 12) {
4446	if (parse_fscrypt_fields(p: &p, end, extra: &extra_info))
4447	goto bad;
4448	}
4449
4450	/* lookup ino */
4451	inode = ceph_find_inode(sb: mdsc->fsc->sb, vino);
4452	doutc(cl, " caps mds%d op %s ino %llx.%llx inode %p seq %u iseq %u mseq %u\n",
4453	session->s_mds, ceph_cap_op_name(op), vino.ino, vino.snap, inode,
4454	seq, issue_seq, mseq);
4455
4456	trace_ceph_handle_caps(mdsc, session, op, vino: &vino, inode: ceph_inode(inode),
4457	seq, mseq: issue_seq, issue_seq: mseq);
4458
4459	mutex_lock(&session->s_mutex);
4460
4461	if (!inode) {
4462	doutc(cl, " i don't have ino %llx\n", vino.ino);
4463
4464	switch (op) {
4465	case CEPH_CAP_OP_IMPORT:
4466	case CEPH_CAP_OP_REVOKE:
4467	case CEPH_CAP_OP_GRANT:
4468	do_cap_release = true;
4469	break;
4470	default:
4471	break;
4472	}
4473	goto flush_cap_releases;
4474	}
4475	ci = ceph_inode(inode);
4476
4477	/* these will work even if we don't have a cap yet */
4478	switch (op) {
4479	case CEPH_CAP_OP_FLUSHSNAP_ACK:
4480	handle_cap_flushsnap_ack(inode, le64_to_cpu(msg->hdr.tid),
4481	m: h, session);
4482	goto done;
4483
4484	case CEPH_CAP_OP_EXPORT:
4485	handle_cap_export(inode, ex: h, ph: peer, session);
4486	goto done_unlocked;
4487
4488	case CEPH_CAP_OP_IMPORT:
4489	realm = NULL;
4490	if (snaptrace_len) {
4491	down_write(sem: &mdsc->snap_rwsem);
4492	if (ceph_update_snap_trace(m: mdsc, p: snaptrace,
4493	e: snaptrace + snaptrace_len,
4494	deletion: false, realm_ret: &realm)) {
4495	up_write(sem: &mdsc->snap_rwsem);
4496	close_sessions = true;
4497	goto done;
4498	}
4499	downgrade_write(sem: &mdsc->snap_rwsem);
4500	} else {
4501	down_read(sem: &mdsc->snap_rwsem);
4502	}
4503	spin_lock(lock: &ci->i_ceph_lock);
4504	handle_cap_import(mdsc, inode, im: h, ph: peer, session,
4505	target_cap: &cap, old_issued: &extra_info.issued);
4506	handle_cap_grant(inode, session, cap,
4507	grant: h, xattr_buf: msg->middle, extra_info: &extra_info);
4508	if (realm)
4509	ceph_put_snap_realm(mdsc, realm);
4510	goto done_unlocked;
4511	}
4512
4513	/* the rest require a cap */
4514	spin_lock(lock: &ci->i_ceph_lock);
4515	cap = __get_cap_for_mds(ci: ceph_inode(inode), mds: session->s_mds);
4516	if (!cap) {
4517	doutc(cl, " no cap on %p ino %llx.%llx from mds%d\n",
4518	inode, ceph_ino(inode), ceph_snap(inode),
4519	session->s_mds);
4520	spin_unlock(lock: &ci->i_ceph_lock);
4521	switch (op) {
4522	case CEPH_CAP_OP_REVOKE:
4523	case CEPH_CAP_OP_GRANT:
4524	do_cap_release = true;
4525	break;
4526	default:
4527	break;
4528	}
4529	goto flush_cap_releases;
4530	}
4531
4532	/* note that each of these drops i_ceph_lock for us */
4533	switch (op) {
4534	case CEPH_CAP_OP_REVOKE:
4535	case CEPH_CAP_OP_GRANT:
4536	__ceph_caps_issued(ci, implemented: &extra_info.issued);
4537	extra_info.issued |= __ceph_caps_dirty(ci);
4538	handle_cap_grant(inode, session, cap,
4539	grant: h, xattr_buf: msg->middle, extra_info: &extra_info);
4540	goto done_unlocked;
4541
4542	case CEPH_CAP_OP_FLUSH_ACK:
4543	handle_cap_flush_ack(inode, le64_to_cpu(msg->hdr.tid),
4544	m: h, session, cap);
4545	break;
4546
4547	case CEPH_CAP_OP_TRUNC:
4548	queue_trunc = handle_cap_trunc(inode, trunc: h, session,
4549	extra_info: &extra_info);
4550	spin_unlock(lock: &ci->i_ceph_lock);
4551	if (queue_trunc)
4552	ceph_queue_vmtruncate(inode);
4553	break;
4554
4555	default:
4556	spin_unlock(lock: &ci->i_ceph_lock);
4557	pr_err_client(cl, "unknown cap op %d %s\n", op,
4558	ceph_cap_op_name(op));
4559	}
4560
4561	done:
4562	mutex_unlock(lock: &session->s_mutex);
4563	done_unlocked:
4564	iput(inode);
4565	out:
4566	ceph_dec_mds_stopping_blocker(mdsc);
4567
4568	ceph_put_string(str: extra_info.pool_ns);
4569
4570	/* Defer closing the sessions after s_mutex lock being released */
4571	if (close_sessions)
4572	ceph_mdsc_close_sessions(mdsc);
4573
4574	kfree(objp: extra_info.fscrypt_auth);
4575	return;
4576
4577	flush_cap_releases:
4578	/*
4579	* send any cap release message to try to move things
4580	* along for the mds (who clearly thinks we still have this
4581	* cap).
4582	*/
4583	if (do_cap_release) {
4584	cap = ceph_get_cap(mdsc, NULL);
4585	cap->cap_ino = vino.ino;
4586	cap->queue_release = 1;
4587	cap->cap_id = le64_to_cpu(h->cap_id);
4588	cap->mseq = mseq;
4589	cap->seq = seq;
4590	cap->issue_seq = seq;
4591	spin_lock(lock: &session->s_cap_lock);
4592	__ceph_queue_cap_release(session, cap);
4593	spin_unlock(lock: &session->s_cap_lock);
4594	}
4595	ceph_flush_session_cap_releases(mdsc, session);
4596	goto done;
4597
4598	bad:
4599	pr_err_client(cl, "corrupt message\n");
4600	ceph_msg_dump(msg);
4601	goto out;
4602	}
4603
4604	/*
4605	* Delayed work handler to process end of delayed cap release LRU list.
4606	*
4607	* If new caps are added to the list while processing it, these won't get
4608	* processed in this run. In this case, the ci->i_hold_caps_max will be
4609	* returned so that the work can be scheduled accordingly.
4610	*/
4611	unsigned long ceph_check_delayed_caps(struct ceph_mds_client *mdsc)
4612	{
4613	struct ceph_client *cl = mdsc->fsc->client;
4614	struct inode *inode;
4615	struct ceph_inode_info *ci;
4616	struct ceph_mount_options *opt = mdsc->fsc->mount_options;
4617	unsigned long delay_max = opt->caps_wanted_delay_max * HZ;
4618	unsigned long loop_start = jiffies;
4619	unsigned long delay = 0;
4620
4621	doutc(cl, "begin\n");
4622	spin_lock(lock: &mdsc->cap_delay_lock);
4623	while (!list_empty(head: &mdsc->cap_delay_list)) {
4624	ci = list_first_entry(&mdsc->cap_delay_list,
4625	struct ceph_inode_info,
4626	i_cap_delay_list);
4627	if (time_before(loop_start, ci->i_hold_caps_max - delay_max)) {
4628	doutc(cl, "caps added recently. Exiting loop");
4629	delay = ci->i_hold_caps_max;
4630	break;
4631	}
4632	if ((ci->i_ceph_flags & CEPH_I_FLUSH) == 0 &&
4633	time_before(jiffies, ci->i_hold_caps_max))
4634	break;
4635	list_del_init(entry: &ci->i_cap_delay_list);
4636
4637	inode = igrab(&ci->netfs.inode);
4638	if (inode) {
4639	spin_unlock(lock: &mdsc->cap_delay_lock);
4640	doutc(cl, "on %p %llx.%llx\n", inode,
4641	ceph_vinop(inode));
4642	ceph_check_caps(ci, flags: 0);
4643	iput(inode);
4644	spin_lock(lock: &mdsc->cap_delay_lock);
4645	}
4646
4647	/*
4648	* Make sure too many dirty caps or general
4649	* slowness doesn't block mdsc delayed work,
4650	* preventing send_renew_caps() from running.
4651	*/
4652	if (time_after_eq(jiffies, loop_start + 5 * HZ))
4653	break;
4654	}
4655	spin_unlock(lock: &mdsc->cap_delay_lock);
4656	doutc(cl, "done\n");
4657
4658	return delay;
4659	}
4660
4661	/*
4662	* Flush all dirty caps to the mds
4663	*/
4664	static void flush_dirty_session_caps(struct ceph_mds_session *s)
4665	{
4666	struct ceph_mds_client *mdsc = s->s_mdsc;
4667	struct ceph_client *cl = mdsc->fsc->client;
4668	struct ceph_inode_info *ci;
4669	struct inode *inode;
4670
4671	doutc(cl, "begin\n");
4672	spin_lock(lock: &mdsc->cap_dirty_lock);
4673	while (!list_empty(head: &s->s_cap_dirty)) {
4674	ci = list_first_entry(&s->s_cap_dirty, struct ceph_inode_info,
4675	i_dirty_item);
4676	inode = &ci->netfs.inode;
4677	ihold(inode);
4678	doutc(cl, "%p %llx.%llx\n", inode, ceph_vinop(inode));
4679	spin_unlock(lock: &mdsc->cap_dirty_lock);
4680	ceph_wait_on_async_create(inode);
4681	ceph_check_caps(ci, CHECK_CAPS_FLUSH);
4682	iput(inode);
4683	spin_lock(lock: &mdsc->cap_dirty_lock);
4684	}
4685	spin_unlock(lock: &mdsc->cap_dirty_lock);
4686	doutc(cl, "done\n");
4687	}
4688
4689	void ceph_flush_dirty_caps(struct ceph_mds_client *mdsc)
4690	{
4691	ceph_mdsc_iterate_sessions(mdsc, cb: flush_dirty_session_caps, check_state: true);
4692	}
4693
4694	/*
4695	* Flush all cap releases to the mds
4696	*/
4697	static void flush_cap_releases(struct ceph_mds_session *s)
4698	{
4699	struct ceph_mds_client *mdsc = s->s_mdsc;
4700	struct ceph_client *cl = mdsc->fsc->client;
4701
4702	doutc(cl, "begin\n");
4703	spin_lock(lock: &s->s_cap_lock);
4704	if (s->s_num_cap_releases)
4705	ceph_flush_session_cap_releases(mdsc, session: s);
4706	spin_unlock(lock: &s->s_cap_lock);
4707	doutc(cl, "done\n");
4708
4709	}
4710
4711	void ceph_flush_cap_releases(struct ceph_mds_client *mdsc)
4712	{
4713	ceph_mdsc_iterate_sessions(mdsc, cb: flush_cap_releases, check_state: true);
4714	}
4715
4716	void __ceph_touch_fmode(struct ceph_inode_info *ci,
4717	struct ceph_mds_client *mdsc, int fmode)
4718	{
4719	unsigned long now = jiffies;
4720	if (fmode & CEPH_FILE_MODE_RD)
4721	ci->i_last_rd = now;
4722	if (fmode & CEPH_FILE_MODE_WR)
4723	ci->i_last_wr = now;
4724	/* queue periodic check */
4725	if (fmode &&
4726	__ceph_is_any_real_caps(ci) &&
4727	list_empty(head: &ci->i_cap_delay_list))
4728	__cap_delay_requeue(mdsc, ci);
4729	}
4730
4731	void ceph_get_fmode(struct ceph_inode_info *ci, int fmode, int count)
4732	{
4733	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(sb: ci->netfs.inode.i_sb);
4734	int bits = (fmode << 1) | 1;
4735	bool already_opened = false;
4736	int i;
4737
4738	if (count == 1)
4739	atomic64_inc(v: &mdsc->metric.opened_files);
4740
4741	spin_lock(lock: &ci->i_ceph_lock);
4742	for (i = 0; i < CEPH_FILE_MODE_BITS; i++) {
4743	/*
4744	* If any of the mode ref is larger than 0,
4745	* that means it has been already opened by
4746	* others. Just skip checking the PIN ref.
4747	*/
4748	if (i && ci->i_nr_by_mode[i])
4749	already_opened = true;
4750
4751	if (bits & (1 << i))
4752	ci->i_nr_by_mode[i] += count;
4753	}
4754
4755	if (!already_opened)
4756	percpu_counter_inc(fbc: &mdsc->metric.opened_inodes);
4757	spin_unlock(lock: &ci->i_ceph_lock);
4758	}
4759
4760	/*
4761	* Drop open file reference. If we were the last open file,
4762	* we may need to release capabilities to the MDS (or schedule
4763	* their delayed release).
4764	*/
4765	void ceph_put_fmode(struct ceph_inode_info *ci, int fmode, int count)
4766	{
4767	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(sb: ci->netfs.inode.i_sb);
4768	int bits = (fmode << 1) | 1;
4769	bool is_closed = true;
4770	int i;
4771
4772	if (count == 1)
4773	atomic64_dec(v: &mdsc->metric.opened_files);
4774
4775	spin_lock(lock: &ci->i_ceph_lock);
4776	for (i = 0; i < CEPH_FILE_MODE_BITS; i++) {
4777	if (bits & (1 << i)) {
4778	BUG_ON(ci->i_nr_by_mode[i] < count);
4779	ci->i_nr_by_mode[i] -= count;
4780	}
4781
4782	/*
4783	* If any of the mode ref is not 0 after
4784	* decreased, that means it is still opened
4785	* by others. Just skip checking the PIN ref.
4786	*/
4787	if (i && ci->i_nr_by_mode[i])
4788	is_closed = false;
4789	}
4790
4791	if (is_closed)
4792	percpu_counter_dec(fbc: &mdsc->metric.opened_inodes);
4793	spin_unlock(lock: &ci->i_ceph_lock);
4794	}
4795
4796	/*
4797	* For a soon-to-be unlinked file, drop the LINK caps. If it
4798	* looks like the link count will hit 0, drop any other caps (other
4799	* than PIN) we don't specifically want (due to the file still being
4800	* open).
4801	*/
4802	int ceph_drop_caps_for_unlink(struct inode *inode)
4803	{
4804	struct ceph_inode_info *ci = ceph_inode(inode);
4805	int drop = CEPH_CAP_LINK_SHARED | CEPH_CAP_LINK_EXCL;
4806
4807	spin_lock(lock: &ci->i_ceph_lock);
4808	if (inode->i_nlink == 1) {
4809	drop |= ~(__ceph_caps_wanted(ci) | CEPH_CAP_PIN);
4810
4811	if (__ceph_caps_dirty(ci)) {
4812	struct ceph_mds_client *mdsc =
4813	ceph_inode_to_fs_client(inode)->mdsc;
4814
4815	doutc(mdsc->fsc->client, "%p %llx.%llx\n", inode,
4816	ceph_vinop(inode));
4817	spin_lock(lock: &mdsc->cap_delay_lock);
4818	ci->i_ceph_flags |= CEPH_I_FLUSH;
4819	if (!list_empty(head: &ci->i_cap_delay_list))
4820	list_del_init(entry: &ci->i_cap_delay_list);
4821	list_add_tail(new: &ci->i_cap_delay_list,
4822	head: &mdsc->cap_unlink_delay_list);
4823	spin_unlock(lock: &mdsc->cap_delay_lock);
4824
4825	/*
4826	* Fire the work immediately, because the MDS maybe
4827	* waiting for caps release.
4828	*/
4829	ceph_queue_cap_unlink_work(mdsc);
4830	}
4831	}
4832	spin_unlock(lock: &ci->i_ceph_lock);
4833	return drop;
4834	}
4835
4836	/*
4837	* Helpers for embedding cap and dentry lease releases into mds
4838	* requests.
4839	*
4840	* @force is used by dentry_release (below) to force inclusion of a
4841	* record for the directory inode, even when there aren't any caps to
4842	* drop.
4843	*/
4844	int ceph_encode_inode_release(void **p, struct inode *inode,
4845	int mds, int drop, int unless, int force)
4846	{
4847	struct ceph_inode_info *ci = ceph_inode(inode);
4848	struct ceph_client *cl = ceph_inode_to_client(inode);
4849	struct ceph_cap *cap;
4850	struct ceph_mds_request_release *rel = *p;
4851	int used, dirty;
4852	int ret = 0;
4853
4854	spin_lock(lock: &ci->i_ceph_lock);
4855	used = __ceph_caps_used(ci);
4856	dirty = __ceph_caps_dirty(ci);
4857
4858	doutc(cl, "%p %llx.%llx mds%d used|dirty %s drop %s unless %s\n",
4859	inode, ceph_vinop(inode), mds, ceph_cap_string(used|dirty),
4860	ceph_cap_string(drop), ceph_cap_string(unless));
4861
4862	/* only drop unused, clean caps */
4863	drop &= ~(used | dirty);
4864
4865	cap = __get_cap_for_mds(ci, mds);
4866	if (cap && __cap_is_valid(cap)) {
4867	unless &= cap->issued;
4868	if (unless) {
4869	if (unless & CEPH_CAP_AUTH_EXCL)
4870	drop &= ~CEPH_CAP_AUTH_SHARED;
4871	if (unless & CEPH_CAP_LINK_EXCL)
4872	drop &= ~CEPH_CAP_LINK_SHARED;
4873	if (unless & CEPH_CAP_XATTR_EXCL)
4874	drop &= ~CEPH_CAP_XATTR_SHARED;
4875	if (unless & CEPH_CAP_FILE_EXCL)
4876	drop &= ~CEPH_CAP_FILE_SHARED;
4877	}
4878
4879	if (force || (cap->issued & drop)) {
4880	if (cap->issued & drop) {
4881	int wanted = __ceph_caps_wanted(ci);
4882	doutc(cl, "%p %llx.%llx cap %p %s -> %s, "
4883	"wanted %s -> %s\n", inode,
4884	ceph_vinop(inode), cap,
4885	ceph_cap_string(cap->issued),
4886	ceph_cap_string(cap->issued & ~drop),
4887	ceph_cap_string(cap->mds_wanted),
4888	ceph_cap_string(wanted));
4889
4890	cap->issued &= ~drop;
4891	cap->implemented &= ~drop;
4892	cap->mds_wanted = wanted;
4893	if (cap == ci->i_auth_cap &&
4894	!(wanted & CEPH_CAP_ANY_FILE_WR))
4895	ci->i_requested_max_size = 0;
4896	} else {
4897	doutc(cl, "%p %llx.%llx cap %p %s (force)\n",
4898	inode, ceph_vinop(inode), cap,
4899	ceph_cap_string(cap->issued));
4900	}
4901
4902	rel->ino = cpu_to_le64(ceph_ino(inode));
4903	rel->cap_id = cpu_to_le64(cap->cap_id);
4904	rel->seq = cpu_to_le32(cap->seq);
4905	rel->issue_seq = cpu_to_le32(cap->issue_seq);
4906	rel->mseq = cpu_to_le32(cap->mseq);
4907	rel->caps = cpu_to_le32(cap->implemented);
4908	rel->wanted = cpu_to_le32(cap->mds_wanted);
4909	rel->dname_len = 0;
4910	rel->dname_seq = 0;
4911	*p += sizeof(*rel);
4912	ret = 1;
4913	} else {
4914	doutc(cl, "%p %llx.%llx cap %p %s (noop)\n",
4915	inode, ceph_vinop(inode), cap,
4916	ceph_cap_string(cap->issued));
4917	}
4918	}
4919	spin_unlock(lock: &ci->i_ceph_lock);
4920	return ret;
4921	}
4922
4923	/**
4924	* ceph_encode_dentry_release - encode a dentry release into an outgoing request
4925	* @p: outgoing request buffer
4926	* @dentry: dentry to release
4927	* @dir: dir to release it from
4928	* @mds: mds that we're speaking to
4929	* @drop: caps being dropped
4930	* @unless: unless we have these caps
4931	*
4932	* Encode a dentry release into an outgoing request buffer. Returns 1 if the
4933	* thing was released, or a negative error code otherwise.
4934	*/
4935	int ceph_encode_dentry_release(void **p, struct dentry *dentry,
4936	struct inode *dir,
4937	int mds, int drop, int unless)
4938	{
4939	struct ceph_mds_request_release *rel = *p;
4940	struct ceph_dentry_info *di = ceph_dentry(dentry);
4941	struct ceph_client *cl;
4942	int force = 0;
4943	int ret;
4944
4945	/* This shouldn't happen */
4946	BUG_ON(!dir);
4947
4948	/*
4949	* force an record for the directory caps if we have a dentry lease.
4950	* this is racy (can't take i_ceph_lock and d_lock together), but it
4951	* doesn't have to be perfect; the mds will revoke anything we don't
4952	* release.
4953	*/
4954	spin_lock(lock: &dentry->d_lock);
4955	if (di->lease_session && di->lease_session->s_mds == mds)
4956	force = 1;
4957	spin_unlock(lock: &dentry->d_lock);
4958
4959	ret = ceph_encode_inode_release(p, inode: dir, mds, drop, unless, force);
4960
4961	cl = ceph_inode_to_client(inode: dir);
4962	spin_lock(lock: &dentry->d_lock);
4963	if (ret && di->lease_session && di->lease_session->s_mds == mds) {
4964	int len = dentry->d_name.len;
4965	doutc(cl, "%p mds%d seq %d\n", dentry, mds,
4966	(int)di->lease_seq);
4967	rel->dname_seq = cpu_to_le32(di->lease_seq);
4968	__ceph_mdsc_drop_dentry_lease(dentry);
4969	memcpy(*p, dentry->d_name.name, len);
4970	spin_unlock(lock: &dentry->d_lock);
4971	if (IS_ENCRYPTED(dir) && fscrypt_has_encryption_key(inode: dir)) {
4972	len = ceph_encode_encrypted_dname(parent: dir, buf: *p, len);
4973	if (len < 0)
4974	return len;
4975	}
4976	rel->dname_len = cpu_to_le32(len);
4977	*p += len;
4978	} else {
4979	spin_unlock(lock: &dentry->d_lock);
4980	}
4981	return ret;
4982	}
4983
4984	static int remove_capsnaps(struct ceph_mds_client *mdsc, struct inode *inode)
4985	{
4986	struct ceph_inode_info *ci = ceph_inode(inode);
4987	struct ceph_client *cl = mdsc->fsc->client;
4988	struct ceph_cap_snap *capsnap;
4989	int capsnap_release = 0;
4990
4991	lockdep_assert_held(&ci->i_ceph_lock);
4992
4993	doutc(cl, "removing capsnaps, ci is %p, %p %llx.%llx\n",
4994	ci, inode, ceph_vinop(inode));
4995
4996	while (!list_empty(head: &ci->i_cap_snaps)) {
4997	capsnap = list_first_entry(&ci->i_cap_snaps,
4998	struct ceph_cap_snap, ci_item);
4999	__ceph_remove_capsnap(inode, capsnap, NULL, NULL);
5000	ceph_put_snap_context(sc: capsnap->context);
5001	ceph_put_cap_snap(capsnap);
5002	capsnap_release++;
5003	}
5004	wake_up_all(&ci->i_cap_wq);
5005	wake_up_all(&mdsc->cap_flushing_wq);
5006	return capsnap_release;
5007	}
5008
5009	int ceph_purge_inode_cap(struct inode *inode, struct ceph_cap *cap, bool *invalidate)
5010	{
5011	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
5012	struct ceph_mds_client *mdsc = fsc->mdsc;
5013	struct ceph_client *cl = fsc->client;
5014	struct ceph_inode_info *ci = ceph_inode(inode);
5015	bool is_auth;
5016	bool dirty_dropped = false;
5017	int iputs = 0;
5018
5019	lockdep_assert_held(&ci->i_ceph_lock);
5020
5021	doutc(cl, "removing cap %p, ci is %p, %p %llx.%llx\n",
5022	cap, ci, inode, ceph_vinop(inode));
5023
5024	is_auth = (cap == ci->i_auth_cap);
5025	__ceph_remove_cap(cap, queue_release: false);
5026	if (is_auth) {
5027	struct ceph_cap_flush *cf;
5028
5029	if (ceph_inode_is_shutdown(inode)) {
5030	if (inode->i_data.nrpages > 0)
5031	*invalidate = true;
5032	if (ci->i_wrbuffer_ref > 0)
5033	mapping_set_error(mapping: &inode->i_data, error: -EIO);
5034	}
5035
5036	spin_lock(lock: &mdsc->cap_dirty_lock);
5037
5038	/* trash all of the cap flushes for this inode */
5039	while (!list_empty(head: &ci->i_cap_flush_list)) {
5040	cf = list_first_entry(&ci->i_cap_flush_list,
5041	struct ceph_cap_flush, i_list);
5042	list_del_init(entry: &cf->g_list);
5043	list_del_init(entry: &cf->i_list);
5044	if (!cf->is_capsnap)
5045	ceph_free_cap_flush(cf);
5046	}
5047
5048	if (!list_empty(head: &ci->i_dirty_item)) {
5049	pr_warn_ratelimited_client(cl,
5050	" dropping dirty %s state for %p %llx.%llx\n",
5051	ceph_cap_string(ci->i_dirty_caps),
5052	inode, ceph_vinop(inode));
5053	ci->i_dirty_caps = 0;
5054	list_del_init(entry: &ci->i_dirty_item);
5055	dirty_dropped = true;
5056	}
5057	if (!list_empty(head: &ci->i_flushing_item)) {
5058	pr_warn_ratelimited_client(cl,
5059	" dropping dirty+flushing %s state for %p %llx.%llx\n",
5060	ceph_cap_string(ci->i_flushing_caps),
5061	inode, ceph_vinop(inode));
5062	ci->i_flushing_caps = 0;
5063	list_del_init(entry: &ci->i_flushing_item);
5064	mdsc->num_cap_flushing--;
5065	dirty_dropped = true;
5066	}
5067	spin_unlock(lock: &mdsc->cap_dirty_lock);
5068
5069	if (dirty_dropped) {
5070	mapping_set_error(mapping: inode->i_mapping, error: -EIO);
5071
5072	if (ci->i_wrbuffer_ref_head == 0 &&
5073	ci->i_wr_ref == 0 &&
5074	ci->i_dirty_caps == 0 &&
5075	ci->i_flushing_caps == 0) {
5076	ceph_put_snap_context(sc: ci->i_head_snapc);
5077	ci->i_head_snapc = NULL;
5078	}
5079	}
5080
5081	if (atomic_read(v: &ci->i_filelock_ref) > 0) {
5082	/* make further file lock syscall return -EIO */
5083	ci->i_ceph_flags |= CEPH_I_ERROR_FILELOCK;
5084	pr_warn_ratelimited_client(cl,
5085	" dropping file locks for %p %llx.%llx\n",
5086	inode, ceph_vinop(inode));
5087	}
5088
5089	if (!ci->i_dirty_caps && ci->i_prealloc_cap_flush) {
5090	cf = ci->i_prealloc_cap_flush;
5091	ci->i_prealloc_cap_flush = NULL;
5092	if (!cf->is_capsnap)
5093	ceph_free_cap_flush(cf);
5094	}
5095
5096	if (!list_empty(head: &ci->i_cap_snaps))
5097	iputs = remove_capsnaps(mdsc, inode);
5098	}
5099	if (dirty_dropped)
5100	++iputs;
5101	return iputs;
5102	}
5103