linux - fs/ceph/caps.c

5192 lines · 141.9 KiB

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