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