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