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