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