xref: /openbmc/linux/fs/ceph/mds_client.c (revision 81d67439)
1 #include <linux/ceph/ceph_debug.h>
2 
3 #include <linux/fs.h>
4 #include <linux/wait.h>
5 #include <linux/slab.h>
6 #include <linux/sched.h>
7 #include <linux/debugfs.h>
8 #include <linux/seq_file.h>
9 
10 #include "super.h"
11 #include "mds_client.h"
12 
13 #include <linux/ceph/messenger.h>
14 #include <linux/ceph/decode.h>
15 #include <linux/ceph/pagelist.h>
16 #include <linux/ceph/auth.h>
17 #include <linux/ceph/debugfs.h>
18 
19 /*
20  * A cluster of MDS (metadata server) daemons is responsible for
21  * managing the file system namespace (the directory hierarchy and
22  * inodes) and for coordinating shared access to storage.  Metadata is
23  * partitioning hierarchically across a number of servers, and that
24  * partition varies over time as the cluster adjusts the distribution
25  * in order to balance load.
26  *
27  * The MDS client is primarily responsible to managing synchronous
28  * metadata requests for operations like open, unlink, and so forth.
29  * If there is a MDS failure, we find out about it when we (possibly
30  * request and) receive a new MDS map, and can resubmit affected
31  * requests.
32  *
33  * For the most part, though, we take advantage of a lossless
34  * communications channel to the MDS, and do not need to worry about
35  * timing out or resubmitting requests.
36  *
37  * We maintain a stateful "session" with each MDS we interact with.
38  * Within each session, we sent periodic heartbeat messages to ensure
39  * any capabilities or leases we have been issues remain valid.  If
40  * the session times out and goes stale, our leases and capabilities
41  * are no longer valid.
42  */
43 
44 struct ceph_reconnect_state {
45 	struct ceph_pagelist *pagelist;
46 	bool flock;
47 };
48 
49 static void __wake_requests(struct ceph_mds_client *mdsc,
50 			    struct list_head *head);
51 
52 static const struct ceph_connection_operations mds_con_ops;
53 
54 
55 /*
56  * mds reply parsing
57  */
58 
59 /*
60  * parse individual inode info
61  */
62 static int parse_reply_info_in(void **p, void *end,
63 			       struct ceph_mds_reply_info_in *info,
64 			       int features)
65 {
66 	int err = -EIO;
67 
68 	info->in = *p;
69 	*p += sizeof(struct ceph_mds_reply_inode) +
70 		sizeof(*info->in->fragtree.splits) *
71 		le32_to_cpu(info->in->fragtree.nsplits);
72 
73 	ceph_decode_32_safe(p, end, info->symlink_len, bad);
74 	ceph_decode_need(p, end, info->symlink_len, bad);
75 	info->symlink = *p;
76 	*p += info->symlink_len;
77 
78 	if (features & CEPH_FEATURE_DIRLAYOUTHASH)
79 		ceph_decode_copy_safe(p, end, &info->dir_layout,
80 				      sizeof(info->dir_layout), bad);
81 	else
82 		memset(&info->dir_layout, 0, sizeof(info->dir_layout));
83 
84 	ceph_decode_32_safe(p, end, info->xattr_len, bad);
85 	ceph_decode_need(p, end, info->xattr_len, bad);
86 	info->xattr_data = *p;
87 	*p += info->xattr_len;
88 	return 0;
89 bad:
90 	return err;
91 }
92 
93 /*
94  * parse a normal reply, which may contain a (dir+)dentry and/or a
95  * target inode.
96  */
97 static int parse_reply_info_trace(void **p, void *end,
98 				  struct ceph_mds_reply_info_parsed *info,
99 				  int features)
100 {
101 	int err;
102 
103 	if (info->head->is_dentry) {
104 		err = parse_reply_info_in(p, end, &info->diri, features);
105 		if (err < 0)
106 			goto out_bad;
107 
108 		if (unlikely(*p + sizeof(*info->dirfrag) > end))
109 			goto bad;
110 		info->dirfrag = *p;
111 		*p += sizeof(*info->dirfrag) +
112 			sizeof(u32)*le32_to_cpu(info->dirfrag->ndist);
113 		if (unlikely(*p > end))
114 			goto bad;
115 
116 		ceph_decode_32_safe(p, end, info->dname_len, bad);
117 		ceph_decode_need(p, end, info->dname_len, bad);
118 		info->dname = *p;
119 		*p += info->dname_len;
120 		info->dlease = *p;
121 		*p += sizeof(*info->dlease);
122 	}
123 
124 	if (info->head->is_target) {
125 		err = parse_reply_info_in(p, end, &info->targeti, features);
126 		if (err < 0)
127 			goto out_bad;
128 	}
129 
130 	if (unlikely(*p != end))
131 		goto bad;
132 	return 0;
133 
134 bad:
135 	err = -EIO;
136 out_bad:
137 	pr_err("problem parsing mds trace %d\n", err);
138 	return err;
139 }
140 
141 /*
142  * parse readdir results
143  */
144 static int parse_reply_info_dir(void **p, void *end,
145 				struct ceph_mds_reply_info_parsed *info,
146 				int features)
147 {
148 	u32 num, i = 0;
149 	int err;
150 
151 	info->dir_dir = *p;
152 	if (*p + sizeof(*info->dir_dir) > end)
153 		goto bad;
154 	*p += sizeof(*info->dir_dir) +
155 		sizeof(u32)*le32_to_cpu(info->dir_dir->ndist);
156 	if (*p > end)
157 		goto bad;
158 
159 	ceph_decode_need(p, end, sizeof(num) + 2, bad);
160 	num = ceph_decode_32(p);
161 	info->dir_end = ceph_decode_8(p);
162 	info->dir_complete = ceph_decode_8(p);
163 	if (num == 0)
164 		goto done;
165 
166 	/* alloc large array */
167 	info->dir_nr = num;
168 	info->dir_in = kcalloc(num, sizeof(*info->dir_in) +
169 			       sizeof(*info->dir_dname) +
170 			       sizeof(*info->dir_dname_len) +
171 			       sizeof(*info->dir_dlease),
172 			       GFP_NOFS);
173 	if (info->dir_in == NULL) {
174 		err = -ENOMEM;
175 		goto out_bad;
176 	}
177 	info->dir_dname = (void *)(info->dir_in + num);
178 	info->dir_dname_len = (void *)(info->dir_dname + num);
179 	info->dir_dlease = (void *)(info->dir_dname_len + num);
180 
181 	while (num) {
182 		/* dentry */
183 		ceph_decode_need(p, end, sizeof(u32)*2, bad);
184 		info->dir_dname_len[i] = ceph_decode_32(p);
185 		ceph_decode_need(p, end, info->dir_dname_len[i], bad);
186 		info->dir_dname[i] = *p;
187 		*p += info->dir_dname_len[i];
188 		dout("parsed dir dname '%.*s'\n", info->dir_dname_len[i],
189 		     info->dir_dname[i]);
190 		info->dir_dlease[i] = *p;
191 		*p += sizeof(struct ceph_mds_reply_lease);
192 
193 		/* inode */
194 		err = parse_reply_info_in(p, end, &info->dir_in[i], features);
195 		if (err < 0)
196 			goto out_bad;
197 		i++;
198 		num--;
199 	}
200 
201 done:
202 	if (*p != end)
203 		goto bad;
204 	return 0;
205 
206 bad:
207 	err = -EIO;
208 out_bad:
209 	pr_err("problem parsing dir contents %d\n", err);
210 	return err;
211 }
212 
213 /*
214  * parse fcntl F_GETLK results
215  */
216 static int parse_reply_info_filelock(void **p, void *end,
217 				     struct ceph_mds_reply_info_parsed *info,
218 				     int features)
219 {
220 	if (*p + sizeof(*info->filelock_reply) > end)
221 		goto bad;
222 
223 	info->filelock_reply = *p;
224 	*p += sizeof(*info->filelock_reply);
225 
226 	if (unlikely(*p != end))
227 		goto bad;
228 	return 0;
229 
230 bad:
231 	return -EIO;
232 }
233 
234 /*
235  * parse extra results
236  */
237 static int parse_reply_info_extra(void **p, void *end,
238 				  struct ceph_mds_reply_info_parsed *info,
239 				  int features)
240 {
241 	if (info->head->op == CEPH_MDS_OP_GETFILELOCK)
242 		return parse_reply_info_filelock(p, end, info, features);
243 	else
244 		return parse_reply_info_dir(p, end, info, features);
245 }
246 
247 /*
248  * parse entire mds reply
249  */
250 static int parse_reply_info(struct ceph_msg *msg,
251 			    struct ceph_mds_reply_info_parsed *info,
252 			    int features)
253 {
254 	void *p, *end;
255 	u32 len;
256 	int err;
257 
258 	info->head = msg->front.iov_base;
259 	p = msg->front.iov_base + sizeof(struct ceph_mds_reply_head);
260 	end = p + msg->front.iov_len - sizeof(struct ceph_mds_reply_head);
261 
262 	/* trace */
263 	ceph_decode_32_safe(&p, end, len, bad);
264 	if (len > 0) {
265 		err = parse_reply_info_trace(&p, p+len, info, features);
266 		if (err < 0)
267 			goto out_bad;
268 	}
269 
270 	/* extra */
271 	ceph_decode_32_safe(&p, end, len, bad);
272 	if (len > 0) {
273 		err = parse_reply_info_extra(&p, p+len, info, features);
274 		if (err < 0)
275 			goto out_bad;
276 	}
277 
278 	/* snap blob */
279 	ceph_decode_32_safe(&p, end, len, bad);
280 	info->snapblob_len = len;
281 	info->snapblob = p;
282 	p += len;
283 
284 	if (p != end)
285 		goto bad;
286 	return 0;
287 
288 bad:
289 	err = -EIO;
290 out_bad:
291 	pr_err("mds parse_reply err %d\n", err);
292 	return err;
293 }
294 
295 static void destroy_reply_info(struct ceph_mds_reply_info_parsed *info)
296 {
297 	kfree(info->dir_in);
298 }
299 
300 
301 /*
302  * sessions
303  */
304 static const char *session_state_name(int s)
305 {
306 	switch (s) {
307 	case CEPH_MDS_SESSION_NEW: return "new";
308 	case CEPH_MDS_SESSION_OPENING: return "opening";
309 	case CEPH_MDS_SESSION_OPEN: return "open";
310 	case CEPH_MDS_SESSION_HUNG: return "hung";
311 	case CEPH_MDS_SESSION_CLOSING: return "closing";
312 	case CEPH_MDS_SESSION_RESTARTING: return "restarting";
313 	case CEPH_MDS_SESSION_RECONNECTING: return "reconnecting";
314 	default: return "???";
315 	}
316 }
317 
318 static struct ceph_mds_session *get_session(struct ceph_mds_session *s)
319 {
320 	if (atomic_inc_not_zero(&s->s_ref)) {
321 		dout("mdsc get_session %p %d -> %d\n", s,
322 		     atomic_read(&s->s_ref)-1, atomic_read(&s->s_ref));
323 		return s;
324 	} else {
325 		dout("mdsc get_session %p 0 -- FAIL", s);
326 		return NULL;
327 	}
328 }
329 
330 void ceph_put_mds_session(struct ceph_mds_session *s)
331 {
332 	dout("mdsc put_session %p %d -> %d\n", s,
333 	     atomic_read(&s->s_ref), atomic_read(&s->s_ref)-1);
334 	if (atomic_dec_and_test(&s->s_ref)) {
335 		if (s->s_authorizer)
336 		     s->s_mdsc->fsc->client->monc.auth->ops->destroy_authorizer(
337 			     s->s_mdsc->fsc->client->monc.auth,
338 			     s->s_authorizer);
339 		kfree(s);
340 	}
341 }
342 
343 /*
344  * called under mdsc->mutex
345  */
346 struct ceph_mds_session *__ceph_lookup_mds_session(struct ceph_mds_client *mdsc,
347 						   int mds)
348 {
349 	struct ceph_mds_session *session;
350 
351 	if (mds >= mdsc->max_sessions || mdsc->sessions[mds] == NULL)
352 		return NULL;
353 	session = mdsc->sessions[mds];
354 	dout("lookup_mds_session %p %d\n", session,
355 	     atomic_read(&session->s_ref));
356 	get_session(session);
357 	return session;
358 }
359 
360 static bool __have_session(struct ceph_mds_client *mdsc, int mds)
361 {
362 	if (mds >= mdsc->max_sessions)
363 		return false;
364 	return mdsc->sessions[mds];
365 }
366 
367 static int __verify_registered_session(struct ceph_mds_client *mdsc,
368 				       struct ceph_mds_session *s)
369 {
370 	if (s->s_mds >= mdsc->max_sessions ||
371 	    mdsc->sessions[s->s_mds] != s)
372 		return -ENOENT;
373 	return 0;
374 }
375 
376 /*
377  * create+register a new session for given mds.
378  * called under mdsc->mutex.
379  */
380 static struct ceph_mds_session *register_session(struct ceph_mds_client *mdsc,
381 						 int mds)
382 {
383 	struct ceph_mds_session *s;
384 
385 	s = kzalloc(sizeof(*s), GFP_NOFS);
386 	if (!s)
387 		return ERR_PTR(-ENOMEM);
388 	s->s_mdsc = mdsc;
389 	s->s_mds = mds;
390 	s->s_state = CEPH_MDS_SESSION_NEW;
391 	s->s_ttl = 0;
392 	s->s_seq = 0;
393 	mutex_init(&s->s_mutex);
394 
395 	ceph_con_init(mdsc->fsc->client->msgr, &s->s_con);
396 	s->s_con.private = s;
397 	s->s_con.ops = &mds_con_ops;
398 	s->s_con.peer_name.type = CEPH_ENTITY_TYPE_MDS;
399 	s->s_con.peer_name.num = cpu_to_le64(mds);
400 
401 	spin_lock_init(&s->s_cap_lock);
402 	s->s_cap_gen = 0;
403 	s->s_cap_ttl = 0;
404 	s->s_renew_requested = 0;
405 	s->s_renew_seq = 0;
406 	INIT_LIST_HEAD(&s->s_caps);
407 	s->s_nr_caps = 0;
408 	s->s_trim_caps = 0;
409 	atomic_set(&s->s_ref, 1);
410 	INIT_LIST_HEAD(&s->s_waiting);
411 	INIT_LIST_HEAD(&s->s_unsafe);
412 	s->s_num_cap_releases = 0;
413 	s->s_cap_iterator = NULL;
414 	INIT_LIST_HEAD(&s->s_cap_releases);
415 	INIT_LIST_HEAD(&s->s_cap_releases_done);
416 	INIT_LIST_HEAD(&s->s_cap_flushing);
417 	INIT_LIST_HEAD(&s->s_cap_snaps_flushing);
418 
419 	dout("register_session mds%d\n", mds);
420 	if (mds >= mdsc->max_sessions) {
421 		int newmax = 1 << get_count_order(mds+1);
422 		struct ceph_mds_session **sa;
423 
424 		dout("register_session realloc to %d\n", newmax);
425 		sa = kcalloc(newmax, sizeof(void *), GFP_NOFS);
426 		if (sa == NULL)
427 			goto fail_realloc;
428 		if (mdsc->sessions) {
429 			memcpy(sa, mdsc->sessions,
430 			       mdsc->max_sessions * sizeof(void *));
431 			kfree(mdsc->sessions);
432 		}
433 		mdsc->sessions = sa;
434 		mdsc->max_sessions = newmax;
435 	}
436 	mdsc->sessions[mds] = s;
437 	atomic_inc(&s->s_ref);  /* one ref to sessions[], one to caller */
438 
439 	ceph_con_open(&s->s_con, ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
440 
441 	return s;
442 
443 fail_realloc:
444 	kfree(s);
445 	return ERR_PTR(-ENOMEM);
446 }
447 
448 /*
449  * called under mdsc->mutex
450  */
451 static void __unregister_session(struct ceph_mds_client *mdsc,
452 			       struct ceph_mds_session *s)
453 {
454 	dout("__unregister_session mds%d %p\n", s->s_mds, s);
455 	BUG_ON(mdsc->sessions[s->s_mds] != s);
456 	mdsc->sessions[s->s_mds] = NULL;
457 	ceph_con_close(&s->s_con);
458 	ceph_put_mds_session(s);
459 }
460 
461 /*
462  * drop session refs in request.
463  *
464  * should be last request ref, or hold mdsc->mutex
465  */
466 static void put_request_session(struct ceph_mds_request *req)
467 {
468 	if (req->r_session) {
469 		ceph_put_mds_session(req->r_session);
470 		req->r_session = NULL;
471 	}
472 }
473 
474 void ceph_mdsc_release_request(struct kref *kref)
475 {
476 	struct ceph_mds_request *req = container_of(kref,
477 						    struct ceph_mds_request,
478 						    r_kref);
479 	if (req->r_request)
480 		ceph_msg_put(req->r_request);
481 	if (req->r_reply) {
482 		ceph_msg_put(req->r_reply);
483 		destroy_reply_info(&req->r_reply_info);
484 	}
485 	if (req->r_inode) {
486 		ceph_put_cap_refs(ceph_inode(req->r_inode),
487 				  CEPH_CAP_PIN);
488 		iput(req->r_inode);
489 	}
490 	if (req->r_locked_dir)
491 		ceph_put_cap_refs(ceph_inode(req->r_locked_dir),
492 				  CEPH_CAP_PIN);
493 	if (req->r_target_inode)
494 		iput(req->r_target_inode);
495 	if (req->r_dentry)
496 		dput(req->r_dentry);
497 	if (req->r_old_dentry) {
498 		ceph_put_cap_refs(
499 			ceph_inode(req->r_old_dentry->d_parent->d_inode),
500 			CEPH_CAP_PIN);
501 		dput(req->r_old_dentry);
502 	}
503 	kfree(req->r_path1);
504 	kfree(req->r_path2);
505 	put_request_session(req);
506 	ceph_unreserve_caps(req->r_mdsc, &req->r_caps_reservation);
507 	kfree(req);
508 }
509 
510 /*
511  * lookup session, bump ref if found.
512  *
513  * called under mdsc->mutex.
514  */
515 static struct ceph_mds_request *__lookup_request(struct ceph_mds_client *mdsc,
516 					     u64 tid)
517 {
518 	struct ceph_mds_request *req;
519 	struct rb_node *n = mdsc->request_tree.rb_node;
520 
521 	while (n) {
522 		req = rb_entry(n, struct ceph_mds_request, r_node);
523 		if (tid < req->r_tid)
524 			n = n->rb_left;
525 		else if (tid > req->r_tid)
526 			n = n->rb_right;
527 		else {
528 			ceph_mdsc_get_request(req);
529 			return req;
530 		}
531 	}
532 	return NULL;
533 }
534 
535 static void __insert_request(struct ceph_mds_client *mdsc,
536 			     struct ceph_mds_request *new)
537 {
538 	struct rb_node **p = &mdsc->request_tree.rb_node;
539 	struct rb_node *parent = NULL;
540 	struct ceph_mds_request *req = NULL;
541 
542 	while (*p) {
543 		parent = *p;
544 		req = rb_entry(parent, struct ceph_mds_request, r_node);
545 		if (new->r_tid < req->r_tid)
546 			p = &(*p)->rb_left;
547 		else if (new->r_tid > req->r_tid)
548 			p = &(*p)->rb_right;
549 		else
550 			BUG();
551 	}
552 
553 	rb_link_node(&new->r_node, parent, p);
554 	rb_insert_color(&new->r_node, &mdsc->request_tree);
555 }
556 
557 /*
558  * Register an in-flight request, and assign a tid.  Link to directory
559  * are modifying (if any).
560  *
561  * Called under mdsc->mutex.
562  */
563 static void __register_request(struct ceph_mds_client *mdsc,
564 			       struct ceph_mds_request *req,
565 			       struct inode *dir)
566 {
567 	req->r_tid = ++mdsc->last_tid;
568 	if (req->r_num_caps)
569 		ceph_reserve_caps(mdsc, &req->r_caps_reservation,
570 				  req->r_num_caps);
571 	dout("__register_request %p tid %lld\n", req, req->r_tid);
572 	ceph_mdsc_get_request(req);
573 	__insert_request(mdsc, req);
574 
575 	req->r_uid = current_fsuid();
576 	req->r_gid = current_fsgid();
577 
578 	if (dir) {
579 		struct ceph_inode_info *ci = ceph_inode(dir);
580 
581 		ihold(dir);
582 		spin_lock(&ci->i_unsafe_lock);
583 		req->r_unsafe_dir = dir;
584 		list_add_tail(&req->r_unsafe_dir_item, &ci->i_unsafe_dirops);
585 		spin_unlock(&ci->i_unsafe_lock);
586 	}
587 }
588 
589 static void __unregister_request(struct ceph_mds_client *mdsc,
590 				 struct ceph_mds_request *req)
591 {
592 	dout("__unregister_request %p tid %lld\n", req, req->r_tid);
593 	rb_erase(&req->r_node, &mdsc->request_tree);
594 	RB_CLEAR_NODE(&req->r_node);
595 
596 	if (req->r_unsafe_dir) {
597 		struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir);
598 
599 		spin_lock(&ci->i_unsafe_lock);
600 		list_del_init(&req->r_unsafe_dir_item);
601 		spin_unlock(&ci->i_unsafe_lock);
602 
603 		iput(req->r_unsafe_dir);
604 		req->r_unsafe_dir = NULL;
605 	}
606 
607 	ceph_mdsc_put_request(req);
608 }
609 
610 /*
611  * Choose mds to send request to next.  If there is a hint set in the
612  * request (e.g., due to a prior forward hint from the mds), use that.
613  * Otherwise, consult frag tree and/or caps to identify the
614  * appropriate mds.  If all else fails, choose randomly.
615  *
616  * Called under mdsc->mutex.
617  */
618 struct dentry *get_nonsnap_parent(struct dentry *dentry)
619 {
620 	while (!IS_ROOT(dentry) && ceph_snap(dentry->d_inode) != CEPH_NOSNAP)
621 		dentry = dentry->d_parent;
622 	return dentry;
623 }
624 
625 static int __choose_mds(struct ceph_mds_client *mdsc,
626 			struct ceph_mds_request *req)
627 {
628 	struct inode *inode;
629 	struct ceph_inode_info *ci;
630 	struct ceph_cap *cap;
631 	int mode = req->r_direct_mode;
632 	int mds = -1;
633 	u32 hash = req->r_direct_hash;
634 	bool is_hash = req->r_direct_is_hash;
635 
636 	/*
637 	 * is there a specific mds we should try?  ignore hint if we have
638 	 * no session and the mds is not up (active or recovering).
639 	 */
640 	if (req->r_resend_mds >= 0 &&
641 	    (__have_session(mdsc, req->r_resend_mds) ||
642 	     ceph_mdsmap_get_state(mdsc->mdsmap, req->r_resend_mds) > 0)) {
643 		dout("choose_mds using resend_mds mds%d\n",
644 		     req->r_resend_mds);
645 		return req->r_resend_mds;
646 	}
647 
648 	if (mode == USE_RANDOM_MDS)
649 		goto random;
650 
651 	inode = NULL;
652 	if (req->r_inode) {
653 		inode = req->r_inode;
654 	} else if (req->r_dentry) {
655 		struct inode *dir = req->r_dentry->d_parent->d_inode;
656 
657 		if (dir->i_sb != mdsc->fsc->sb) {
658 			/* not this fs! */
659 			inode = req->r_dentry->d_inode;
660 		} else if (ceph_snap(dir) != CEPH_NOSNAP) {
661 			/* direct snapped/virtual snapdir requests
662 			 * based on parent dir inode */
663 			struct dentry *dn =
664 				get_nonsnap_parent(req->r_dentry->d_parent);
665 			inode = dn->d_inode;
666 			dout("__choose_mds using nonsnap parent %p\n", inode);
667 		} else if (req->r_dentry->d_inode) {
668 			/* dentry target */
669 			inode = req->r_dentry->d_inode;
670 		} else {
671 			/* dir + name */
672 			inode = dir;
673 			hash = ceph_dentry_hash(req->r_dentry);
674 			is_hash = true;
675 		}
676 	}
677 
678 	dout("__choose_mds %p is_hash=%d (%d) mode %d\n", inode, (int)is_hash,
679 	     (int)hash, mode);
680 	if (!inode)
681 		goto random;
682 	ci = ceph_inode(inode);
683 
684 	if (is_hash && S_ISDIR(inode->i_mode)) {
685 		struct ceph_inode_frag frag;
686 		int found;
687 
688 		ceph_choose_frag(ci, hash, &frag, &found);
689 		if (found) {
690 			if (mode == USE_ANY_MDS && frag.ndist > 0) {
691 				u8 r;
692 
693 				/* choose a random replica */
694 				get_random_bytes(&r, 1);
695 				r %= frag.ndist;
696 				mds = frag.dist[r];
697 				dout("choose_mds %p %llx.%llx "
698 				     "frag %u mds%d (%d/%d)\n",
699 				     inode, ceph_vinop(inode),
700 				     frag.frag, mds,
701 				     (int)r, frag.ndist);
702 				if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
703 				    CEPH_MDS_STATE_ACTIVE)
704 					return mds;
705 			}
706 
707 			/* since this file/dir wasn't known to be
708 			 * replicated, then we want to look for the
709 			 * authoritative mds. */
710 			mode = USE_AUTH_MDS;
711 			if (frag.mds >= 0) {
712 				/* choose auth mds */
713 				mds = frag.mds;
714 				dout("choose_mds %p %llx.%llx "
715 				     "frag %u mds%d (auth)\n",
716 				     inode, ceph_vinop(inode), frag.frag, mds);
717 				if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
718 				    CEPH_MDS_STATE_ACTIVE)
719 					return mds;
720 			}
721 		}
722 	}
723 
724 	spin_lock(&inode->i_lock);
725 	cap = NULL;
726 	if (mode == USE_AUTH_MDS)
727 		cap = ci->i_auth_cap;
728 	if (!cap && !RB_EMPTY_ROOT(&ci->i_caps))
729 		cap = rb_entry(rb_first(&ci->i_caps), struct ceph_cap, ci_node);
730 	if (!cap) {
731 		spin_unlock(&inode->i_lock);
732 		goto random;
733 	}
734 	mds = cap->session->s_mds;
735 	dout("choose_mds %p %llx.%llx mds%d (%scap %p)\n",
736 	     inode, ceph_vinop(inode), mds,
737 	     cap == ci->i_auth_cap ? "auth " : "", cap);
738 	spin_unlock(&inode->i_lock);
739 	return mds;
740 
741 random:
742 	mds = ceph_mdsmap_get_random_mds(mdsc->mdsmap);
743 	dout("choose_mds chose random mds%d\n", mds);
744 	return mds;
745 }
746 
747 
748 /*
749  * session messages
750  */
751 static struct ceph_msg *create_session_msg(u32 op, u64 seq)
752 {
753 	struct ceph_msg *msg;
754 	struct ceph_mds_session_head *h;
755 
756 	msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h), GFP_NOFS);
757 	if (!msg) {
758 		pr_err("create_session_msg ENOMEM creating msg\n");
759 		return NULL;
760 	}
761 	h = msg->front.iov_base;
762 	h->op = cpu_to_le32(op);
763 	h->seq = cpu_to_le64(seq);
764 	return msg;
765 }
766 
767 /*
768  * send session open request.
769  *
770  * called under mdsc->mutex
771  */
772 static int __open_session(struct ceph_mds_client *mdsc,
773 			  struct ceph_mds_session *session)
774 {
775 	struct ceph_msg *msg;
776 	int mstate;
777 	int mds = session->s_mds;
778 
779 	/* wait for mds to go active? */
780 	mstate = ceph_mdsmap_get_state(mdsc->mdsmap, mds);
781 	dout("open_session to mds%d (%s)\n", mds,
782 	     ceph_mds_state_name(mstate));
783 	session->s_state = CEPH_MDS_SESSION_OPENING;
784 	session->s_renew_requested = jiffies;
785 
786 	/* send connect message */
787 	msg = create_session_msg(CEPH_SESSION_REQUEST_OPEN, session->s_seq);
788 	if (!msg)
789 		return -ENOMEM;
790 	ceph_con_send(&session->s_con, msg);
791 	return 0;
792 }
793 
794 /*
795  * open sessions for any export targets for the given mds
796  *
797  * called under mdsc->mutex
798  */
799 static void __open_export_target_sessions(struct ceph_mds_client *mdsc,
800 					  struct ceph_mds_session *session)
801 {
802 	struct ceph_mds_info *mi;
803 	struct ceph_mds_session *ts;
804 	int i, mds = session->s_mds;
805 	int target;
806 
807 	if (mds >= mdsc->mdsmap->m_max_mds)
808 		return;
809 	mi = &mdsc->mdsmap->m_info[mds];
810 	dout("open_export_target_sessions for mds%d (%d targets)\n",
811 	     session->s_mds, mi->num_export_targets);
812 
813 	for (i = 0; i < mi->num_export_targets; i++) {
814 		target = mi->export_targets[i];
815 		ts = __ceph_lookup_mds_session(mdsc, target);
816 		if (!ts) {
817 			ts = register_session(mdsc, target);
818 			if (IS_ERR(ts))
819 				return;
820 		}
821 		if (session->s_state == CEPH_MDS_SESSION_NEW ||
822 		    session->s_state == CEPH_MDS_SESSION_CLOSING)
823 			__open_session(mdsc, session);
824 		else
825 			dout(" mds%d target mds%d %p is %s\n", session->s_mds,
826 			     i, ts, session_state_name(ts->s_state));
827 		ceph_put_mds_session(ts);
828 	}
829 }
830 
831 void ceph_mdsc_open_export_target_sessions(struct ceph_mds_client *mdsc,
832 					   struct ceph_mds_session *session)
833 {
834 	mutex_lock(&mdsc->mutex);
835 	__open_export_target_sessions(mdsc, session);
836 	mutex_unlock(&mdsc->mutex);
837 }
838 
839 /*
840  * session caps
841  */
842 
843 /*
844  * Free preallocated cap messages assigned to this session
845  */
846 static void cleanup_cap_releases(struct ceph_mds_session *session)
847 {
848 	struct ceph_msg *msg;
849 
850 	spin_lock(&session->s_cap_lock);
851 	while (!list_empty(&session->s_cap_releases)) {
852 		msg = list_first_entry(&session->s_cap_releases,
853 				       struct ceph_msg, list_head);
854 		list_del_init(&msg->list_head);
855 		ceph_msg_put(msg);
856 	}
857 	while (!list_empty(&session->s_cap_releases_done)) {
858 		msg = list_first_entry(&session->s_cap_releases_done,
859 				       struct ceph_msg, list_head);
860 		list_del_init(&msg->list_head);
861 		ceph_msg_put(msg);
862 	}
863 	spin_unlock(&session->s_cap_lock);
864 }
865 
866 /*
867  * Helper to safely iterate over all caps associated with a session, with
868  * special care taken to handle a racing __ceph_remove_cap().
869  *
870  * Caller must hold session s_mutex.
871  */
872 static int iterate_session_caps(struct ceph_mds_session *session,
873 				 int (*cb)(struct inode *, struct ceph_cap *,
874 					    void *), void *arg)
875 {
876 	struct list_head *p;
877 	struct ceph_cap *cap;
878 	struct inode *inode, *last_inode = NULL;
879 	struct ceph_cap *old_cap = NULL;
880 	int ret;
881 
882 	dout("iterate_session_caps %p mds%d\n", session, session->s_mds);
883 	spin_lock(&session->s_cap_lock);
884 	p = session->s_caps.next;
885 	while (p != &session->s_caps) {
886 		cap = list_entry(p, struct ceph_cap, session_caps);
887 		inode = igrab(&cap->ci->vfs_inode);
888 		if (!inode) {
889 			p = p->next;
890 			continue;
891 		}
892 		session->s_cap_iterator = cap;
893 		spin_unlock(&session->s_cap_lock);
894 
895 		if (last_inode) {
896 			iput(last_inode);
897 			last_inode = NULL;
898 		}
899 		if (old_cap) {
900 			ceph_put_cap(session->s_mdsc, old_cap);
901 			old_cap = NULL;
902 		}
903 
904 		ret = cb(inode, cap, arg);
905 		last_inode = inode;
906 
907 		spin_lock(&session->s_cap_lock);
908 		p = p->next;
909 		if (cap->ci == NULL) {
910 			dout("iterate_session_caps  finishing cap %p removal\n",
911 			     cap);
912 			BUG_ON(cap->session != session);
913 			list_del_init(&cap->session_caps);
914 			session->s_nr_caps--;
915 			cap->session = NULL;
916 			old_cap = cap;  /* put_cap it w/o locks held */
917 		}
918 		if (ret < 0)
919 			goto out;
920 	}
921 	ret = 0;
922 out:
923 	session->s_cap_iterator = NULL;
924 	spin_unlock(&session->s_cap_lock);
925 
926 	if (last_inode)
927 		iput(last_inode);
928 	if (old_cap)
929 		ceph_put_cap(session->s_mdsc, old_cap);
930 
931 	return ret;
932 }
933 
934 static int remove_session_caps_cb(struct inode *inode, struct ceph_cap *cap,
935 				  void *arg)
936 {
937 	struct ceph_inode_info *ci = ceph_inode(inode);
938 	int drop = 0;
939 
940 	dout("removing cap %p, ci is %p, inode is %p\n",
941 	     cap, ci, &ci->vfs_inode);
942 	spin_lock(&inode->i_lock);
943 	__ceph_remove_cap(cap);
944 	if (!__ceph_is_any_real_caps(ci)) {
945 		struct ceph_mds_client *mdsc =
946 			ceph_sb_to_client(inode->i_sb)->mdsc;
947 
948 		spin_lock(&mdsc->cap_dirty_lock);
949 		if (!list_empty(&ci->i_dirty_item)) {
950 			pr_info(" dropping dirty %s state for %p %lld\n",
951 				ceph_cap_string(ci->i_dirty_caps),
952 				inode, ceph_ino(inode));
953 			ci->i_dirty_caps = 0;
954 			list_del_init(&ci->i_dirty_item);
955 			drop = 1;
956 		}
957 		if (!list_empty(&ci->i_flushing_item)) {
958 			pr_info(" dropping dirty+flushing %s state for %p %lld\n",
959 				ceph_cap_string(ci->i_flushing_caps),
960 				inode, ceph_ino(inode));
961 			ci->i_flushing_caps = 0;
962 			list_del_init(&ci->i_flushing_item);
963 			mdsc->num_cap_flushing--;
964 			drop = 1;
965 		}
966 		if (drop && ci->i_wrbuffer_ref) {
967 			pr_info(" dropping dirty data for %p %lld\n",
968 				inode, ceph_ino(inode));
969 			ci->i_wrbuffer_ref = 0;
970 			ci->i_wrbuffer_ref_head = 0;
971 			drop++;
972 		}
973 		spin_unlock(&mdsc->cap_dirty_lock);
974 	}
975 	spin_unlock(&inode->i_lock);
976 	while (drop--)
977 		iput(inode);
978 	return 0;
979 }
980 
981 /*
982  * caller must hold session s_mutex
983  */
984 static void remove_session_caps(struct ceph_mds_session *session)
985 {
986 	dout("remove_session_caps on %p\n", session);
987 	iterate_session_caps(session, remove_session_caps_cb, NULL);
988 	BUG_ON(session->s_nr_caps > 0);
989 	BUG_ON(!list_empty(&session->s_cap_flushing));
990 	cleanup_cap_releases(session);
991 }
992 
993 /*
994  * wake up any threads waiting on this session's caps.  if the cap is
995  * old (didn't get renewed on the client reconnect), remove it now.
996  *
997  * caller must hold s_mutex.
998  */
999 static int wake_up_session_cb(struct inode *inode, struct ceph_cap *cap,
1000 			      void *arg)
1001 {
1002 	struct ceph_inode_info *ci = ceph_inode(inode);
1003 
1004 	wake_up_all(&ci->i_cap_wq);
1005 	if (arg) {
1006 		spin_lock(&inode->i_lock);
1007 		ci->i_wanted_max_size = 0;
1008 		ci->i_requested_max_size = 0;
1009 		spin_unlock(&inode->i_lock);
1010 	}
1011 	return 0;
1012 }
1013 
1014 static void wake_up_session_caps(struct ceph_mds_session *session,
1015 				 int reconnect)
1016 {
1017 	dout("wake_up_session_caps %p mds%d\n", session, session->s_mds);
1018 	iterate_session_caps(session, wake_up_session_cb,
1019 			     (void *)(unsigned long)reconnect);
1020 }
1021 
1022 /*
1023  * Send periodic message to MDS renewing all currently held caps.  The
1024  * ack will reset the expiration for all caps from this session.
1025  *
1026  * caller holds s_mutex
1027  */
1028 static int send_renew_caps(struct ceph_mds_client *mdsc,
1029 			   struct ceph_mds_session *session)
1030 {
1031 	struct ceph_msg *msg;
1032 	int state;
1033 
1034 	if (time_after_eq(jiffies, session->s_cap_ttl) &&
1035 	    time_after_eq(session->s_cap_ttl, session->s_renew_requested))
1036 		pr_info("mds%d caps stale\n", session->s_mds);
1037 	session->s_renew_requested = jiffies;
1038 
1039 	/* do not try to renew caps until a recovering mds has reconnected
1040 	 * with its clients. */
1041 	state = ceph_mdsmap_get_state(mdsc->mdsmap, session->s_mds);
1042 	if (state < CEPH_MDS_STATE_RECONNECT) {
1043 		dout("send_renew_caps ignoring mds%d (%s)\n",
1044 		     session->s_mds, ceph_mds_state_name(state));
1045 		return 0;
1046 	}
1047 
1048 	dout("send_renew_caps to mds%d (%s)\n", session->s_mds,
1049 		ceph_mds_state_name(state));
1050 	msg = create_session_msg(CEPH_SESSION_REQUEST_RENEWCAPS,
1051 				 ++session->s_renew_seq);
1052 	if (!msg)
1053 		return -ENOMEM;
1054 	ceph_con_send(&session->s_con, msg);
1055 	return 0;
1056 }
1057 
1058 /*
1059  * Note new cap ttl, and any transition from stale -> not stale (fresh?).
1060  *
1061  * Called under session->s_mutex
1062  */
1063 static void renewed_caps(struct ceph_mds_client *mdsc,
1064 			 struct ceph_mds_session *session, int is_renew)
1065 {
1066 	int was_stale;
1067 	int wake = 0;
1068 
1069 	spin_lock(&session->s_cap_lock);
1070 	was_stale = is_renew && (session->s_cap_ttl == 0 ||
1071 				 time_after_eq(jiffies, session->s_cap_ttl));
1072 
1073 	session->s_cap_ttl = session->s_renew_requested +
1074 		mdsc->mdsmap->m_session_timeout*HZ;
1075 
1076 	if (was_stale) {
1077 		if (time_before(jiffies, session->s_cap_ttl)) {
1078 			pr_info("mds%d caps renewed\n", session->s_mds);
1079 			wake = 1;
1080 		} else {
1081 			pr_info("mds%d caps still stale\n", session->s_mds);
1082 		}
1083 	}
1084 	dout("renewed_caps mds%d ttl now %lu, was %s, now %s\n",
1085 	     session->s_mds, session->s_cap_ttl, was_stale ? "stale" : "fresh",
1086 	     time_before(jiffies, session->s_cap_ttl) ? "stale" : "fresh");
1087 	spin_unlock(&session->s_cap_lock);
1088 
1089 	if (wake)
1090 		wake_up_session_caps(session, 0);
1091 }
1092 
1093 /*
1094  * send a session close request
1095  */
1096 static int request_close_session(struct ceph_mds_client *mdsc,
1097 				 struct ceph_mds_session *session)
1098 {
1099 	struct ceph_msg *msg;
1100 
1101 	dout("request_close_session mds%d state %s seq %lld\n",
1102 	     session->s_mds, session_state_name(session->s_state),
1103 	     session->s_seq);
1104 	msg = create_session_msg(CEPH_SESSION_REQUEST_CLOSE, session->s_seq);
1105 	if (!msg)
1106 		return -ENOMEM;
1107 	ceph_con_send(&session->s_con, msg);
1108 	return 0;
1109 }
1110 
1111 /*
1112  * Called with s_mutex held.
1113  */
1114 static int __close_session(struct ceph_mds_client *mdsc,
1115 			 struct ceph_mds_session *session)
1116 {
1117 	if (session->s_state >= CEPH_MDS_SESSION_CLOSING)
1118 		return 0;
1119 	session->s_state = CEPH_MDS_SESSION_CLOSING;
1120 	return request_close_session(mdsc, session);
1121 }
1122 
1123 /*
1124  * Trim old(er) caps.
1125  *
1126  * Because we can't cache an inode without one or more caps, we do
1127  * this indirectly: if a cap is unused, we prune its aliases, at which
1128  * point the inode will hopefully get dropped to.
1129  *
1130  * Yes, this is a bit sloppy.  Our only real goal here is to respond to
1131  * memory pressure from the MDS, though, so it needn't be perfect.
1132  */
1133 static int trim_caps_cb(struct inode *inode, struct ceph_cap *cap, void *arg)
1134 {
1135 	struct ceph_mds_session *session = arg;
1136 	struct ceph_inode_info *ci = ceph_inode(inode);
1137 	int used, oissued, mine;
1138 
1139 	if (session->s_trim_caps <= 0)
1140 		return -1;
1141 
1142 	spin_lock(&inode->i_lock);
1143 	mine = cap->issued | cap->implemented;
1144 	used = __ceph_caps_used(ci);
1145 	oissued = __ceph_caps_issued_other(ci, cap);
1146 
1147 	dout("trim_caps_cb %p cap %p mine %s oissued %s used %s\n",
1148 	     inode, cap, ceph_cap_string(mine), ceph_cap_string(oissued),
1149 	     ceph_cap_string(used));
1150 	if (ci->i_dirty_caps)
1151 		goto out;   /* dirty caps */
1152 	if ((used & ~oissued) & mine)
1153 		goto out;   /* we need these caps */
1154 
1155 	session->s_trim_caps--;
1156 	if (oissued) {
1157 		/* we aren't the only cap.. just remove us */
1158 		__ceph_remove_cap(cap);
1159 	} else {
1160 		/* try to drop referring dentries */
1161 		spin_unlock(&inode->i_lock);
1162 		d_prune_aliases(inode);
1163 		dout("trim_caps_cb %p cap %p  pruned, count now %d\n",
1164 		     inode, cap, atomic_read(&inode->i_count));
1165 		return 0;
1166 	}
1167 
1168 out:
1169 	spin_unlock(&inode->i_lock);
1170 	return 0;
1171 }
1172 
1173 /*
1174  * Trim session cap count down to some max number.
1175  */
1176 static int trim_caps(struct ceph_mds_client *mdsc,
1177 		     struct ceph_mds_session *session,
1178 		     int max_caps)
1179 {
1180 	int trim_caps = session->s_nr_caps - max_caps;
1181 
1182 	dout("trim_caps mds%d start: %d / %d, trim %d\n",
1183 	     session->s_mds, session->s_nr_caps, max_caps, trim_caps);
1184 	if (trim_caps > 0) {
1185 		session->s_trim_caps = trim_caps;
1186 		iterate_session_caps(session, trim_caps_cb, session);
1187 		dout("trim_caps mds%d done: %d / %d, trimmed %d\n",
1188 		     session->s_mds, session->s_nr_caps, max_caps,
1189 			trim_caps - session->s_trim_caps);
1190 		session->s_trim_caps = 0;
1191 	}
1192 	return 0;
1193 }
1194 
1195 /*
1196  * Allocate cap_release messages.  If there is a partially full message
1197  * in the queue, try to allocate enough to cover it's remainder, so that
1198  * we can send it immediately.
1199  *
1200  * Called under s_mutex.
1201  */
1202 int ceph_add_cap_releases(struct ceph_mds_client *mdsc,
1203 			  struct ceph_mds_session *session)
1204 {
1205 	struct ceph_msg *msg, *partial = NULL;
1206 	struct ceph_mds_cap_release *head;
1207 	int err = -ENOMEM;
1208 	int extra = mdsc->fsc->mount_options->cap_release_safety;
1209 	int num;
1210 
1211 	dout("add_cap_releases %p mds%d extra %d\n", session, session->s_mds,
1212 	     extra);
1213 
1214 	spin_lock(&session->s_cap_lock);
1215 
1216 	if (!list_empty(&session->s_cap_releases)) {
1217 		msg = list_first_entry(&session->s_cap_releases,
1218 				       struct ceph_msg,
1219 				 list_head);
1220 		head = msg->front.iov_base;
1221 		num = le32_to_cpu(head->num);
1222 		if (num) {
1223 			dout(" partial %p with (%d/%d)\n", msg, num,
1224 			     (int)CEPH_CAPS_PER_RELEASE);
1225 			extra += CEPH_CAPS_PER_RELEASE - num;
1226 			partial = msg;
1227 		}
1228 	}
1229 	while (session->s_num_cap_releases < session->s_nr_caps + extra) {
1230 		spin_unlock(&session->s_cap_lock);
1231 		msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPRELEASE, PAGE_CACHE_SIZE,
1232 				   GFP_NOFS);
1233 		if (!msg)
1234 			goto out_unlocked;
1235 		dout("add_cap_releases %p msg %p now %d\n", session, msg,
1236 		     (int)msg->front.iov_len);
1237 		head = msg->front.iov_base;
1238 		head->num = cpu_to_le32(0);
1239 		msg->front.iov_len = sizeof(*head);
1240 		spin_lock(&session->s_cap_lock);
1241 		list_add(&msg->list_head, &session->s_cap_releases);
1242 		session->s_num_cap_releases += CEPH_CAPS_PER_RELEASE;
1243 	}
1244 
1245 	if (partial) {
1246 		head = partial->front.iov_base;
1247 		num = le32_to_cpu(head->num);
1248 		dout(" queueing partial %p with %d/%d\n", partial, num,
1249 		     (int)CEPH_CAPS_PER_RELEASE);
1250 		list_move_tail(&partial->list_head,
1251 			       &session->s_cap_releases_done);
1252 		session->s_num_cap_releases -= CEPH_CAPS_PER_RELEASE - num;
1253 	}
1254 	err = 0;
1255 	spin_unlock(&session->s_cap_lock);
1256 out_unlocked:
1257 	return err;
1258 }
1259 
1260 /*
1261  * flush all dirty inode data to disk.
1262  *
1263  * returns true if we've flushed through want_flush_seq
1264  */
1265 static int check_cap_flush(struct ceph_mds_client *mdsc, u64 want_flush_seq)
1266 {
1267 	int mds, ret = 1;
1268 
1269 	dout("check_cap_flush want %lld\n", want_flush_seq);
1270 	mutex_lock(&mdsc->mutex);
1271 	for (mds = 0; ret && mds < mdsc->max_sessions; mds++) {
1272 		struct ceph_mds_session *session = mdsc->sessions[mds];
1273 
1274 		if (!session)
1275 			continue;
1276 		get_session(session);
1277 		mutex_unlock(&mdsc->mutex);
1278 
1279 		mutex_lock(&session->s_mutex);
1280 		if (!list_empty(&session->s_cap_flushing)) {
1281 			struct ceph_inode_info *ci =
1282 				list_entry(session->s_cap_flushing.next,
1283 					   struct ceph_inode_info,
1284 					   i_flushing_item);
1285 			struct inode *inode = &ci->vfs_inode;
1286 
1287 			spin_lock(&inode->i_lock);
1288 			if (ci->i_cap_flush_seq <= want_flush_seq) {
1289 				dout("check_cap_flush still flushing %p "
1290 				     "seq %lld <= %lld to mds%d\n", inode,
1291 				     ci->i_cap_flush_seq, want_flush_seq,
1292 				     session->s_mds);
1293 				ret = 0;
1294 			}
1295 			spin_unlock(&inode->i_lock);
1296 		}
1297 		mutex_unlock(&session->s_mutex);
1298 		ceph_put_mds_session(session);
1299 
1300 		if (!ret)
1301 			return ret;
1302 		mutex_lock(&mdsc->mutex);
1303 	}
1304 
1305 	mutex_unlock(&mdsc->mutex);
1306 	dout("check_cap_flush ok, flushed thru %lld\n", want_flush_seq);
1307 	return ret;
1308 }
1309 
1310 /*
1311  * called under s_mutex
1312  */
1313 void ceph_send_cap_releases(struct ceph_mds_client *mdsc,
1314 			    struct ceph_mds_session *session)
1315 {
1316 	struct ceph_msg *msg;
1317 
1318 	dout("send_cap_releases mds%d\n", session->s_mds);
1319 	spin_lock(&session->s_cap_lock);
1320 	while (!list_empty(&session->s_cap_releases_done)) {
1321 		msg = list_first_entry(&session->s_cap_releases_done,
1322 				 struct ceph_msg, list_head);
1323 		list_del_init(&msg->list_head);
1324 		spin_unlock(&session->s_cap_lock);
1325 		msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
1326 		dout("send_cap_releases mds%d %p\n", session->s_mds, msg);
1327 		ceph_con_send(&session->s_con, msg);
1328 		spin_lock(&session->s_cap_lock);
1329 	}
1330 	spin_unlock(&session->s_cap_lock);
1331 }
1332 
1333 static void discard_cap_releases(struct ceph_mds_client *mdsc,
1334 				 struct ceph_mds_session *session)
1335 {
1336 	struct ceph_msg *msg;
1337 	struct ceph_mds_cap_release *head;
1338 	unsigned num;
1339 
1340 	dout("discard_cap_releases mds%d\n", session->s_mds);
1341 	spin_lock(&session->s_cap_lock);
1342 
1343 	/* zero out the in-progress message */
1344 	msg = list_first_entry(&session->s_cap_releases,
1345 			       struct ceph_msg, list_head);
1346 	head = msg->front.iov_base;
1347 	num = le32_to_cpu(head->num);
1348 	dout("discard_cap_releases mds%d %p %u\n", session->s_mds, msg, num);
1349 	head->num = cpu_to_le32(0);
1350 	session->s_num_cap_releases += num;
1351 
1352 	/* requeue completed messages */
1353 	while (!list_empty(&session->s_cap_releases_done)) {
1354 		msg = list_first_entry(&session->s_cap_releases_done,
1355 				 struct ceph_msg, list_head);
1356 		list_del_init(&msg->list_head);
1357 
1358 		head = msg->front.iov_base;
1359 		num = le32_to_cpu(head->num);
1360 		dout("discard_cap_releases mds%d %p %u\n", session->s_mds, msg,
1361 		     num);
1362 		session->s_num_cap_releases += num;
1363 		head->num = cpu_to_le32(0);
1364 		msg->front.iov_len = sizeof(*head);
1365 		list_add(&msg->list_head, &session->s_cap_releases);
1366 	}
1367 
1368 	spin_unlock(&session->s_cap_lock);
1369 }
1370 
1371 /*
1372  * requests
1373  */
1374 
1375 /*
1376  * Create an mds request.
1377  */
1378 struct ceph_mds_request *
1379 ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode)
1380 {
1381 	struct ceph_mds_request *req = kzalloc(sizeof(*req), GFP_NOFS);
1382 
1383 	if (!req)
1384 		return ERR_PTR(-ENOMEM);
1385 
1386 	mutex_init(&req->r_fill_mutex);
1387 	req->r_mdsc = mdsc;
1388 	req->r_started = jiffies;
1389 	req->r_resend_mds = -1;
1390 	INIT_LIST_HEAD(&req->r_unsafe_dir_item);
1391 	req->r_fmode = -1;
1392 	kref_init(&req->r_kref);
1393 	INIT_LIST_HEAD(&req->r_wait);
1394 	init_completion(&req->r_completion);
1395 	init_completion(&req->r_safe_completion);
1396 	INIT_LIST_HEAD(&req->r_unsafe_item);
1397 
1398 	req->r_op = op;
1399 	req->r_direct_mode = mode;
1400 	return req;
1401 }
1402 
1403 /*
1404  * return oldest (lowest) request, tid in request tree, 0 if none.
1405  *
1406  * called under mdsc->mutex.
1407  */
1408 static struct ceph_mds_request *__get_oldest_req(struct ceph_mds_client *mdsc)
1409 {
1410 	if (RB_EMPTY_ROOT(&mdsc->request_tree))
1411 		return NULL;
1412 	return rb_entry(rb_first(&mdsc->request_tree),
1413 			struct ceph_mds_request, r_node);
1414 }
1415 
1416 static u64 __get_oldest_tid(struct ceph_mds_client *mdsc)
1417 {
1418 	struct ceph_mds_request *req = __get_oldest_req(mdsc);
1419 
1420 	if (req)
1421 		return req->r_tid;
1422 	return 0;
1423 }
1424 
1425 /*
1426  * Build a dentry's path.  Allocate on heap; caller must kfree.  Based
1427  * on build_path_from_dentry in fs/cifs/dir.c.
1428  *
1429  * If @stop_on_nosnap, generate path relative to the first non-snapped
1430  * inode.
1431  *
1432  * Encode hidden .snap dirs as a double /, i.e.
1433  *   foo/.snap/bar -> foo//bar
1434  */
1435 char *ceph_mdsc_build_path(struct dentry *dentry, int *plen, u64 *base,
1436 			   int stop_on_nosnap)
1437 {
1438 	struct dentry *temp;
1439 	char *path;
1440 	int len, pos;
1441 	unsigned seq;
1442 
1443 	if (dentry == NULL)
1444 		return ERR_PTR(-EINVAL);
1445 
1446 retry:
1447 	len = 0;
1448 	seq = read_seqbegin(&rename_lock);
1449 	rcu_read_lock();
1450 	for (temp = dentry; !IS_ROOT(temp);) {
1451 		struct inode *inode = temp->d_inode;
1452 		if (inode && ceph_snap(inode) == CEPH_SNAPDIR)
1453 			len++;  /* slash only */
1454 		else if (stop_on_nosnap && inode &&
1455 			 ceph_snap(inode) == CEPH_NOSNAP)
1456 			break;
1457 		else
1458 			len += 1 + temp->d_name.len;
1459 		temp = temp->d_parent;
1460 		if (temp == NULL) {
1461 			rcu_read_unlock();
1462 			pr_err("build_path corrupt dentry %p\n", dentry);
1463 			return ERR_PTR(-EINVAL);
1464 		}
1465 	}
1466 	rcu_read_unlock();
1467 	if (len)
1468 		len--;  /* no leading '/' */
1469 
1470 	path = kmalloc(len+1, GFP_NOFS);
1471 	if (path == NULL)
1472 		return ERR_PTR(-ENOMEM);
1473 	pos = len;
1474 	path[pos] = 0;	/* trailing null */
1475 	rcu_read_lock();
1476 	for (temp = dentry; !IS_ROOT(temp) && pos != 0; ) {
1477 		struct inode *inode;
1478 
1479 		spin_lock(&temp->d_lock);
1480 		inode = temp->d_inode;
1481 		if (inode && ceph_snap(inode) == CEPH_SNAPDIR) {
1482 			dout("build_path path+%d: %p SNAPDIR\n",
1483 			     pos, temp);
1484 		} else if (stop_on_nosnap && inode &&
1485 			   ceph_snap(inode) == CEPH_NOSNAP) {
1486 			break;
1487 		} else {
1488 			pos -= temp->d_name.len;
1489 			if (pos < 0) {
1490 				spin_unlock(&temp->d_lock);
1491 				break;
1492 			}
1493 			strncpy(path + pos, temp->d_name.name,
1494 				temp->d_name.len);
1495 		}
1496 		spin_unlock(&temp->d_lock);
1497 		if (pos)
1498 			path[--pos] = '/';
1499 		temp = temp->d_parent;
1500 		if (temp == NULL) {
1501 			rcu_read_unlock();
1502 			pr_err("build_path corrupt dentry\n");
1503 			kfree(path);
1504 			return ERR_PTR(-EINVAL);
1505 		}
1506 	}
1507 	rcu_read_unlock();
1508 	if (pos != 0 || read_seqretry(&rename_lock, seq)) {
1509 		pr_err("build_path did not end path lookup where "
1510 		       "expected, namelen is %d, pos is %d\n", len, pos);
1511 		/* presumably this is only possible if racing with a
1512 		   rename of one of the parent directories (we can not
1513 		   lock the dentries above us to prevent this, but
1514 		   retrying should be harmless) */
1515 		kfree(path);
1516 		goto retry;
1517 	}
1518 
1519 	*base = ceph_ino(temp->d_inode);
1520 	*plen = len;
1521 	dout("build_path on %p %d built %llx '%.*s'\n",
1522 	     dentry, dentry->d_count, *base, len, path);
1523 	return path;
1524 }
1525 
1526 static int build_dentry_path(struct dentry *dentry,
1527 			     const char **ppath, int *ppathlen, u64 *pino,
1528 			     int *pfreepath)
1529 {
1530 	char *path;
1531 
1532 	if (ceph_snap(dentry->d_parent->d_inode) == CEPH_NOSNAP) {
1533 		*pino = ceph_ino(dentry->d_parent->d_inode);
1534 		*ppath = dentry->d_name.name;
1535 		*ppathlen = dentry->d_name.len;
1536 		return 0;
1537 	}
1538 	path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
1539 	if (IS_ERR(path))
1540 		return PTR_ERR(path);
1541 	*ppath = path;
1542 	*pfreepath = 1;
1543 	return 0;
1544 }
1545 
1546 static int build_inode_path(struct inode *inode,
1547 			    const char **ppath, int *ppathlen, u64 *pino,
1548 			    int *pfreepath)
1549 {
1550 	struct dentry *dentry;
1551 	char *path;
1552 
1553 	if (ceph_snap(inode) == CEPH_NOSNAP) {
1554 		*pino = ceph_ino(inode);
1555 		*ppathlen = 0;
1556 		return 0;
1557 	}
1558 	dentry = d_find_alias(inode);
1559 	path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
1560 	dput(dentry);
1561 	if (IS_ERR(path))
1562 		return PTR_ERR(path);
1563 	*ppath = path;
1564 	*pfreepath = 1;
1565 	return 0;
1566 }
1567 
1568 /*
1569  * request arguments may be specified via an inode *, a dentry *, or
1570  * an explicit ino+path.
1571  */
1572 static int set_request_path_attr(struct inode *rinode, struct dentry *rdentry,
1573 				  const char *rpath, u64 rino,
1574 				  const char **ppath, int *pathlen,
1575 				  u64 *ino, int *freepath)
1576 {
1577 	int r = 0;
1578 
1579 	if (rinode) {
1580 		r = build_inode_path(rinode, ppath, pathlen, ino, freepath);
1581 		dout(" inode %p %llx.%llx\n", rinode, ceph_ino(rinode),
1582 		     ceph_snap(rinode));
1583 	} else if (rdentry) {
1584 		r = build_dentry_path(rdentry, ppath, pathlen, ino, freepath);
1585 		dout(" dentry %p %llx/%.*s\n", rdentry, *ino, *pathlen,
1586 		     *ppath);
1587 	} else if (rpath) {
1588 		*ino = rino;
1589 		*ppath = rpath;
1590 		*pathlen = strlen(rpath);
1591 		dout(" path %.*s\n", *pathlen, rpath);
1592 	}
1593 
1594 	return r;
1595 }
1596 
1597 /*
1598  * called under mdsc->mutex
1599  */
1600 static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
1601 					       struct ceph_mds_request *req,
1602 					       int mds)
1603 {
1604 	struct ceph_msg *msg;
1605 	struct ceph_mds_request_head *head;
1606 	const char *path1 = NULL;
1607 	const char *path2 = NULL;
1608 	u64 ino1 = 0, ino2 = 0;
1609 	int pathlen1 = 0, pathlen2 = 0;
1610 	int freepath1 = 0, freepath2 = 0;
1611 	int len;
1612 	u16 releases;
1613 	void *p, *end;
1614 	int ret;
1615 
1616 	ret = set_request_path_attr(req->r_inode, req->r_dentry,
1617 			      req->r_path1, req->r_ino1.ino,
1618 			      &path1, &pathlen1, &ino1, &freepath1);
1619 	if (ret < 0) {
1620 		msg = ERR_PTR(ret);
1621 		goto out;
1622 	}
1623 
1624 	ret = set_request_path_attr(NULL, req->r_old_dentry,
1625 			      req->r_path2, req->r_ino2.ino,
1626 			      &path2, &pathlen2, &ino2, &freepath2);
1627 	if (ret < 0) {
1628 		msg = ERR_PTR(ret);
1629 		goto out_free1;
1630 	}
1631 
1632 	len = sizeof(*head) +
1633 		pathlen1 + pathlen2 + 2*(1 + sizeof(u32) + sizeof(u64));
1634 
1635 	/* calculate (max) length for cap releases */
1636 	len += sizeof(struct ceph_mds_request_release) *
1637 		(!!req->r_inode_drop + !!req->r_dentry_drop +
1638 		 !!req->r_old_inode_drop + !!req->r_old_dentry_drop);
1639 	if (req->r_dentry_drop)
1640 		len += req->r_dentry->d_name.len;
1641 	if (req->r_old_dentry_drop)
1642 		len += req->r_old_dentry->d_name.len;
1643 
1644 	msg = ceph_msg_new(CEPH_MSG_CLIENT_REQUEST, len, GFP_NOFS);
1645 	if (!msg) {
1646 		msg = ERR_PTR(-ENOMEM);
1647 		goto out_free2;
1648 	}
1649 
1650 	msg->hdr.tid = cpu_to_le64(req->r_tid);
1651 
1652 	head = msg->front.iov_base;
1653 	p = msg->front.iov_base + sizeof(*head);
1654 	end = msg->front.iov_base + msg->front.iov_len;
1655 
1656 	head->mdsmap_epoch = cpu_to_le32(mdsc->mdsmap->m_epoch);
1657 	head->op = cpu_to_le32(req->r_op);
1658 	head->caller_uid = cpu_to_le32(req->r_uid);
1659 	head->caller_gid = cpu_to_le32(req->r_gid);
1660 	head->args = req->r_args;
1661 
1662 	ceph_encode_filepath(&p, end, ino1, path1);
1663 	ceph_encode_filepath(&p, end, ino2, path2);
1664 
1665 	/* make note of release offset, in case we need to replay */
1666 	req->r_request_release_offset = p - msg->front.iov_base;
1667 
1668 	/* cap releases */
1669 	releases = 0;
1670 	if (req->r_inode_drop)
1671 		releases += ceph_encode_inode_release(&p,
1672 		      req->r_inode ? req->r_inode : req->r_dentry->d_inode,
1673 		      mds, req->r_inode_drop, req->r_inode_unless, 0);
1674 	if (req->r_dentry_drop)
1675 		releases += ceph_encode_dentry_release(&p, req->r_dentry,
1676 		       mds, req->r_dentry_drop, req->r_dentry_unless);
1677 	if (req->r_old_dentry_drop)
1678 		releases += ceph_encode_dentry_release(&p, req->r_old_dentry,
1679 		       mds, req->r_old_dentry_drop, req->r_old_dentry_unless);
1680 	if (req->r_old_inode_drop)
1681 		releases += ceph_encode_inode_release(&p,
1682 		      req->r_old_dentry->d_inode,
1683 		      mds, req->r_old_inode_drop, req->r_old_inode_unless, 0);
1684 	head->num_releases = cpu_to_le16(releases);
1685 
1686 	BUG_ON(p > end);
1687 	msg->front.iov_len = p - msg->front.iov_base;
1688 	msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
1689 
1690 	msg->pages = req->r_pages;
1691 	msg->nr_pages = req->r_num_pages;
1692 	msg->hdr.data_len = cpu_to_le32(req->r_data_len);
1693 	msg->hdr.data_off = cpu_to_le16(0);
1694 
1695 out_free2:
1696 	if (freepath2)
1697 		kfree((char *)path2);
1698 out_free1:
1699 	if (freepath1)
1700 		kfree((char *)path1);
1701 out:
1702 	return msg;
1703 }
1704 
1705 /*
1706  * called under mdsc->mutex if error, under no mutex if
1707  * success.
1708  */
1709 static void complete_request(struct ceph_mds_client *mdsc,
1710 			     struct ceph_mds_request *req)
1711 {
1712 	if (req->r_callback)
1713 		req->r_callback(mdsc, req);
1714 	else
1715 		complete_all(&req->r_completion);
1716 }
1717 
1718 /*
1719  * called under mdsc->mutex
1720  */
1721 static int __prepare_send_request(struct ceph_mds_client *mdsc,
1722 				  struct ceph_mds_request *req,
1723 				  int mds)
1724 {
1725 	struct ceph_mds_request_head *rhead;
1726 	struct ceph_msg *msg;
1727 	int flags = 0;
1728 
1729 	req->r_attempts++;
1730 	if (req->r_inode) {
1731 		struct ceph_cap *cap =
1732 			ceph_get_cap_for_mds(ceph_inode(req->r_inode), mds);
1733 
1734 		if (cap)
1735 			req->r_sent_on_mseq = cap->mseq;
1736 		else
1737 			req->r_sent_on_mseq = -1;
1738 	}
1739 	dout("prepare_send_request %p tid %lld %s (attempt %d)\n", req,
1740 	     req->r_tid, ceph_mds_op_name(req->r_op), req->r_attempts);
1741 
1742 	if (req->r_got_unsafe) {
1743 		/*
1744 		 * Replay.  Do not regenerate message (and rebuild
1745 		 * paths, etc.); just use the original message.
1746 		 * Rebuilding paths will break for renames because
1747 		 * d_move mangles the src name.
1748 		 */
1749 		msg = req->r_request;
1750 		rhead = msg->front.iov_base;
1751 
1752 		flags = le32_to_cpu(rhead->flags);
1753 		flags |= CEPH_MDS_FLAG_REPLAY;
1754 		rhead->flags = cpu_to_le32(flags);
1755 
1756 		if (req->r_target_inode)
1757 			rhead->ino = cpu_to_le64(ceph_ino(req->r_target_inode));
1758 
1759 		rhead->num_retry = req->r_attempts - 1;
1760 
1761 		/* remove cap/dentry releases from message */
1762 		rhead->num_releases = 0;
1763 		msg->hdr.front_len = cpu_to_le32(req->r_request_release_offset);
1764 		msg->front.iov_len = req->r_request_release_offset;
1765 		return 0;
1766 	}
1767 
1768 	if (req->r_request) {
1769 		ceph_msg_put(req->r_request);
1770 		req->r_request = NULL;
1771 	}
1772 	msg = create_request_message(mdsc, req, mds);
1773 	if (IS_ERR(msg)) {
1774 		req->r_err = PTR_ERR(msg);
1775 		complete_request(mdsc, req);
1776 		return PTR_ERR(msg);
1777 	}
1778 	req->r_request = msg;
1779 
1780 	rhead = msg->front.iov_base;
1781 	rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
1782 	if (req->r_got_unsafe)
1783 		flags |= CEPH_MDS_FLAG_REPLAY;
1784 	if (req->r_locked_dir)
1785 		flags |= CEPH_MDS_FLAG_WANT_DENTRY;
1786 	rhead->flags = cpu_to_le32(flags);
1787 	rhead->num_fwd = req->r_num_fwd;
1788 	rhead->num_retry = req->r_attempts - 1;
1789 	rhead->ino = 0;
1790 
1791 	dout(" r_locked_dir = %p\n", req->r_locked_dir);
1792 	return 0;
1793 }
1794 
1795 /*
1796  * send request, or put it on the appropriate wait list.
1797  */
1798 static int __do_request(struct ceph_mds_client *mdsc,
1799 			struct ceph_mds_request *req)
1800 {
1801 	struct ceph_mds_session *session = NULL;
1802 	int mds = -1;
1803 	int err = -EAGAIN;
1804 
1805 	if (req->r_err || req->r_got_result)
1806 		goto out;
1807 
1808 	if (req->r_timeout &&
1809 	    time_after_eq(jiffies, req->r_started + req->r_timeout)) {
1810 		dout("do_request timed out\n");
1811 		err = -EIO;
1812 		goto finish;
1813 	}
1814 
1815 	put_request_session(req);
1816 
1817 	mds = __choose_mds(mdsc, req);
1818 	if (mds < 0 ||
1819 	    ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) {
1820 		dout("do_request no mds or not active, waiting for map\n");
1821 		list_add(&req->r_wait, &mdsc->waiting_for_map);
1822 		goto out;
1823 	}
1824 
1825 	/* get, open session */
1826 	session = __ceph_lookup_mds_session(mdsc, mds);
1827 	if (!session) {
1828 		session = register_session(mdsc, mds);
1829 		if (IS_ERR(session)) {
1830 			err = PTR_ERR(session);
1831 			goto finish;
1832 		}
1833 	}
1834 	req->r_session = get_session(session);
1835 
1836 	dout("do_request mds%d session %p state %s\n", mds, session,
1837 	     session_state_name(session->s_state));
1838 	if (session->s_state != CEPH_MDS_SESSION_OPEN &&
1839 	    session->s_state != CEPH_MDS_SESSION_HUNG) {
1840 		if (session->s_state == CEPH_MDS_SESSION_NEW ||
1841 		    session->s_state == CEPH_MDS_SESSION_CLOSING)
1842 			__open_session(mdsc, session);
1843 		list_add(&req->r_wait, &session->s_waiting);
1844 		goto out_session;
1845 	}
1846 
1847 	/* send request */
1848 	req->r_resend_mds = -1;   /* forget any previous mds hint */
1849 
1850 	if (req->r_request_started == 0)   /* note request start time */
1851 		req->r_request_started = jiffies;
1852 
1853 	err = __prepare_send_request(mdsc, req, mds);
1854 	if (!err) {
1855 		ceph_msg_get(req->r_request);
1856 		ceph_con_send(&session->s_con, req->r_request);
1857 	}
1858 
1859 out_session:
1860 	ceph_put_mds_session(session);
1861 out:
1862 	return err;
1863 
1864 finish:
1865 	req->r_err = err;
1866 	complete_request(mdsc, req);
1867 	goto out;
1868 }
1869 
1870 /*
1871  * called under mdsc->mutex
1872  */
1873 static void __wake_requests(struct ceph_mds_client *mdsc,
1874 			    struct list_head *head)
1875 {
1876 	struct ceph_mds_request *req, *nreq;
1877 
1878 	list_for_each_entry_safe(req, nreq, head, r_wait) {
1879 		list_del_init(&req->r_wait);
1880 		__do_request(mdsc, req);
1881 	}
1882 }
1883 
1884 /*
1885  * Wake up threads with requests pending for @mds, so that they can
1886  * resubmit their requests to a possibly different mds.
1887  */
1888 static void kick_requests(struct ceph_mds_client *mdsc, int mds)
1889 {
1890 	struct ceph_mds_request *req;
1891 	struct rb_node *p;
1892 
1893 	dout("kick_requests mds%d\n", mds);
1894 	for (p = rb_first(&mdsc->request_tree); p; p = rb_next(p)) {
1895 		req = rb_entry(p, struct ceph_mds_request, r_node);
1896 		if (req->r_got_unsafe)
1897 			continue;
1898 		if (req->r_session &&
1899 		    req->r_session->s_mds == mds) {
1900 			dout(" kicking tid %llu\n", req->r_tid);
1901 			__do_request(mdsc, req);
1902 		}
1903 	}
1904 }
1905 
1906 void ceph_mdsc_submit_request(struct ceph_mds_client *mdsc,
1907 			      struct ceph_mds_request *req)
1908 {
1909 	dout("submit_request on %p\n", req);
1910 	mutex_lock(&mdsc->mutex);
1911 	__register_request(mdsc, req, NULL);
1912 	__do_request(mdsc, req);
1913 	mutex_unlock(&mdsc->mutex);
1914 }
1915 
1916 /*
1917  * Synchrously perform an mds request.  Take care of all of the
1918  * session setup, forwarding, retry details.
1919  */
1920 int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
1921 			 struct inode *dir,
1922 			 struct ceph_mds_request *req)
1923 {
1924 	int err;
1925 
1926 	dout("do_request on %p\n", req);
1927 
1928 	/* take CAP_PIN refs for r_inode, r_locked_dir, r_old_dentry */
1929 	if (req->r_inode)
1930 		ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
1931 	if (req->r_locked_dir)
1932 		ceph_get_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
1933 	if (req->r_old_dentry)
1934 		ceph_get_cap_refs(
1935 			ceph_inode(req->r_old_dentry->d_parent->d_inode),
1936 			CEPH_CAP_PIN);
1937 
1938 	/* issue */
1939 	mutex_lock(&mdsc->mutex);
1940 	__register_request(mdsc, req, dir);
1941 	__do_request(mdsc, req);
1942 
1943 	if (req->r_err) {
1944 		err = req->r_err;
1945 		__unregister_request(mdsc, req);
1946 		dout("do_request early error %d\n", err);
1947 		goto out;
1948 	}
1949 
1950 	/* wait */
1951 	mutex_unlock(&mdsc->mutex);
1952 	dout("do_request waiting\n");
1953 	if (req->r_timeout) {
1954 		err = (long)wait_for_completion_killable_timeout(
1955 			&req->r_completion, req->r_timeout);
1956 		if (err == 0)
1957 			err = -EIO;
1958 	} else {
1959 		err = wait_for_completion_killable(&req->r_completion);
1960 	}
1961 	dout("do_request waited, got %d\n", err);
1962 	mutex_lock(&mdsc->mutex);
1963 
1964 	/* only abort if we didn't race with a real reply */
1965 	if (req->r_got_result) {
1966 		err = le32_to_cpu(req->r_reply_info.head->result);
1967 	} else if (err < 0) {
1968 		dout("aborted request %lld with %d\n", req->r_tid, err);
1969 
1970 		/*
1971 		 * ensure we aren't running concurrently with
1972 		 * ceph_fill_trace or ceph_readdir_prepopulate, which
1973 		 * rely on locks (dir mutex) held by our caller.
1974 		 */
1975 		mutex_lock(&req->r_fill_mutex);
1976 		req->r_err = err;
1977 		req->r_aborted = true;
1978 		mutex_unlock(&req->r_fill_mutex);
1979 
1980 		if (req->r_locked_dir &&
1981 		    (req->r_op & CEPH_MDS_OP_WRITE))
1982 			ceph_invalidate_dir_request(req);
1983 	} else {
1984 		err = req->r_err;
1985 	}
1986 
1987 out:
1988 	mutex_unlock(&mdsc->mutex);
1989 	dout("do_request %p done, result %d\n", req, err);
1990 	return err;
1991 }
1992 
1993 /*
1994  * Invalidate dir I_COMPLETE, dentry lease state on an aborted MDS
1995  * namespace request.
1996  */
1997 void ceph_invalidate_dir_request(struct ceph_mds_request *req)
1998 {
1999 	struct inode *inode = req->r_locked_dir;
2000 	struct ceph_inode_info *ci = ceph_inode(inode);
2001 
2002 	dout("invalidate_dir_request %p (I_COMPLETE, lease(s))\n", inode);
2003 	spin_lock(&inode->i_lock);
2004 	ci->i_ceph_flags &= ~CEPH_I_COMPLETE;
2005 	ci->i_release_count++;
2006 	spin_unlock(&inode->i_lock);
2007 
2008 	if (req->r_dentry)
2009 		ceph_invalidate_dentry_lease(req->r_dentry);
2010 	if (req->r_old_dentry)
2011 		ceph_invalidate_dentry_lease(req->r_old_dentry);
2012 }
2013 
2014 /*
2015  * Handle mds reply.
2016  *
2017  * We take the session mutex and parse and process the reply immediately.
2018  * This preserves the logical ordering of replies, capabilities, etc., sent
2019  * by the MDS as they are applied to our local cache.
2020  */
2021 static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
2022 {
2023 	struct ceph_mds_client *mdsc = session->s_mdsc;
2024 	struct ceph_mds_request *req;
2025 	struct ceph_mds_reply_head *head = msg->front.iov_base;
2026 	struct ceph_mds_reply_info_parsed *rinfo;  /* parsed reply info */
2027 	u64 tid;
2028 	int err, result;
2029 	int mds = session->s_mds;
2030 
2031 	if (msg->front.iov_len < sizeof(*head)) {
2032 		pr_err("mdsc_handle_reply got corrupt (short) reply\n");
2033 		ceph_msg_dump(msg);
2034 		return;
2035 	}
2036 
2037 	/* get request, session */
2038 	tid = le64_to_cpu(msg->hdr.tid);
2039 	mutex_lock(&mdsc->mutex);
2040 	req = __lookup_request(mdsc, tid);
2041 	if (!req) {
2042 		dout("handle_reply on unknown tid %llu\n", tid);
2043 		mutex_unlock(&mdsc->mutex);
2044 		return;
2045 	}
2046 	dout("handle_reply %p\n", req);
2047 
2048 	/* correct session? */
2049 	if (req->r_session != session) {
2050 		pr_err("mdsc_handle_reply got %llu on session mds%d"
2051 		       " not mds%d\n", tid, session->s_mds,
2052 		       req->r_session ? req->r_session->s_mds : -1);
2053 		mutex_unlock(&mdsc->mutex);
2054 		goto out;
2055 	}
2056 
2057 	/* dup? */
2058 	if ((req->r_got_unsafe && !head->safe) ||
2059 	    (req->r_got_safe && head->safe)) {
2060 		pr_warning("got a dup %s reply on %llu from mds%d\n",
2061 			   head->safe ? "safe" : "unsafe", tid, mds);
2062 		mutex_unlock(&mdsc->mutex);
2063 		goto out;
2064 	}
2065 	if (req->r_got_safe && !head->safe) {
2066 		pr_warning("got unsafe after safe on %llu from mds%d\n",
2067 			   tid, mds);
2068 		mutex_unlock(&mdsc->mutex);
2069 		goto out;
2070 	}
2071 
2072 	result = le32_to_cpu(head->result);
2073 
2074 	/*
2075 	 * Handle an ESTALE
2076 	 * if we're not talking to the authority, send to them
2077 	 * if the authority has changed while we weren't looking,
2078 	 * send to new authority
2079 	 * Otherwise we just have to return an ESTALE
2080 	 */
2081 	if (result == -ESTALE) {
2082 		dout("got ESTALE on request %llu", req->r_tid);
2083 		if (!req->r_inode) {
2084 			/* do nothing; not an authority problem */
2085 		} else if (req->r_direct_mode != USE_AUTH_MDS) {
2086 			dout("not using auth, setting for that now");
2087 			req->r_direct_mode = USE_AUTH_MDS;
2088 			__do_request(mdsc, req);
2089 			mutex_unlock(&mdsc->mutex);
2090 			goto out;
2091 		} else  {
2092 			struct ceph_inode_info *ci = ceph_inode(req->r_inode);
2093 			struct ceph_cap *cap = NULL;
2094 
2095 			if (req->r_session)
2096 				cap = ceph_get_cap_for_mds(ci,
2097 						   req->r_session->s_mds);
2098 
2099 			dout("already using auth");
2100 			if ((!cap || cap != ci->i_auth_cap) ||
2101 			    (cap->mseq != req->r_sent_on_mseq)) {
2102 				dout("but cap changed, so resending");
2103 				__do_request(mdsc, req);
2104 				mutex_unlock(&mdsc->mutex);
2105 				goto out;
2106 			}
2107 		}
2108 		dout("have to return ESTALE on request %llu", req->r_tid);
2109 	}
2110 
2111 
2112 	if (head->safe) {
2113 		req->r_got_safe = true;
2114 		__unregister_request(mdsc, req);
2115 		complete_all(&req->r_safe_completion);
2116 
2117 		if (req->r_got_unsafe) {
2118 			/*
2119 			 * We already handled the unsafe response, now do the
2120 			 * cleanup.  No need to examine the response; the MDS
2121 			 * doesn't include any result info in the safe
2122 			 * response.  And even if it did, there is nothing
2123 			 * useful we could do with a revised return value.
2124 			 */
2125 			dout("got safe reply %llu, mds%d\n", tid, mds);
2126 			list_del_init(&req->r_unsafe_item);
2127 
2128 			/* last unsafe request during umount? */
2129 			if (mdsc->stopping && !__get_oldest_req(mdsc))
2130 				complete_all(&mdsc->safe_umount_waiters);
2131 			mutex_unlock(&mdsc->mutex);
2132 			goto out;
2133 		}
2134 	} else {
2135 		req->r_got_unsafe = true;
2136 		list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);
2137 	}
2138 
2139 	dout("handle_reply tid %lld result %d\n", tid, result);
2140 	rinfo = &req->r_reply_info;
2141 	err = parse_reply_info(msg, rinfo, session->s_con.peer_features);
2142 	mutex_unlock(&mdsc->mutex);
2143 
2144 	mutex_lock(&session->s_mutex);
2145 	if (err < 0) {
2146 		pr_err("mdsc_handle_reply got corrupt reply mds%d(tid:%lld)\n", mds, tid);
2147 		ceph_msg_dump(msg);
2148 		goto out_err;
2149 	}
2150 
2151 	/* snap trace */
2152 	if (rinfo->snapblob_len) {
2153 		down_write(&mdsc->snap_rwsem);
2154 		ceph_update_snap_trace(mdsc, rinfo->snapblob,
2155 			       rinfo->snapblob + rinfo->snapblob_len,
2156 			       le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP);
2157 		downgrade_write(&mdsc->snap_rwsem);
2158 	} else {
2159 		down_read(&mdsc->snap_rwsem);
2160 	}
2161 
2162 	/* insert trace into our cache */
2163 	mutex_lock(&req->r_fill_mutex);
2164 	err = ceph_fill_trace(mdsc->fsc->sb, req, req->r_session);
2165 	if (err == 0) {
2166 		if (result == 0 && req->r_op != CEPH_MDS_OP_GETFILELOCK &&
2167 		    rinfo->dir_nr)
2168 			ceph_readdir_prepopulate(req, req->r_session);
2169 		ceph_unreserve_caps(mdsc, &req->r_caps_reservation);
2170 	}
2171 	mutex_unlock(&req->r_fill_mutex);
2172 
2173 	up_read(&mdsc->snap_rwsem);
2174 out_err:
2175 	mutex_lock(&mdsc->mutex);
2176 	if (!req->r_aborted) {
2177 		if (err) {
2178 			req->r_err = err;
2179 		} else {
2180 			req->r_reply = msg;
2181 			ceph_msg_get(msg);
2182 			req->r_got_result = true;
2183 		}
2184 	} else {
2185 		dout("reply arrived after request %lld was aborted\n", tid);
2186 	}
2187 	mutex_unlock(&mdsc->mutex);
2188 
2189 	ceph_add_cap_releases(mdsc, req->r_session);
2190 	mutex_unlock(&session->s_mutex);
2191 
2192 	/* kick calling process */
2193 	complete_request(mdsc, req);
2194 out:
2195 	ceph_mdsc_put_request(req);
2196 	return;
2197 }
2198 
2199 
2200 
2201 /*
2202  * handle mds notification that our request has been forwarded.
2203  */
2204 static void handle_forward(struct ceph_mds_client *mdsc,
2205 			   struct ceph_mds_session *session,
2206 			   struct ceph_msg *msg)
2207 {
2208 	struct ceph_mds_request *req;
2209 	u64 tid = le64_to_cpu(msg->hdr.tid);
2210 	u32 next_mds;
2211 	u32 fwd_seq;
2212 	int err = -EINVAL;
2213 	void *p = msg->front.iov_base;
2214 	void *end = p + msg->front.iov_len;
2215 
2216 	ceph_decode_need(&p, end, 2*sizeof(u32), bad);
2217 	next_mds = ceph_decode_32(&p);
2218 	fwd_seq = ceph_decode_32(&p);
2219 
2220 	mutex_lock(&mdsc->mutex);
2221 	req = __lookup_request(mdsc, tid);
2222 	if (!req) {
2223 		dout("forward tid %llu to mds%d - req dne\n", tid, next_mds);
2224 		goto out;  /* dup reply? */
2225 	}
2226 
2227 	if (req->r_aborted) {
2228 		dout("forward tid %llu aborted, unregistering\n", tid);
2229 		__unregister_request(mdsc, req);
2230 	} else if (fwd_seq <= req->r_num_fwd) {
2231 		dout("forward tid %llu to mds%d - old seq %d <= %d\n",
2232 		     tid, next_mds, req->r_num_fwd, fwd_seq);
2233 	} else {
2234 		/* resend. forward race not possible; mds would drop */
2235 		dout("forward tid %llu to mds%d (we resend)\n", tid, next_mds);
2236 		BUG_ON(req->r_err);
2237 		BUG_ON(req->r_got_result);
2238 		req->r_num_fwd = fwd_seq;
2239 		req->r_resend_mds = next_mds;
2240 		put_request_session(req);
2241 		__do_request(mdsc, req);
2242 	}
2243 	ceph_mdsc_put_request(req);
2244 out:
2245 	mutex_unlock(&mdsc->mutex);
2246 	return;
2247 
2248 bad:
2249 	pr_err("mdsc_handle_forward decode error err=%d\n", err);
2250 }
2251 
2252 /*
2253  * handle a mds session control message
2254  */
2255 static void handle_session(struct ceph_mds_session *session,
2256 			   struct ceph_msg *msg)
2257 {
2258 	struct ceph_mds_client *mdsc = session->s_mdsc;
2259 	u32 op;
2260 	u64 seq;
2261 	int mds = session->s_mds;
2262 	struct ceph_mds_session_head *h = msg->front.iov_base;
2263 	int wake = 0;
2264 
2265 	/* decode */
2266 	if (msg->front.iov_len != sizeof(*h))
2267 		goto bad;
2268 	op = le32_to_cpu(h->op);
2269 	seq = le64_to_cpu(h->seq);
2270 
2271 	mutex_lock(&mdsc->mutex);
2272 	if (op == CEPH_SESSION_CLOSE)
2273 		__unregister_session(mdsc, session);
2274 	/* FIXME: this ttl calculation is generous */
2275 	session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose;
2276 	mutex_unlock(&mdsc->mutex);
2277 
2278 	mutex_lock(&session->s_mutex);
2279 
2280 	dout("handle_session mds%d %s %p state %s seq %llu\n",
2281 	     mds, ceph_session_op_name(op), session,
2282 	     session_state_name(session->s_state), seq);
2283 
2284 	if (session->s_state == CEPH_MDS_SESSION_HUNG) {
2285 		session->s_state = CEPH_MDS_SESSION_OPEN;
2286 		pr_info("mds%d came back\n", session->s_mds);
2287 	}
2288 
2289 	switch (op) {
2290 	case CEPH_SESSION_OPEN:
2291 		if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
2292 			pr_info("mds%d reconnect success\n", session->s_mds);
2293 		session->s_state = CEPH_MDS_SESSION_OPEN;
2294 		renewed_caps(mdsc, session, 0);
2295 		wake = 1;
2296 		if (mdsc->stopping)
2297 			__close_session(mdsc, session);
2298 		break;
2299 
2300 	case CEPH_SESSION_RENEWCAPS:
2301 		if (session->s_renew_seq == seq)
2302 			renewed_caps(mdsc, session, 1);
2303 		break;
2304 
2305 	case CEPH_SESSION_CLOSE:
2306 		if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
2307 			pr_info("mds%d reconnect denied\n", session->s_mds);
2308 		remove_session_caps(session);
2309 		wake = 1; /* for good measure */
2310 		wake_up_all(&mdsc->session_close_wq);
2311 		kick_requests(mdsc, mds);
2312 		break;
2313 
2314 	case CEPH_SESSION_STALE:
2315 		pr_info("mds%d caps went stale, renewing\n",
2316 			session->s_mds);
2317 		spin_lock(&session->s_cap_lock);
2318 		session->s_cap_gen++;
2319 		session->s_cap_ttl = 0;
2320 		spin_unlock(&session->s_cap_lock);
2321 		send_renew_caps(mdsc, session);
2322 		break;
2323 
2324 	case CEPH_SESSION_RECALL_STATE:
2325 		trim_caps(mdsc, session, le32_to_cpu(h->max_caps));
2326 		break;
2327 
2328 	default:
2329 		pr_err("mdsc_handle_session bad op %d mds%d\n", op, mds);
2330 		WARN_ON(1);
2331 	}
2332 
2333 	mutex_unlock(&session->s_mutex);
2334 	if (wake) {
2335 		mutex_lock(&mdsc->mutex);
2336 		__wake_requests(mdsc, &session->s_waiting);
2337 		mutex_unlock(&mdsc->mutex);
2338 	}
2339 	return;
2340 
2341 bad:
2342 	pr_err("mdsc_handle_session corrupt message mds%d len %d\n", mds,
2343 	       (int)msg->front.iov_len);
2344 	ceph_msg_dump(msg);
2345 	return;
2346 }
2347 
2348 
2349 /*
2350  * called under session->mutex.
2351  */
2352 static void replay_unsafe_requests(struct ceph_mds_client *mdsc,
2353 				   struct ceph_mds_session *session)
2354 {
2355 	struct ceph_mds_request *req, *nreq;
2356 	int err;
2357 
2358 	dout("replay_unsafe_requests mds%d\n", session->s_mds);
2359 
2360 	mutex_lock(&mdsc->mutex);
2361 	list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item) {
2362 		err = __prepare_send_request(mdsc, req, session->s_mds);
2363 		if (!err) {
2364 			ceph_msg_get(req->r_request);
2365 			ceph_con_send(&session->s_con, req->r_request);
2366 		}
2367 	}
2368 	mutex_unlock(&mdsc->mutex);
2369 }
2370 
2371 /*
2372  * Encode information about a cap for a reconnect with the MDS.
2373  */
2374 static int encode_caps_cb(struct inode *inode, struct ceph_cap *cap,
2375 			  void *arg)
2376 {
2377 	union {
2378 		struct ceph_mds_cap_reconnect v2;
2379 		struct ceph_mds_cap_reconnect_v1 v1;
2380 	} rec;
2381 	size_t reclen;
2382 	struct ceph_inode_info *ci;
2383 	struct ceph_reconnect_state *recon_state = arg;
2384 	struct ceph_pagelist *pagelist = recon_state->pagelist;
2385 	char *path;
2386 	int pathlen, err;
2387 	u64 pathbase;
2388 	struct dentry *dentry;
2389 
2390 	ci = cap->ci;
2391 
2392 	dout(" adding %p ino %llx.%llx cap %p %lld %s\n",
2393 	     inode, ceph_vinop(inode), cap, cap->cap_id,
2394 	     ceph_cap_string(cap->issued));
2395 	err = ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
2396 	if (err)
2397 		return err;
2398 
2399 	dentry = d_find_alias(inode);
2400 	if (dentry) {
2401 		path = ceph_mdsc_build_path(dentry, &pathlen, &pathbase, 0);
2402 		if (IS_ERR(path)) {
2403 			err = PTR_ERR(path);
2404 			goto out_dput;
2405 		}
2406 	} else {
2407 		path = NULL;
2408 		pathlen = 0;
2409 	}
2410 	err = ceph_pagelist_encode_string(pagelist, path, pathlen);
2411 	if (err)
2412 		goto out_free;
2413 
2414 	spin_lock(&inode->i_lock);
2415 	cap->seq = 0;        /* reset cap seq */
2416 	cap->issue_seq = 0;  /* and issue_seq */
2417 
2418 	if (recon_state->flock) {
2419 		rec.v2.cap_id = cpu_to_le64(cap->cap_id);
2420 		rec.v2.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
2421 		rec.v2.issued = cpu_to_le32(cap->issued);
2422 		rec.v2.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
2423 		rec.v2.pathbase = cpu_to_le64(pathbase);
2424 		rec.v2.flock_len = 0;
2425 		reclen = sizeof(rec.v2);
2426 	} else {
2427 		rec.v1.cap_id = cpu_to_le64(cap->cap_id);
2428 		rec.v1.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
2429 		rec.v1.issued = cpu_to_le32(cap->issued);
2430 		rec.v1.size = cpu_to_le64(inode->i_size);
2431 		ceph_encode_timespec(&rec.v1.mtime, &inode->i_mtime);
2432 		ceph_encode_timespec(&rec.v1.atime, &inode->i_atime);
2433 		rec.v1.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
2434 		rec.v1.pathbase = cpu_to_le64(pathbase);
2435 		reclen = sizeof(rec.v1);
2436 	}
2437 	spin_unlock(&inode->i_lock);
2438 
2439 	if (recon_state->flock) {
2440 		int num_fcntl_locks, num_flock_locks;
2441 		struct ceph_pagelist_cursor trunc_point;
2442 
2443 		ceph_pagelist_set_cursor(pagelist, &trunc_point);
2444 		do {
2445 			lock_flocks();
2446 			ceph_count_locks(inode, &num_fcntl_locks,
2447 					 &num_flock_locks);
2448 			rec.v2.flock_len = (2*sizeof(u32) +
2449 					    (num_fcntl_locks+num_flock_locks) *
2450 					    sizeof(struct ceph_filelock));
2451 			unlock_flocks();
2452 
2453 			/* pre-alloc pagelist */
2454 			ceph_pagelist_truncate(pagelist, &trunc_point);
2455 			err = ceph_pagelist_append(pagelist, &rec, reclen);
2456 			if (!err)
2457 				err = ceph_pagelist_reserve(pagelist,
2458 							    rec.v2.flock_len);
2459 
2460 			/* encode locks */
2461 			if (!err) {
2462 				lock_flocks();
2463 				err = ceph_encode_locks(inode,
2464 							pagelist,
2465 							num_fcntl_locks,
2466 							num_flock_locks);
2467 				unlock_flocks();
2468 			}
2469 		} while (err == -ENOSPC);
2470 	} else {
2471 		err = ceph_pagelist_append(pagelist, &rec, reclen);
2472 	}
2473 
2474 out_free:
2475 	kfree(path);
2476 out_dput:
2477 	dput(dentry);
2478 	return err;
2479 }
2480 
2481 
2482 /*
2483  * If an MDS fails and recovers, clients need to reconnect in order to
2484  * reestablish shared state.  This includes all caps issued through
2485  * this session _and_ the snap_realm hierarchy.  Because it's not
2486  * clear which snap realms the mds cares about, we send everything we
2487  * know about.. that ensures we'll then get any new info the
2488  * recovering MDS might have.
2489  *
2490  * This is a relatively heavyweight operation, but it's rare.
2491  *
2492  * called with mdsc->mutex held.
2493  */
2494 static void send_mds_reconnect(struct ceph_mds_client *mdsc,
2495 			       struct ceph_mds_session *session)
2496 {
2497 	struct ceph_msg *reply;
2498 	struct rb_node *p;
2499 	int mds = session->s_mds;
2500 	int err = -ENOMEM;
2501 	struct ceph_pagelist *pagelist;
2502 	struct ceph_reconnect_state recon_state;
2503 
2504 	pr_info("mds%d reconnect start\n", mds);
2505 
2506 	pagelist = kmalloc(sizeof(*pagelist), GFP_NOFS);
2507 	if (!pagelist)
2508 		goto fail_nopagelist;
2509 	ceph_pagelist_init(pagelist);
2510 
2511 	reply = ceph_msg_new(CEPH_MSG_CLIENT_RECONNECT, 0, GFP_NOFS);
2512 	if (!reply)
2513 		goto fail_nomsg;
2514 
2515 	mutex_lock(&session->s_mutex);
2516 	session->s_state = CEPH_MDS_SESSION_RECONNECTING;
2517 	session->s_seq = 0;
2518 
2519 	ceph_con_open(&session->s_con,
2520 		      ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
2521 
2522 	/* replay unsafe requests */
2523 	replay_unsafe_requests(mdsc, session);
2524 
2525 	down_read(&mdsc->snap_rwsem);
2526 
2527 	dout("session %p state %s\n", session,
2528 	     session_state_name(session->s_state));
2529 
2530 	/* drop old cap expires; we're about to reestablish that state */
2531 	discard_cap_releases(mdsc, session);
2532 
2533 	/* traverse this session's caps */
2534 	err = ceph_pagelist_encode_32(pagelist, session->s_nr_caps);
2535 	if (err)
2536 		goto fail;
2537 
2538 	recon_state.pagelist = pagelist;
2539 	recon_state.flock = session->s_con.peer_features & CEPH_FEATURE_FLOCK;
2540 	err = iterate_session_caps(session, encode_caps_cb, &recon_state);
2541 	if (err < 0)
2542 		goto fail;
2543 
2544 	/*
2545 	 * snaprealms.  we provide mds with the ino, seq (version), and
2546 	 * parent for all of our realms.  If the mds has any newer info,
2547 	 * it will tell us.
2548 	 */
2549 	for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) {
2550 		struct ceph_snap_realm *realm =
2551 			rb_entry(p, struct ceph_snap_realm, node);
2552 		struct ceph_mds_snaprealm_reconnect sr_rec;
2553 
2554 		dout(" adding snap realm %llx seq %lld parent %llx\n",
2555 		     realm->ino, realm->seq, realm->parent_ino);
2556 		sr_rec.ino = cpu_to_le64(realm->ino);
2557 		sr_rec.seq = cpu_to_le64(realm->seq);
2558 		sr_rec.parent = cpu_to_le64(realm->parent_ino);
2559 		err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec));
2560 		if (err)
2561 			goto fail;
2562 	}
2563 
2564 	reply->pagelist = pagelist;
2565 	if (recon_state.flock)
2566 		reply->hdr.version = cpu_to_le16(2);
2567 	reply->hdr.data_len = cpu_to_le32(pagelist->length);
2568 	reply->nr_pages = calc_pages_for(0, pagelist->length);
2569 	ceph_con_send(&session->s_con, reply);
2570 
2571 	mutex_unlock(&session->s_mutex);
2572 
2573 	mutex_lock(&mdsc->mutex);
2574 	__wake_requests(mdsc, &session->s_waiting);
2575 	mutex_unlock(&mdsc->mutex);
2576 
2577 	up_read(&mdsc->snap_rwsem);
2578 	return;
2579 
2580 fail:
2581 	ceph_msg_put(reply);
2582 	up_read(&mdsc->snap_rwsem);
2583 	mutex_unlock(&session->s_mutex);
2584 fail_nomsg:
2585 	ceph_pagelist_release(pagelist);
2586 	kfree(pagelist);
2587 fail_nopagelist:
2588 	pr_err("error %d preparing reconnect for mds%d\n", err, mds);
2589 	return;
2590 }
2591 
2592 
2593 /*
2594  * compare old and new mdsmaps, kicking requests
2595  * and closing out old connections as necessary
2596  *
2597  * called under mdsc->mutex.
2598  */
2599 static void check_new_map(struct ceph_mds_client *mdsc,
2600 			  struct ceph_mdsmap *newmap,
2601 			  struct ceph_mdsmap *oldmap)
2602 {
2603 	int i;
2604 	int oldstate, newstate;
2605 	struct ceph_mds_session *s;
2606 
2607 	dout("check_new_map new %u old %u\n",
2608 	     newmap->m_epoch, oldmap->m_epoch);
2609 
2610 	for (i = 0; i < oldmap->m_max_mds && i < mdsc->max_sessions; i++) {
2611 		if (mdsc->sessions[i] == NULL)
2612 			continue;
2613 		s = mdsc->sessions[i];
2614 		oldstate = ceph_mdsmap_get_state(oldmap, i);
2615 		newstate = ceph_mdsmap_get_state(newmap, i);
2616 
2617 		dout("check_new_map mds%d state %s%s -> %s%s (session %s)\n",
2618 		     i, ceph_mds_state_name(oldstate),
2619 		     ceph_mdsmap_is_laggy(oldmap, i) ? " (laggy)" : "",
2620 		     ceph_mds_state_name(newstate),
2621 		     ceph_mdsmap_is_laggy(newmap, i) ? " (laggy)" : "",
2622 		     session_state_name(s->s_state));
2623 
2624 		if (memcmp(ceph_mdsmap_get_addr(oldmap, i),
2625 			   ceph_mdsmap_get_addr(newmap, i),
2626 			   sizeof(struct ceph_entity_addr))) {
2627 			if (s->s_state == CEPH_MDS_SESSION_OPENING) {
2628 				/* the session never opened, just close it
2629 				 * out now */
2630 				__wake_requests(mdsc, &s->s_waiting);
2631 				__unregister_session(mdsc, s);
2632 			} else {
2633 				/* just close it */
2634 				mutex_unlock(&mdsc->mutex);
2635 				mutex_lock(&s->s_mutex);
2636 				mutex_lock(&mdsc->mutex);
2637 				ceph_con_close(&s->s_con);
2638 				mutex_unlock(&s->s_mutex);
2639 				s->s_state = CEPH_MDS_SESSION_RESTARTING;
2640 			}
2641 
2642 			/* kick any requests waiting on the recovering mds */
2643 			kick_requests(mdsc, i);
2644 		} else if (oldstate == newstate) {
2645 			continue;  /* nothing new with this mds */
2646 		}
2647 
2648 		/*
2649 		 * send reconnect?
2650 		 */
2651 		if (s->s_state == CEPH_MDS_SESSION_RESTARTING &&
2652 		    newstate >= CEPH_MDS_STATE_RECONNECT) {
2653 			mutex_unlock(&mdsc->mutex);
2654 			send_mds_reconnect(mdsc, s);
2655 			mutex_lock(&mdsc->mutex);
2656 		}
2657 
2658 		/*
2659 		 * kick request on any mds that has gone active.
2660 		 */
2661 		if (oldstate < CEPH_MDS_STATE_ACTIVE &&
2662 		    newstate >= CEPH_MDS_STATE_ACTIVE) {
2663 			if (oldstate != CEPH_MDS_STATE_CREATING &&
2664 			    oldstate != CEPH_MDS_STATE_STARTING)
2665 				pr_info("mds%d recovery completed\n", s->s_mds);
2666 			kick_requests(mdsc, i);
2667 			ceph_kick_flushing_caps(mdsc, s);
2668 			wake_up_session_caps(s, 1);
2669 		}
2670 	}
2671 
2672 	for (i = 0; i < newmap->m_max_mds && i < mdsc->max_sessions; i++) {
2673 		s = mdsc->sessions[i];
2674 		if (!s)
2675 			continue;
2676 		if (!ceph_mdsmap_is_laggy(newmap, i))
2677 			continue;
2678 		if (s->s_state == CEPH_MDS_SESSION_OPEN ||
2679 		    s->s_state == CEPH_MDS_SESSION_HUNG ||
2680 		    s->s_state == CEPH_MDS_SESSION_CLOSING) {
2681 			dout(" connecting to export targets of laggy mds%d\n",
2682 			     i);
2683 			__open_export_target_sessions(mdsc, s);
2684 		}
2685 	}
2686 }
2687 
2688 
2689 
2690 /*
2691  * leases
2692  */
2693 
2694 /*
2695  * caller must hold session s_mutex, dentry->d_lock
2696  */
2697 void __ceph_mdsc_drop_dentry_lease(struct dentry *dentry)
2698 {
2699 	struct ceph_dentry_info *di = ceph_dentry(dentry);
2700 
2701 	ceph_put_mds_session(di->lease_session);
2702 	di->lease_session = NULL;
2703 }
2704 
2705 static void handle_lease(struct ceph_mds_client *mdsc,
2706 			 struct ceph_mds_session *session,
2707 			 struct ceph_msg *msg)
2708 {
2709 	struct super_block *sb = mdsc->fsc->sb;
2710 	struct inode *inode;
2711 	struct dentry *parent, *dentry;
2712 	struct ceph_dentry_info *di;
2713 	int mds = session->s_mds;
2714 	struct ceph_mds_lease *h = msg->front.iov_base;
2715 	u32 seq;
2716 	struct ceph_vino vino;
2717 	int mask;
2718 	struct qstr dname;
2719 	int release = 0;
2720 
2721 	dout("handle_lease from mds%d\n", mds);
2722 
2723 	/* decode */
2724 	if (msg->front.iov_len < sizeof(*h) + sizeof(u32))
2725 		goto bad;
2726 	vino.ino = le64_to_cpu(h->ino);
2727 	vino.snap = CEPH_NOSNAP;
2728 	mask = le16_to_cpu(h->mask);
2729 	seq = le32_to_cpu(h->seq);
2730 	dname.name = (void *)h + sizeof(*h) + sizeof(u32);
2731 	dname.len = msg->front.iov_len - sizeof(*h) - sizeof(u32);
2732 	if (dname.len != get_unaligned_le32(h+1))
2733 		goto bad;
2734 
2735 	mutex_lock(&session->s_mutex);
2736 	session->s_seq++;
2737 
2738 	/* lookup inode */
2739 	inode = ceph_find_inode(sb, vino);
2740 	dout("handle_lease %s, mask %d, ino %llx %p %.*s\n",
2741 	     ceph_lease_op_name(h->action), mask, vino.ino, inode,
2742 	     dname.len, dname.name);
2743 	if (inode == NULL) {
2744 		dout("handle_lease no inode %llx\n", vino.ino);
2745 		goto release;
2746 	}
2747 
2748 	/* dentry */
2749 	parent = d_find_alias(inode);
2750 	if (!parent) {
2751 		dout("no parent dentry on inode %p\n", inode);
2752 		WARN_ON(1);
2753 		goto release;  /* hrm... */
2754 	}
2755 	dname.hash = full_name_hash(dname.name, dname.len);
2756 	dentry = d_lookup(parent, &dname);
2757 	dput(parent);
2758 	if (!dentry)
2759 		goto release;
2760 
2761 	spin_lock(&dentry->d_lock);
2762 	di = ceph_dentry(dentry);
2763 	switch (h->action) {
2764 	case CEPH_MDS_LEASE_REVOKE:
2765 		if (di && di->lease_session == session) {
2766 			if (ceph_seq_cmp(di->lease_seq, seq) > 0)
2767 				h->seq = cpu_to_le32(di->lease_seq);
2768 			__ceph_mdsc_drop_dentry_lease(dentry);
2769 		}
2770 		release = 1;
2771 		break;
2772 
2773 	case CEPH_MDS_LEASE_RENEW:
2774 		if (di && di->lease_session == session &&
2775 		    di->lease_gen == session->s_cap_gen &&
2776 		    di->lease_renew_from &&
2777 		    di->lease_renew_after == 0) {
2778 			unsigned long duration =
2779 				le32_to_cpu(h->duration_ms) * HZ / 1000;
2780 
2781 			di->lease_seq = seq;
2782 			dentry->d_time = di->lease_renew_from + duration;
2783 			di->lease_renew_after = di->lease_renew_from +
2784 				(duration >> 1);
2785 			di->lease_renew_from = 0;
2786 		}
2787 		break;
2788 	}
2789 	spin_unlock(&dentry->d_lock);
2790 	dput(dentry);
2791 
2792 	if (!release)
2793 		goto out;
2794 
2795 release:
2796 	/* let's just reuse the same message */
2797 	h->action = CEPH_MDS_LEASE_REVOKE_ACK;
2798 	ceph_msg_get(msg);
2799 	ceph_con_send(&session->s_con, msg);
2800 
2801 out:
2802 	iput(inode);
2803 	mutex_unlock(&session->s_mutex);
2804 	return;
2805 
2806 bad:
2807 	pr_err("corrupt lease message\n");
2808 	ceph_msg_dump(msg);
2809 }
2810 
2811 void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session,
2812 			      struct inode *inode,
2813 			      struct dentry *dentry, char action,
2814 			      u32 seq)
2815 {
2816 	struct ceph_msg *msg;
2817 	struct ceph_mds_lease *lease;
2818 	int len = sizeof(*lease) + sizeof(u32);
2819 	int dnamelen = 0;
2820 
2821 	dout("lease_send_msg inode %p dentry %p %s to mds%d\n",
2822 	     inode, dentry, ceph_lease_op_name(action), session->s_mds);
2823 	dnamelen = dentry->d_name.len;
2824 	len += dnamelen;
2825 
2826 	msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS);
2827 	if (!msg)
2828 		return;
2829 	lease = msg->front.iov_base;
2830 	lease->action = action;
2831 	lease->mask = cpu_to_le16(1);
2832 	lease->ino = cpu_to_le64(ceph_vino(inode).ino);
2833 	lease->first = lease->last = cpu_to_le64(ceph_vino(inode).snap);
2834 	lease->seq = cpu_to_le32(seq);
2835 	put_unaligned_le32(dnamelen, lease + 1);
2836 	memcpy((void *)(lease + 1) + 4, dentry->d_name.name, dnamelen);
2837 
2838 	/*
2839 	 * if this is a preemptive lease RELEASE, no need to
2840 	 * flush request stream, since the actual request will
2841 	 * soon follow.
2842 	 */
2843 	msg->more_to_follow = (action == CEPH_MDS_LEASE_RELEASE);
2844 
2845 	ceph_con_send(&session->s_con, msg);
2846 }
2847 
2848 /*
2849  * Preemptively release a lease we expect to invalidate anyway.
2850  * Pass @inode always, @dentry is optional.
2851  */
2852 void ceph_mdsc_lease_release(struct ceph_mds_client *mdsc, struct inode *inode,
2853 			     struct dentry *dentry, int mask)
2854 {
2855 	struct ceph_dentry_info *di;
2856 	struct ceph_mds_session *session;
2857 	u32 seq;
2858 
2859 	BUG_ON(inode == NULL);
2860 	BUG_ON(dentry == NULL);
2861 	BUG_ON(mask == 0);
2862 
2863 	/* is dentry lease valid? */
2864 	spin_lock(&dentry->d_lock);
2865 	di = ceph_dentry(dentry);
2866 	if (!di || !di->lease_session ||
2867 	    di->lease_session->s_mds < 0 ||
2868 	    di->lease_gen != di->lease_session->s_cap_gen ||
2869 	    !time_before(jiffies, dentry->d_time)) {
2870 		dout("lease_release inode %p dentry %p -- "
2871 		     "no lease on %d\n",
2872 		     inode, dentry, mask);
2873 		spin_unlock(&dentry->d_lock);
2874 		return;
2875 	}
2876 
2877 	/* we do have a lease on this dentry; note mds and seq */
2878 	session = ceph_get_mds_session(di->lease_session);
2879 	seq = di->lease_seq;
2880 	__ceph_mdsc_drop_dentry_lease(dentry);
2881 	spin_unlock(&dentry->d_lock);
2882 
2883 	dout("lease_release inode %p dentry %p mask %d to mds%d\n",
2884 	     inode, dentry, mask, session->s_mds);
2885 	ceph_mdsc_lease_send_msg(session, inode, dentry,
2886 				 CEPH_MDS_LEASE_RELEASE, seq);
2887 	ceph_put_mds_session(session);
2888 }
2889 
2890 /*
2891  * drop all leases (and dentry refs) in preparation for umount
2892  */
2893 static void drop_leases(struct ceph_mds_client *mdsc)
2894 {
2895 	int i;
2896 
2897 	dout("drop_leases\n");
2898 	mutex_lock(&mdsc->mutex);
2899 	for (i = 0; i < mdsc->max_sessions; i++) {
2900 		struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
2901 		if (!s)
2902 			continue;
2903 		mutex_unlock(&mdsc->mutex);
2904 		mutex_lock(&s->s_mutex);
2905 		mutex_unlock(&s->s_mutex);
2906 		ceph_put_mds_session(s);
2907 		mutex_lock(&mdsc->mutex);
2908 	}
2909 	mutex_unlock(&mdsc->mutex);
2910 }
2911 
2912 
2913 
2914 /*
2915  * delayed work -- periodically trim expired leases, renew caps with mds
2916  */
2917 static void schedule_delayed(struct ceph_mds_client *mdsc)
2918 {
2919 	int delay = 5;
2920 	unsigned hz = round_jiffies_relative(HZ * delay);
2921 	schedule_delayed_work(&mdsc->delayed_work, hz);
2922 }
2923 
2924 static void delayed_work(struct work_struct *work)
2925 {
2926 	int i;
2927 	struct ceph_mds_client *mdsc =
2928 		container_of(work, struct ceph_mds_client, delayed_work.work);
2929 	int renew_interval;
2930 	int renew_caps;
2931 
2932 	dout("mdsc delayed_work\n");
2933 	ceph_check_delayed_caps(mdsc);
2934 
2935 	mutex_lock(&mdsc->mutex);
2936 	renew_interval = mdsc->mdsmap->m_session_timeout >> 2;
2937 	renew_caps = time_after_eq(jiffies, HZ*renew_interval +
2938 				   mdsc->last_renew_caps);
2939 	if (renew_caps)
2940 		mdsc->last_renew_caps = jiffies;
2941 
2942 	for (i = 0; i < mdsc->max_sessions; i++) {
2943 		struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
2944 		if (s == NULL)
2945 			continue;
2946 		if (s->s_state == CEPH_MDS_SESSION_CLOSING) {
2947 			dout("resending session close request for mds%d\n",
2948 			     s->s_mds);
2949 			request_close_session(mdsc, s);
2950 			ceph_put_mds_session(s);
2951 			continue;
2952 		}
2953 		if (s->s_ttl && time_after(jiffies, s->s_ttl)) {
2954 			if (s->s_state == CEPH_MDS_SESSION_OPEN) {
2955 				s->s_state = CEPH_MDS_SESSION_HUNG;
2956 				pr_info("mds%d hung\n", s->s_mds);
2957 			}
2958 		}
2959 		if (s->s_state < CEPH_MDS_SESSION_OPEN) {
2960 			/* this mds is failed or recovering, just wait */
2961 			ceph_put_mds_session(s);
2962 			continue;
2963 		}
2964 		mutex_unlock(&mdsc->mutex);
2965 
2966 		mutex_lock(&s->s_mutex);
2967 		if (renew_caps)
2968 			send_renew_caps(mdsc, s);
2969 		else
2970 			ceph_con_keepalive(&s->s_con);
2971 		ceph_add_cap_releases(mdsc, s);
2972 		if (s->s_state == CEPH_MDS_SESSION_OPEN ||
2973 		    s->s_state == CEPH_MDS_SESSION_HUNG)
2974 			ceph_send_cap_releases(mdsc, s);
2975 		mutex_unlock(&s->s_mutex);
2976 		ceph_put_mds_session(s);
2977 
2978 		mutex_lock(&mdsc->mutex);
2979 	}
2980 	mutex_unlock(&mdsc->mutex);
2981 
2982 	schedule_delayed(mdsc);
2983 }
2984 
2985 int ceph_mdsc_init(struct ceph_fs_client *fsc)
2986 
2987 {
2988 	struct ceph_mds_client *mdsc;
2989 
2990 	mdsc = kzalloc(sizeof(struct ceph_mds_client), GFP_NOFS);
2991 	if (!mdsc)
2992 		return -ENOMEM;
2993 	mdsc->fsc = fsc;
2994 	fsc->mdsc = mdsc;
2995 	mutex_init(&mdsc->mutex);
2996 	mdsc->mdsmap = kzalloc(sizeof(*mdsc->mdsmap), GFP_NOFS);
2997 	if (mdsc->mdsmap == NULL)
2998 		return -ENOMEM;
2999 
3000 	init_completion(&mdsc->safe_umount_waiters);
3001 	init_waitqueue_head(&mdsc->session_close_wq);
3002 	INIT_LIST_HEAD(&mdsc->waiting_for_map);
3003 	mdsc->sessions = NULL;
3004 	mdsc->max_sessions = 0;
3005 	mdsc->stopping = 0;
3006 	init_rwsem(&mdsc->snap_rwsem);
3007 	mdsc->snap_realms = RB_ROOT;
3008 	INIT_LIST_HEAD(&mdsc->snap_empty);
3009 	spin_lock_init(&mdsc->snap_empty_lock);
3010 	mdsc->last_tid = 0;
3011 	mdsc->request_tree = RB_ROOT;
3012 	INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work);
3013 	mdsc->last_renew_caps = jiffies;
3014 	INIT_LIST_HEAD(&mdsc->cap_delay_list);
3015 	spin_lock_init(&mdsc->cap_delay_lock);
3016 	INIT_LIST_HEAD(&mdsc->snap_flush_list);
3017 	spin_lock_init(&mdsc->snap_flush_lock);
3018 	mdsc->cap_flush_seq = 0;
3019 	INIT_LIST_HEAD(&mdsc->cap_dirty);
3020 	INIT_LIST_HEAD(&mdsc->cap_dirty_migrating);
3021 	mdsc->num_cap_flushing = 0;
3022 	spin_lock_init(&mdsc->cap_dirty_lock);
3023 	init_waitqueue_head(&mdsc->cap_flushing_wq);
3024 	spin_lock_init(&mdsc->dentry_lru_lock);
3025 	INIT_LIST_HEAD(&mdsc->dentry_lru);
3026 
3027 	ceph_caps_init(mdsc);
3028 	ceph_adjust_min_caps(mdsc, fsc->min_caps);
3029 
3030 	return 0;
3031 }
3032 
3033 /*
3034  * Wait for safe replies on open mds requests.  If we time out, drop
3035  * all requests from the tree to avoid dangling dentry refs.
3036  */
3037 static void wait_requests(struct ceph_mds_client *mdsc)
3038 {
3039 	struct ceph_mds_request *req;
3040 	struct ceph_fs_client *fsc = mdsc->fsc;
3041 
3042 	mutex_lock(&mdsc->mutex);
3043 	if (__get_oldest_req(mdsc)) {
3044 		mutex_unlock(&mdsc->mutex);
3045 
3046 		dout("wait_requests waiting for requests\n");
3047 		wait_for_completion_timeout(&mdsc->safe_umount_waiters,
3048 				    fsc->client->options->mount_timeout * HZ);
3049 
3050 		/* tear down remaining requests */
3051 		mutex_lock(&mdsc->mutex);
3052 		while ((req = __get_oldest_req(mdsc))) {
3053 			dout("wait_requests timed out on tid %llu\n",
3054 			     req->r_tid);
3055 			__unregister_request(mdsc, req);
3056 		}
3057 	}
3058 	mutex_unlock(&mdsc->mutex);
3059 	dout("wait_requests done\n");
3060 }
3061 
3062 /*
3063  * called before mount is ro, and before dentries are torn down.
3064  * (hmm, does this still race with new lookups?)
3065  */
3066 void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc)
3067 {
3068 	dout("pre_umount\n");
3069 	mdsc->stopping = 1;
3070 
3071 	drop_leases(mdsc);
3072 	ceph_flush_dirty_caps(mdsc);
3073 	wait_requests(mdsc);
3074 
3075 	/*
3076 	 * wait for reply handlers to drop their request refs and
3077 	 * their inode/dcache refs
3078 	 */
3079 	ceph_msgr_flush();
3080 }
3081 
3082 /*
3083  * wait for all write mds requests to flush.
3084  */
3085 static void wait_unsafe_requests(struct ceph_mds_client *mdsc, u64 want_tid)
3086 {
3087 	struct ceph_mds_request *req = NULL, *nextreq;
3088 	struct rb_node *n;
3089 
3090 	mutex_lock(&mdsc->mutex);
3091 	dout("wait_unsafe_requests want %lld\n", want_tid);
3092 restart:
3093 	req = __get_oldest_req(mdsc);
3094 	while (req && req->r_tid <= want_tid) {
3095 		/* find next request */
3096 		n = rb_next(&req->r_node);
3097 		if (n)
3098 			nextreq = rb_entry(n, struct ceph_mds_request, r_node);
3099 		else
3100 			nextreq = NULL;
3101 		if ((req->r_op & CEPH_MDS_OP_WRITE)) {
3102 			/* write op */
3103 			ceph_mdsc_get_request(req);
3104 			if (nextreq)
3105 				ceph_mdsc_get_request(nextreq);
3106 			mutex_unlock(&mdsc->mutex);
3107 			dout("wait_unsafe_requests  wait on %llu (want %llu)\n",
3108 			     req->r_tid, want_tid);
3109 			wait_for_completion(&req->r_safe_completion);
3110 			mutex_lock(&mdsc->mutex);
3111 			ceph_mdsc_put_request(req);
3112 			if (!nextreq)
3113 				break;  /* next dne before, so we're done! */
3114 			if (RB_EMPTY_NODE(&nextreq->r_node)) {
3115 				/* next request was removed from tree */
3116 				ceph_mdsc_put_request(nextreq);
3117 				goto restart;
3118 			}
3119 			ceph_mdsc_put_request(nextreq);  /* won't go away */
3120 		}
3121 		req = nextreq;
3122 	}
3123 	mutex_unlock(&mdsc->mutex);
3124 	dout("wait_unsafe_requests done\n");
3125 }
3126 
3127 void ceph_mdsc_sync(struct ceph_mds_client *mdsc)
3128 {
3129 	u64 want_tid, want_flush;
3130 
3131 	if (mdsc->fsc->mount_state == CEPH_MOUNT_SHUTDOWN)
3132 		return;
3133 
3134 	dout("sync\n");
3135 	mutex_lock(&mdsc->mutex);
3136 	want_tid = mdsc->last_tid;
3137 	want_flush = mdsc->cap_flush_seq;
3138 	mutex_unlock(&mdsc->mutex);
3139 	dout("sync want tid %lld flush_seq %lld\n", want_tid, want_flush);
3140 
3141 	ceph_flush_dirty_caps(mdsc);
3142 
3143 	wait_unsafe_requests(mdsc, want_tid);
3144 	wait_event(mdsc->cap_flushing_wq, check_cap_flush(mdsc, want_flush));
3145 }
3146 
3147 /*
3148  * true if all sessions are closed, or we force unmount
3149  */
3150 bool done_closing_sessions(struct ceph_mds_client *mdsc)
3151 {
3152 	int i, n = 0;
3153 
3154 	if (mdsc->fsc->mount_state == CEPH_MOUNT_SHUTDOWN)
3155 		return true;
3156 
3157 	mutex_lock(&mdsc->mutex);
3158 	for (i = 0; i < mdsc->max_sessions; i++)
3159 		if (mdsc->sessions[i])
3160 			n++;
3161 	mutex_unlock(&mdsc->mutex);
3162 	return n == 0;
3163 }
3164 
3165 /*
3166  * called after sb is ro.
3167  */
3168 void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc)
3169 {
3170 	struct ceph_mds_session *session;
3171 	int i;
3172 	struct ceph_fs_client *fsc = mdsc->fsc;
3173 	unsigned long timeout = fsc->client->options->mount_timeout * HZ;
3174 
3175 	dout("close_sessions\n");
3176 
3177 	/* close sessions */
3178 	mutex_lock(&mdsc->mutex);
3179 	for (i = 0; i < mdsc->max_sessions; i++) {
3180 		session = __ceph_lookup_mds_session(mdsc, i);
3181 		if (!session)
3182 			continue;
3183 		mutex_unlock(&mdsc->mutex);
3184 		mutex_lock(&session->s_mutex);
3185 		__close_session(mdsc, session);
3186 		mutex_unlock(&session->s_mutex);
3187 		ceph_put_mds_session(session);
3188 		mutex_lock(&mdsc->mutex);
3189 	}
3190 	mutex_unlock(&mdsc->mutex);
3191 
3192 	dout("waiting for sessions to close\n");
3193 	wait_event_timeout(mdsc->session_close_wq, done_closing_sessions(mdsc),
3194 			   timeout);
3195 
3196 	/* tear down remaining sessions */
3197 	mutex_lock(&mdsc->mutex);
3198 	for (i = 0; i < mdsc->max_sessions; i++) {
3199 		if (mdsc->sessions[i]) {
3200 			session = get_session(mdsc->sessions[i]);
3201 			__unregister_session(mdsc, session);
3202 			mutex_unlock(&mdsc->mutex);
3203 			mutex_lock(&session->s_mutex);
3204 			remove_session_caps(session);
3205 			mutex_unlock(&session->s_mutex);
3206 			ceph_put_mds_session(session);
3207 			mutex_lock(&mdsc->mutex);
3208 		}
3209 	}
3210 	WARN_ON(!list_empty(&mdsc->cap_delay_list));
3211 	mutex_unlock(&mdsc->mutex);
3212 
3213 	ceph_cleanup_empty_realms(mdsc);
3214 
3215 	cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
3216 
3217 	dout("stopped\n");
3218 }
3219 
3220 static void ceph_mdsc_stop(struct ceph_mds_client *mdsc)
3221 {
3222 	dout("stop\n");
3223 	cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
3224 	if (mdsc->mdsmap)
3225 		ceph_mdsmap_destroy(mdsc->mdsmap);
3226 	kfree(mdsc->sessions);
3227 	ceph_caps_finalize(mdsc);
3228 }
3229 
3230 void ceph_mdsc_destroy(struct ceph_fs_client *fsc)
3231 {
3232 	struct ceph_mds_client *mdsc = fsc->mdsc;
3233 
3234 	dout("mdsc_destroy %p\n", mdsc);
3235 	ceph_mdsc_stop(mdsc);
3236 
3237 	/* flush out any connection work with references to us */
3238 	ceph_msgr_flush();
3239 
3240 	fsc->mdsc = NULL;
3241 	kfree(mdsc);
3242 	dout("mdsc_destroy %p done\n", mdsc);
3243 }
3244 
3245 
3246 /*
3247  * handle mds map update.
3248  */
3249 void ceph_mdsc_handle_map(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
3250 {
3251 	u32 epoch;
3252 	u32 maplen;
3253 	void *p = msg->front.iov_base;
3254 	void *end = p + msg->front.iov_len;
3255 	struct ceph_mdsmap *newmap, *oldmap;
3256 	struct ceph_fsid fsid;
3257 	int err = -EINVAL;
3258 
3259 	ceph_decode_need(&p, end, sizeof(fsid)+2*sizeof(u32), bad);
3260 	ceph_decode_copy(&p, &fsid, sizeof(fsid));
3261 	if (ceph_check_fsid(mdsc->fsc->client, &fsid) < 0)
3262 		return;
3263 	epoch = ceph_decode_32(&p);
3264 	maplen = ceph_decode_32(&p);
3265 	dout("handle_map epoch %u len %d\n", epoch, (int)maplen);
3266 
3267 	/* do we need it? */
3268 	ceph_monc_got_mdsmap(&mdsc->fsc->client->monc, epoch);
3269 	mutex_lock(&mdsc->mutex);
3270 	if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) {
3271 		dout("handle_map epoch %u <= our %u\n",
3272 		     epoch, mdsc->mdsmap->m_epoch);
3273 		mutex_unlock(&mdsc->mutex);
3274 		return;
3275 	}
3276 
3277 	newmap = ceph_mdsmap_decode(&p, end);
3278 	if (IS_ERR(newmap)) {
3279 		err = PTR_ERR(newmap);
3280 		goto bad_unlock;
3281 	}
3282 
3283 	/* swap into place */
3284 	if (mdsc->mdsmap) {
3285 		oldmap = mdsc->mdsmap;
3286 		mdsc->mdsmap = newmap;
3287 		check_new_map(mdsc, newmap, oldmap);
3288 		ceph_mdsmap_destroy(oldmap);
3289 	} else {
3290 		mdsc->mdsmap = newmap;  /* first mds map */
3291 	}
3292 	mdsc->fsc->sb->s_maxbytes = mdsc->mdsmap->m_max_file_size;
3293 
3294 	__wake_requests(mdsc, &mdsc->waiting_for_map);
3295 
3296 	mutex_unlock(&mdsc->mutex);
3297 	schedule_delayed(mdsc);
3298 	return;
3299 
3300 bad_unlock:
3301 	mutex_unlock(&mdsc->mutex);
3302 bad:
3303 	pr_err("error decoding mdsmap %d\n", err);
3304 	return;
3305 }
3306 
3307 static struct ceph_connection *con_get(struct ceph_connection *con)
3308 {
3309 	struct ceph_mds_session *s = con->private;
3310 
3311 	if (get_session(s)) {
3312 		dout("mdsc con_get %p ok (%d)\n", s, atomic_read(&s->s_ref));
3313 		return con;
3314 	}
3315 	dout("mdsc con_get %p FAIL\n", s);
3316 	return NULL;
3317 }
3318 
3319 static void con_put(struct ceph_connection *con)
3320 {
3321 	struct ceph_mds_session *s = con->private;
3322 
3323 	dout("mdsc con_put %p (%d)\n", s, atomic_read(&s->s_ref) - 1);
3324 	ceph_put_mds_session(s);
3325 }
3326 
3327 /*
3328  * if the client is unresponsive for long enough, the mds will kill
3329  * the session entirely.
3330  */
3331 static void peer_reset(struct ceph_connection *con)
3332 {
3333 	struct ceph_mds_session *s = con->private;
3334 	struct ceph_mds_client *mdsc = s->s_mdsc;
3335 
3336 	pr_warning("mds%d closed our session\n", s->s_mds);
3337 	send_mds_reconnect(mdsc, s);
3338 }
3339 
3340 static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
3341 {
3342 	struct ceph_mds_session *s = con->private;
3343 	struct ceph_mds_client *mdsc = s->s_mdsc;
3344 	int type = le16_to_cpu(msg->hdr.type);
3345 
3346 	mutex_lock(&mdsc->mutex);
3347 	if (__verify_registered_session(mdsc, s) < 0) {
3348 		mutex_unlock(&mdsc->mutex);
3349 		goto out;
3350 	}
3351 	mutex_unlock(&mdsc->mutex);
3352 
3353 	switch (type) {
3354 	case CEPH_MSG_MDS_MAP:
3355 		ceph_mdsc_handle_map(mdsc, msg);
3356 		break;
3357 	case CEPH_MSG_CLIENT_SESSION:
3358 		handle_session(s, msg);
3359 		break;
3360 	case CEPH_MSG_CLIENT_REPLY:
3361 		handle_reply(s, msg);
3362 		break;
3363 	case CEPH_MSG_CLIENT_REQUEST_FORWARD:
3364 		handle_forward(mdsc, s, msg);
3365 		break;
3366 	case CEPH_MSG_CLIENT_CAPS:
3367 		ceph_handle_caps(s, msg);
3368 		break;
3369 	case CEPH_MSG_CLIENT_SNAP:
3370 		ceph_handle_snap(mdsc, s, msg);
3371 		break;
3372 	case CEPH_MSG_CLIENT_LEASE:
3373 		handle_lease(mdsc, s, msg);
3374 		break;
3375 
3376 	default:
3377 		pr_err("received unknown message type %d %s\n", type,
3378 		       ceph_msg_type_name(type));
3379 	}
3380 out:
3381 	ceph_msg_put(msg);
3382 }
3383 
3384 /*
3385  * authentication
3386  */
3387 static int get_authorizer(struct ceph_connection *con,
3388 			  void **buf, int *len, int *proto,
3389 			  void **reply_buf, int *reply_len, int force_new)
3390 {
3391 	struct ceph_mds_session *s = con->private;
3392 	struct ceph_mds_client *mdsc = s->s_mdsc;
3393 	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
3394 	int ret = 0;
3395 
3396 	if (force_new && s->s_authorizer) {
3397 		ac->ops->destroy_authorizer(ac, s->s_authorizer);
3398 		s->s_authorizer = NULL;
3399 	}
3400 	if (s->s_authorizer == NULL) {
3401 		if (ac->ops->create_authorizer) {
3402 			ret = ac->ops->create_authorizer(
3403 				ac, CEPH_ENTITY_TYPE_MDS,
3404 				&s->s_authorizer,
3405 				&s->s_authorizer_buf,
3406 				&s->s_authorizer_buf_len,
3407 				&s->s_authorizer_reply_buf,
3408 				&s->s_authorizer_reply_buf_len);
3409 			if (ret)
3410 				return ret;
3411 		}
3412 	}
3413 
3414 	*proto = ac->protocol;
3415 	*buf = s->s_authorizer_buf;
3416 	*len = s->s_authorizer_buf_len;
3417 	*reply_buf = s->s_authorizer_reply_buf;
3418 	*reply_len = s->s_authorizer_reply_buf_len;
3419 	return 0;
3420 }
3421 
3422 
3423 static int verify_authorizer_reply(struct ceph_connection *con, int len)
3424 {
3425 	struct ceph_mds_session *s = con->private;
3426 	struct ceph_mds_client *mdsc = s->s_mdsc;
3427 	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
3428 
3429 	return ac->ops->verify_authorizer_reply(ac, s->s_authorizer, len);
3430 }
3431 
3432 static int invalidate_authorizer(struct ceph_connection *con)
3433 {
3434 	struct ceph_mds_session *s = con->private;
3435 	struct ceph_mds_client *mdsc = s->s_mdsc;
3436 	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
3437 
3438 	if (ac->ops->invalidate_authorizer)
3439 		ac->ops->invalidate_authorizer(ac, CEPH_ENTITY_TYPE_MDS);
3440 
3441 	return ceph_monc_validate_auth(&mdsc->fsc->client->monc);
3442 }
3443 
3444 static const struct ceph_connection_operations mds_con_ops = {
3445 	.get = con_get,
3446 	.put = con_put,
3447 	.dispatch = dispatch,
3448 	.get_authorizer = get_authorizer,
3449 	.verify_authorizer_reply = verify_authorizer_reply,
3450 	.invalidate_authorizer = invalidate_authorizer,
3451 	.peer_reset = peer_reset,
3452 };
3453 
3454 /* eof */
3455