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