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