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 2104 put_request_session(req); 2105 2106 mds = __choose_mds(mdsc, req); 2107 if (mds < 0 || 2108 ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) { 2109 if (mdsc->mdsmap_err) { 2110 err = mdsc->mdsmap_err; 2111 dout("do_request mdsmap err %d\n", err); 2112 goto finish; 2113 } 2114 dout("do_request no mds or not active, waiting for map\n"); 2115 list_add(&req->r_wait, &mdsc->waiting_for_map); 2116 goto out; 2117 } 2118 2119 /* get, open session */ 2120 session = __ceph_lookup_mds_session(mdsc, mds); 2121 if (!session) { 2122 session = register_session(mdsc, mds); 2123 if (IS_ERR(session)) { 2124 err = PTR_ERR(session); 2125 goto finish; 2126 } 2127 } 2128 req->r_session = get_session(session); 2129 2130 dout("do_request mds%d session %p state %s\n", mds, session, 2131 ceph_session_state_name(session->s_state)); 2132 if (session->s_state != CEPH_MDS_SESSION_OPEN && 2133 session->s_state != CEPH_MDS_SESSION_HUNG) { 2134 if (session->s_state == CEPH_MDS_SESSION_REJECTED) { 2135 err = -EACCES; 2136 goto out_session; 2137 } 2138 if (session->s_state == CEPH_MDS_SESSION_NEW || 2139 session->s_state == CEPH_MDS_SESSION_CLOSING) 2140 __open_session(mdsc, session); 2141 list_add(&req->r_wait, &session->s_waiting); 2142 goto out_session; 2143 } 2144 2145 /* send request */ 2146 req->r_resend_mds = -1; /* forget any previous mds hint */ 2147 2148 if (req->r_request_started == 0) /* note request start time */ 2149 req->r_request_started = jiffies; 2150 2151 err = __prepare_send_request(mdsc, req, mds, false); 2152 if (!err) { 2153 ceph_msg_get(req->r_request); 2154 ceph_con_send(&session->s_con, req->r_request); 2155 } 2156 2157 out_session: 2158 ceph_put_mds_session(session); 2159 finish: 2160 if (err) { 2161 dout("__do_request early error %d\n", err); 2162 req->r_err = err; 2163 complete_request(mdsc, req); 2164 __unregister_request(mdsc, req); 2165 } 2166 out: 2167 return err; 2168 } 2169 2170 /* 2171 * called under mdsc->mutex 2172 */ 2173 static void __wake_requests(struct ceph_mds_client *mdsc, 2174 struct list_head *head) 2175 { 2176 struct ceph_mds_request *req; 2177 LIST_HEAD(tmp_list); 2178 2179 list_splice_init(head, &tmp_list); 2180 2181 while (!list_empty(&tmp_list)) { 2182 req = list_entry(tmp_list.next, 2183 struct ceph_mds_request, r_wait); 2184 list_del_init(&req->r_wait); 2185 dout(" wake request %p tid %llu\n", req, req->r_tid); 2186 __do_request(mdsc, req); 2187 } 2188 } 2189 2190 /* 2191 * Wake up threads with requests pending for @mds, so that they can 2192 * resubmit their requests to a possibly different mds. 2193 */ 2194 static void kick_requests(struct ceph_mds_client *mdsc, int mds) 2195 { 2196 struct ceph_mds_request *req; 2197 struct rb_node *p = rb_first(&mdsc->request_tree); 2198 2199 dout("kick_requests mds%d\n", mds); 2200 while (p) { 2201 req = rb_entry(p, struct ceph_mds_request, r_node); 2202 p = rb_next(p); 2203 if (req->r_got_unsafe) 2204 continue; 2205 if (req->r_attempts > 0) 2206 continue; /* only new requests */ 2207 if (req->r_session && 2208 req->r_session->s_mds == mds) { 2209 dout(" kicking tid %llu\n", req->r_tid); 2210 list_del_init(&req->r_wait); 2211 __do_request(mdsc, req); 2212 } 2213 } 2214 } 2215 2216 void ceph_mdsc_submit_request(struct ceph_mds_client *mdsc, 2217 struct ceph_mds_request *req) 2218 { 2219 dout("submit_request on %p\n", req); 2220 mutex_lock(&mdsc->mutex); 2221 __register_request(mdsc, req, NULL); 2222 __do_request(mdsc, req); 2223 mutex_unlock(&mdsc->mutex); 2224 } 2225 2226 /* 2227 * Synchrously perform an mds request. Take care of all of the 2228 * session setup, forwarding, retry details. 2229 */ 2230 int ceph_mdsc_do_request(struct ceph_mds_client *mdsc, 2231 struct inode *dir, 2232 struct ceph_mds_request *req) 2233 { 2234 int err; 2235 2236 dout("do_request on %p\n", req); 2237 2238 /* take CAP_PIN refs for r_inode, r_locked_dir, r_old_dentry */ 2239 if (req->r_inode) 2240 ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN); 2241 if (req->r_locked_dir) 2242 ceph_get_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN); 2243 if (req->r_old_dentry_dir) 2244 ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir), 2245 CEPH_CAP_PIN); 2246 2247 /* issue */ 2248 mutex_lock(&mdsc->mutex); 2249 __register_request(mdsc, req, dir); 2250 __do_request(mdsc, req); 2251 2252 if (req->r_err) { 2253 err = req->r_err; 2254 goto out; 2255 } 2256 2257 /* wait */ 2258 mutex_unlock(&mdsc->mutex); 2259 dout("do_request waiting\n"); 2260 if (!req->r_timeout && req->r_wait_for_completion) { 2261 err = req->r_wait_for_completion(mdsc, req); 2262 } else { 2263 long timeleft = wait_for_completion_killable_timeout( 2264 &req->r_completion, 2265 ceph_timeout_jiffies(req->r_timeout)); 2266 if (timeleft > 0) 2267 err = 0; 2268 else if (!timeleft) 2269 err = -EIO; /* timed out */ 2270 else 2271 err = timeleft; /* killed */ 2272 } 2273 dout("do_request waited, got %d\n", err); 2274 mutex_lock(&mdsc->mutex); 2275 2276 /* only abort if we didn't race with a real reply */ 2277 if (req->r_got_result) { 2278 err = le32_to_cpu(req->r_reply_info.head->result); 2279 } else if (err < 0) { 2280 dout("aborted request %lld with %d\n", req->r_tid, err); 2281 2282 /* 2283 * ensure we aren't running concurrently with 2284 * ceph_fill_trace or ceph_readdir_prepopulate, which 2285 * rely on locks (dir mutex) held by our caller. 2286 */ 2287 mutex_lock(&req->r_fill_mutex); 2288 req->r_err = err; 2289 req->r_aborted = true; 2290 mutex_unlock(&req->r_fill_mutex); 2291 2292 if (req->r_locked_dir && 2293 (req->r_op & CEPH_MDS_OP_WRITE)) 2294 ceph_invalidate_dir_request(req); 2295 } else { 2296 err = req->r_err; 2297 } 2298 2299 out: 2300 mutex_unlock(&mdsc->mutex); 2301 dout("do_request %p done, result %d\n", req, err); 2302 return err; 2303 } 2304 2305 /* 2306 * Invalidate dir's completeness, dentry lease state on an aborted MDS 2307 * namespace request. 2308 */ 2309 void ceph_invalidate_dir_request(struct ceph_mds_request *req) 2310 { 2311 struct inode *inode = req->r_locked_dir; 2312 2313 dout("invalidate_dir_request %p (complete, lease(s))\n", inode); 2314 2315 ceph_dir_clear_complete(inode); 2316 if (req->r_dentry) 2317 ceph_invalidate_dentry_lease(req->r_dentry); 2318 if (req->r_old_dentry) 2319 ceph_invalidate_dentry_lease(req->r_old_dentry); 2320 } 2321 2322 /* 2323 * Handle mds reply. 2324 * 2325 * We take the session mutex and parse and process the reply immediately. 2326 * This preserves the logical ordering of replies, capabilities, etc., sent 2327 * by the MDS as they are applied to our local cache. 2328 */ 2329 static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg) 2330 { 2331 struct ceph_mds_client *mdsc = session->s_mdsc; 2332 struct ceph_mds_request *req; 2333 struct ceph_mds_reply_head *head = msg->front.iov_base; 2334 struct ceph_mds_reply_info_parsed *rinfo; /* parsed reply info */ 2335 struct ceph_snap_realm *realm; 2336 u64 tid; 2337 int err, result; 2338 int mds = session->s_mds; 2339 2340 if (msg->front.iov_len < sizeof(*head)) { 2341 pr_err("mdsc_handle_reply got corrupt (short) reply\n"); 2342 ceph_msg_dump(msg); 2343 return; 2344 } 2345 2346 /* get request, session */ 2347 tid = le64_to_cpu(msg->hdr.tid); 2348 mutex_lock(&mdsc->mutex); 2349 req = lookup_get_request(mdsc, tid); 2350 if (!req) { 2351 dout("handle_reply on unknown tid %llu\n", tid); 2352 mutex_unlock(&mdsc->mutex); 2353 return; 2354 } 2355 dout("handle_reply %p\n", req); 2356 2357 /* correct session? */ 2358 if (req->r_session != session) { 2359 pr_err("mdsc_handle_reply got %llu on session mds%d" 2360 " not mds%d\n", tid, session->s_mds, 2361 req->r_session ? req->r_session->s_mds : -1); 2362 mutex_unlock(&mdsc->mutex); 2363 goto out; 2364 } 2365 2366 /* dup? */ 2367 if ((req->r_got_unsafe && !head->safe) || 2368 (req->r_got_safe && head->safe)) { 2369 pr_warn("got a dup %s reply on %llu from mds%d\n", 2370 head->safe ? "safe" : "unsafe", tid, mds); 2371 mutex_unlock(&mdsc->mutex); 2372 goto out; 2373 } 2374 if (req->r_got_safe) { 2375 pr_warn("got unsafe after safe on %llu from mds%d\n", 2376 tid, mds); 2377 mutex_unlock(&mdsc->mutex); 2378 goto out; 2379 } 2380 2381 result = le32_to_cpu(head->result); 2382 2383 /* 2384 * Handle an ESTALE 2385 * if we're not talking to the authority, send to them 2386 * if the authority has changed while we weren't looking, 2387 * send to new authority 2388 * Otherwise we just have to return an ESTALE 2389 */ 2390 if (result == -ESTALE) { 2391 dout("got ESTALE on request %llu", req->r_tid); 2392 req->r_resend_mds = -1; 2393 if (req->r_direct_mode != USE_AUTH_MDS) { 2394 dout("not using auth, setting for that now"); 2395 req->r_direct_mode = USE_AUTH_MDS; 2396 __do_request(mdsc, req); 2397 mutex_unlock(&mdsc->mutex); 2398 goto out; 2399 } else { 2400 int mds = __choose_mds(mdsc, req); 2401 if (mds >= 0 && mds != req->r_session->s_mds) { 2402 dout("but auth changed, so resending"); 2403 __do_request(mdsc, req); 2404 mutex_unlock(&mdsc->mutex); 2405 goto out; 2406 } 2407 } 2408 dout("have to return ESTALE on request %llu", req->r_tid); 2409 } 2410 2411 2412 if (head->safe) { 2413 req->r_got_safe = true; 2414 __unregister_request(mdsc, req); 2415 2416 if (req->r_got_unsafe) { 2417 /* 2418 * We already handled the unsafe response, now do the 2419 * cleanup. No need to examine the response; the MDS 2420 * doesn't include any result info in the safe 2421 * response. And even if it did, there is nothing 2422 * useful we could do with a revised return value. 2423 */ 2424 dout("got safe reply %llu, mds%d\n", tid, mds); 2425 list_del_init(&req->r_unsafe_item); 2426 2427 /* last unsafe request during umount? */ 2428 if (mdsc->stopping && !__get_oldest_req(mdsc)) 2429 complete_all(&mdsc->safe_umount_waiters); 2430 mutex_unlock(&mdsc->mutex); 2431 goto out; 2432 } 2433 } else { 2434 req->r_got_unsafe = true; 2435 list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe); 2436 if (req->r_unsafe_dir) { 2437 struct ceph_inode_info *ci = 2438 ceph_inode(req->r_unsafe_dir); 2439 spin_lock(&ci->i_unsafe_lock); 2440 list_add_tail(&req->r_unsafe_dir_item, 2441 &ci->i_unsafe_dirops); 2442 spin_unlock(&ci->i_unsafe_lock); 2443 } 2444 } 2445 2446 dout("handle_reply tid %lld result %d\n", tid, result); 2447 rinfo = &req->r_reply_info; 2448 err = parse_reply_info(msg, rinfo, session->s_con.peer_features); 2449 mutex_unlock(&mdsc->mutex); 2450 2451 mutex_lock(&session->s_mutex); 2452 if (err < 0) { 2453 pr_err("mdsc_handle_reply got corrupt reply mds%d(tid:%lld)\n", mds, tid); 2454 ceph_msg_dump(msg); 2455 goto out_err; 2456 } 2457 2458 /* snap trace */ 2459 realm = NULL; 2460 if (rinfo->snapblob_len) { 2461 down_write(&mdsc->snap_rwsem); 2462 ceph_update_snap_trace(mdsc, rinfo->snapblob, 2463 rinfo->snapblob + rinfo->snapblob_len, 2464 le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP, 2465 &realm); 2466 downgrade_write(&mdsc->snap_rwsem); 2467 } else { 2468 down_read(&mdsc->snap_rwsem); 2469 } 2470 2471 /* insert trace into our cache */ 2472 mutex_lock(&req->r_fill_mutex); 2473 current->journal_info = req; 2474 err = ceph_fill_trace(mdsc->fsc->sb, req, req->r_session); 2475 if (err == 0) { 2476 if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR || 2477 req->r_op == CEPH_MDS_OP_LSSNAP)) 2478 ceph_readdir_prepopulate(req, req->r_session); 2479 ceph_unreserve_caps(mdsc, &req->r_caps_reservation); 2480 } 2481 current->journal_info = NULL; 2482 mutex_unlock(&req->r_fill_mutex); 2483 2484 up_read(&mdsc->snap_rwsem); 2485 if (realm) 2486 ceph_put_snap_realm(mdsc, realm); 2487 2488 if (err == 0 && req->r_got_unsafe && req->r_target_inode) { 2489 struct ceph_inode_info *ci = ceph_inode(req->r_target_inode); 2490 spin_lock(&ci->i_unsafe_lock); 2491 list_add_tail(&req->r_unsafe_target_item, &ci->i_unsafe_iops); 2492 spin_unlock(&ci->i_unsafe_lock); 2493 } 2494 out_err: 2495 mutex_lock(&mdsc->mutex); 2496 if (!req->r_aborted) { 2497 if (err) { 2498 req->r_err = err; 2499 } else { 2500 req->r_reply = ceph_msg_get(msg); 2501 req->r_got_result = true; 2502 } 2503 } else { 2504 dout("reply arrived after request %lld was aborted\n", tid); 2505 } 2506 mutex_unlock(&mdsc->mutex); 2507 2508 mutex_unlock(&session->s_mutex); 2509 2510 /* kick calling process */ 2511 complete_request(mdsc, req); 2512 out: 2513 ceph_mdsc_put_request(req); 2514 return; 2515 } 2516 2517 2518 2519 /* 2520 * handle mds notification that our request has been forwarded. 2521 */ 2522 static void handle_forward(struct ceph_mds_client *mdsc, 2523 struct ceph_mds_session *session, 2524 struct ceph_msg *msg) 2525 { 2526 struct ceph_mds_request *req; 2527 u64 tid = le64_to_cpu(msg->hdr.tid); 2528 u32 next_mds; 2529 u32 fwd_seq; 2530 int err = -EINVAL; 2531 void *p = msg->front.iov_base; 2532 void *end = p + msg->front.iov_len; 2533 2534 ceph_decode_need(&p, end, 2*sizeof(u32), bad); 2535 next_mds = ceph_decode_32(&p); 2536 fwd_seq = ceph_decode_32(&p); 2537 2538 mutex_lock(&mdsc->mutex); 2539 req = lookup_get_request(mdsc, tid); 2540 if (!req) { 2541 dout("forward tid %llu to mds%d - req dne\n", tid, next_mds); 2542 goto out; /* dup reply? */ 2543 } 2544 2545 if (req->r_aborted) { 2546 dout("forward tid %llu aborted, unregistering\n", tid); 2547 __unregister_request(mdsc, req); 2548 } else if (fwd_seq <= req->r_num_fwd) { 2549 dout("forward tid %llu to mds%d - old seq %d <= %d\n", 2550 tid, next_mds, req->r_num_fwd, fwd_seq); 2551 } else { 2552 /* resend. forward race not possible; mds would drop */ 2553 dout("forward tid %llu to mds%d (we resend)\n", tid, next_mds); 2554 BUG_ON(req->r_err); 2555 BUG_ON(req->r_got_result); 2556 req->r_attempts = 0; 2557 req->r_num_fwd = fwd_seq; 2558 req->r_resend_mds = next_mds; 2559 put_request_session(req); 2560 __do_request(mdsc, req); 2561 } 2562 ceph_mdsc_put_request(req); 2563 out: 2564 mutex_unlock(&mdsc->mutex); 2565 return; 2566 2567 bad: 2568 pr_err("mdsc_handle_forward decode error err=%d\n", err); 2569 } 2570 2571 /* 2572 * handle a mds session control message 2573 */ 2574 static void handle_session(struct ceph_mds_session *session, 2575 struct ceph_msg *msg) 2576 { 2577 struct ceph_mds_client *mdsc = session->s_mdsc; 2578 u32 op; 2579 u64 seq; 2580 int mds = session->s_mds; 2581 struct ceph_mds_session_head *h = msg->front.iov_base; 2582 int wake = 0; 2583 2584 /* decode */ 2585 if (msg->front.iov_len != sizeof(*h)) 2586 goto bad; 2587 op = le32_to_cpu(h->op); 2588 seq = le64_to_cpu(h->seq); 2589 2590 mutex_lock(&mdsc->mutex); 2591 if (op == CEPH_SESSION_CLOSE) 2592 __unregister_session(mdsc, session); 2593 /* FIXME: this ttl calculation is generous */ 2594 session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose; 2595 mutex_unlock(&mdsc->mutex); 2596 2597 mutex_lock(&session->s_mutex); 2598 2599 dout("handle_session mds%d %s %p state %s seq %llu\n", 2600 mds, ceph_session_op_name(op), session, 2601 ceph_session_state_name(session->s_state), seq); 2602 2603 if (session->s_state == CEPH_MDS_SESSION_HUNG) { 2604 session->s_state = CEPH_MDS_SESSION_OPEN; 2605 pr_info("mds%d came back\n", session->s_mds); 2606 } 2607 2608 switch (op) { 2609 case CEPH_SESSION_OPEN: 2610 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING) 2611 pr_info("mds%d reconnect success\n", session->s_mds); 2612 session->s_state = CEPH_MDS_SESSION_OPEN; 2613 renewed_caps(mdsc, session, 0); 2614 wake = 1; 2615 if (mdsc->stopping) 2616 __close_session(mdsc, session); 2617 break; 2618 2619 case CEPH_SESSION_RENEWCAPS: 2620 if (session->s_renew_seq == seq) 2621 renewed_caps(mdsc, session, 1); 2622 break; 2623 2624 case CEPH_SESSION_CLOSE: 2625 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING) 2626 pr_info("mds%d reconnect denied\n", session->s_mds); 2627 cleanup_session_requests(mdsc, session); 2628 remove_session_caps(session); 2629 wake = 2; /* for good measure */ 2630 wake_up_all(&mdsc->session_close_wq); 2631 break; 2632 2633 case CEPH_SESSION_STALE: 2634 pr_info("mds%d caps went stale, renewing\n", 2635 session->s_mds); 2636 spin_lock(&session->s_gen_ttl_lock); 2637 session->s_cap_gen++; 2638 session->s_cap_ttl = jiffies - 1; 2639 spin_unlock(&session->s_gen_ttl_lock); 2640 send_renew_caps(mdsc, session); 2641 break; 2642 2643 case CEPH_SESSION_RECALL_STATE: 2644 trim_caps(mdsc, session, le32_to_cpu(h->max_caps)); 2645 break; 2646 2647 case CEPH_SESSION_FLUSHMSG: 2648 send_flushmsg_ack(mdsc, session, seq); 2649 break; 2650 2651 case CEPH_SESSION_FORCE_RO: 2652 dout("force_session_readonly %p\n", session); 2653 spin_lock(&session->s_cap_lock); 2654 session->s_readonly = true; 2655 spin_unlock(&session->s_cap_lock); 2656 wake_up_session_caps(session, 0); 2657 break; 2658 2659 case CEPH_SESSION_REJECT: 2660 WARN_ON(session->s_state != CEPH_MDS_SESSION_OPENING); 2661 pr_info("mds%d rejected session\n", session->s_mds); 2662 session->s_state = CEPH_MDS_SESSION_REJECTED; 2663 cleanup_session_requests(mdsc, session); 2664 remove_session_caps(session); 2665 wake = 2; /* for good measure */ 2666 break; 2667 2668 default: 2669 pr_err("mdsc_handle_session bad op %d mds%d\n", op, mds); 2670 WARN_ON(1); 2671 } 2672 2673 mutex_unlock(&session->s_mutex); 2674 if (wake) { 2675 mutex_lock(&mdsc->mutex); 2676 __wake_requests(mdsc, &session->s_waiting); 2677 if (wake == 2) 2678 kick_requests(mdsc, mds); 2679 mutex_unlock(&mdsc->mutex); 2680 } 2681 return; 2682 2683 bad: 2684 pr_err("mdsc_handle_session corrupt message mds%d len %d\n", mds, 2685 (int)msg->front.iov_len); 2686 ceph_msg_dump(msg); 2687 return; 2688 } 2689 2690 2691 /* 2692 * called under session->mutex. 2693 */ 2694 static void replay_unsafe_requests(struct ceph_mds_client *mdsc, 2695 struct ceph_mds_session *session) 2696 { 2697 struct ceph_mds_request *req, *nreq; 2698 struct rb_node *p; 2699 int err; 2700 2701 dout("replay_unsafe_requests mds%d\n", session->s_mds); 2702 2703 mutex_lock(&mdsc->mutex); 2704 list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item) { 2705 err = __prepare_send_request(mdsc, req, session->s_mds, true); 2706 if (!err) { 2707 ceph_msg_get(req->r_request); 2708 ceph_con_send(&session->s_con, req->r_request); 2709 } 2710 } 2711 2712 /* 2713 * also re-send old requests when MDS enters reconnect stage. So that MDS 2714 * can process completed request in clientreplay stage. 2715 */ 2716 p = rb_first(&mdsc->request_tree); 2717 while (p) { 2718 req = rb_entry(p, struct ceph_mds_request, r_node); 2719 p = rb_next(p); 2720 if (req->r_got_unsafe) 2721 continue; 2722 if (req->r_attempts == 0) 2723 continue; /* only old requests */ 2724 if (req->r_session && 2725 req->r_session->s_mds == session->s_mds) { 2726 err = __prepare_send_request(mdsc, req, 2727 session->s_mds, true); 2728 if (!err) { 2729 ceph_msg_get(req->r_request); 2730 ceph_con_send(&session->s_con, req->r_request); 2731 } 2732 } 2733 } 2734 mutex_unlock(&mdsc->mutex); 2735 } 2736 2737 /* 2738 * Encode information about a cap for a reconnect with the MDS. 2739 */ 2740 static int encode_caps_cb(struct inode *inode, struct ceph_cap *cap, 2741 void *arg) 2742 { 2743 union { 2744 struct ceph_mds_cap_reconnect v2; 2745 struct ceph_mds_cap_reconnect_v1 v1; 2746 } rec; 2747 struct ceph_inode_info *ci; 2748 struct ceph_reconnect_state *recon_state = arg; 2749 struct ceph_pagelist *pagelist = recon_state->pagelist; 2750 char *path; 2751 int pathlen, err; 2752 u64 pathbase; 2753 u64 snap_follows; 2754 struct dentry *dentry; 2755 2756 ci = cap->ci; 2757 2758 dout(" adding %p ino %llx.%llx cap %p %lld %s\n", 2759 inode, ceph_vinop(inode), cap, cap->cap_id, 2760 ceph_cap_string(cap->issued)); 2761 err = ceph_pagelist_encode_64(pagelist, ceph_ino(inode)); 2762 if (err) 2763 return err; 2764 2765 dentry = d_find_alias(inode); 2766 if (dentry) { 2767 path = ceph_mdsc_build_path(dentry, &pathlen, &pathbase, 0); 2768 if (IS_ERR(path)) { 2769 err = PTR_ERR(path); 2770 goto out_dput; 2771 } 2772 } else { 2773 path = NULL; 2774 pathlen = 0; 2775 pathbase = 0; 2776 } 2777 2778 spin_lock(&ci->i_ceph_lock); 2779 cap->seq = 0; /* reset cap seq */ 2780 cap->issue_seq = 0; /* and issue_seq */ 2781 cap->mseq = 0; /* and migrate_seq */ 2782 cap->cap_gen = cap->session->s_cap_gen; 2783 2784 if (recon_state->msg_version >= 2) { 2785 rec.v2.cap_id = cpu_to_le64(cap->cap_id); 2786 rec.v2.wanted = cpu_to_le32(__ceph_caps_wanted(ci)); 2787 rec.v2.issued = cpu_to_le32(cap->issued); 2788 rec.v2.snaprealm = cpu_to_le64(ci->i_snap_realm->ino); 2789 rec.v2.pathbase = cpu_to_le64(pathbase); 2790 rec.v2.flock_len = 0; 2791 } else { 2792 rec.v1.cap_id = cpu_to_le64(cap->cap_id); 2793 rec.v1.wanted = cpu_to_le32(__ceph_caps_wanted(ci)); 2794 rec.v1.issued = cpu_to_le32(cap->issued); 2795 rec.v1.size = cpu_to_le64(inode->i_size); 2796 ceph_encode_timespec(&rec.v1.mtime, &inode->i_mtime); 2797 ceph_encode_timespec(&rec.v1.atime, &inode->i_atime); 2798 rec.v1.snaprealm = cpu_to_le64(ci->i_snap_realm->ino); 2799 rec.v1.pathbase = cpu_to_le64(pathbase); 2800 } 2801 2802 if (list_empty(&ci->i_cap_snaps)) { 2803 snap_follows = 0; 2804 } else { 2805 struct ceph_cap_snap *capsnap = 2806 list_first_entry(&ci->i_cap_snaps, 2807 struct ceph_cap_snap, ci_item); 2808 snap_follows = capsnap->follows; 2809 } 2810 spin_unlock(&ci->i_ceph_lock); 2811 2812 if (recon_state->msg_version >= 2) { 2813 int num_fcntl_locks, num_flock_locks; 2814 struct ceph_filelock *flocks; 2815 size_t struct_len, total_len = 0; 2816 u8 struct_v = 0; 2817 2818 encode_again: 2819 ceph_count_locks(inode, &num_fcntl_locks, &num_flock_locks); 2820 flocks = kmalloc((num_fcntl_locks+num_flock_locks) * 2821 sizeof(struct ceph_filelock), GFP_NOFS); 2822 if (!flocks) { 2823 err = -ENOMEM; 2824 goto out_free; 2825 } 2826 err = ceph_encode_locks_to_buffer(inode, flocks, 2827 num_fcntl_locks, 2828 num_flock_locks); 2829 if (err) { 2830 kfree(flocks); 2831 if (err == -ENOSPC) 2832 goto encode_again; 2833 goto out_free; 2834 } 2835 2836 if (recon_state->msg_version >= 3) { 2837 /* version, compat_version and struct_len */ 2838 total_len = 2 * sizeof(u8) + sizeof(u32); 2839 struct_v = 2; 2840 } 2841 /* 2842 * number of encoded locks is stable, so copy to pagelist 2843 */ 2844 struct_len = 2 * sizeof(u32) + 2845 (num_fcntl_locks + num_flock_locks) * 2846 sizeof(struct ceph_filelock); 2847 rec.v2.flock_len = cpu_to_le32(struct_len); 2848 2849 struct_len += sizeof(rec.v2); 2850 struct_len += sizeof(u32) + pathlen; 2851 2852 if (struct_v >= 2) 2853 struct_len += sizeof(u64); /* snap_follows */ 2854 2855 total_len += struct_len; 2856 err = ceph_pagelist_reserve(pagelist, total_len); 2857 2858 if (!err) { 2859 if (recon_state->msg_version >= 3) { 2860 ceph_pagelist_encode_8(pagelist, struct_v); 2861 ceph_pagelist_encode_8(pagelist, 1); 2862 ceph_pagelist_encode_32(pagelist, struct_len); 2863 } 2864 ceph_pagelist_encode_string(pagelist, path, pathlen); 2865 ceph_pagelist_append(pagelist, &rec, sizeof(rec.v2)); 2866 ceph_locks_to_pagelist(flocks, pagelist, 2867 num_fcntl_locks, 2868 num_flock_locks); 2869 if (struct_v >= 2) 2870 ceph_pagelist_encode_64(pagelist, snap_follows); 2871 } 2872 kfree(flocks); 2873 } else { 2874 size_t size = sizeof(u32) + pathlen + sizeof(rec.v1); 2875 err = ceph_pagelist_reserve(pagelist, size); 2876 if (!err) { 2877 ceph_pagelist_encode_string(pagelist, path, pathlen); 2878 ceph_pagelist_append(pagelist, &rec, sizeof(rec.v1)); 2879 } 2880 } 2881 2882 recon_state->nr_caps++; 2883 out_free: 2884 kfree(path); 2885 out_dput: 2886 dput(dentry); 2887 return err; 2888 } 2889 2890 2891 /* 2892 * If an MDS fails and recovers, clients need to reconnect in order to 2893 * reestablish shared state. This includes all caps issued through 2894 * this session _and_ the snap_realm hierarchy. Because it's not 2895 * clear which snap realms the mds cares about, we send everything we 2896 * know about.. that ensures we'll then get any new info the 2897 * recovering MDS might have. 2898 * 2899 * This is a relatively heavyweight operation, but it's rare. 2900 * 2901 * called with mdsc->mutex held. 2902 */ 2903 static void send_mds_reconnect(struct ceph_mds_client *mdsc, 2904 struct ceph_mds_session *session) 2905 { 2906 struct ceph_msg *reply; 2907 struct rb_node *p; 2908 int mds = session->s_mds; 2909 int err = -ENOMEM; 2910 int s_nr_caps; 2911 struct ceph_pagelist *pagelist; 2912 struct ceph_reconnect_state recon_state; 2913 2914 pr_info("mds%d reconnect start\n", mds); 2915 2916 pagelist = kmalloc(sizeof(*pagelist), GFP_NOFS); 2917 if (!pagelist) 2918 goto fail_nopagelist; 2919 ceph_pagelist_init(pagelist); 2920 2921 reply = ceph_msg_new(CEPH_MSG_CLIENT_RECONNECT, 0, GFP_NOFS, false); 2922 if (!reply) 2923 goto fail_nomsg; 2924 2925 mutex_lock(&session->s_mutex); 2926 session->s_state = CEPH_MDS_SESSION_RECONNECTING; 2927 session->s_seq = 0; 2928 2929 dout("session %p state %s\n", session, 2930 ceph_session_state_name(session->s_state)); 2931 2932 spin_lock(&session->s_gen_ttl_lock); 2933 session->s_cap_gen++; 2934 spin_unlock(&session->s_gen_ttl_lock); 2935 2936 spin_lock(&session->s_cap_lock); 2937 /* don't know if session is readonly */ 2938 session->s_readonly = 0; 2939 /* 2940 * notify __ceph_remove_cap() that we are composing cap reconnect. 2941 * If a cap get released before being added to the cap reconnect, 2942 * __ceph_remove_cap() should skip queuing cap release. 2943 */ 2944 session->s_cap_reconnect = 1; 2945 /* drop old cap expires; we're about to reestablish that state */ 2946 cleanup_cap_releases(mdsc, session); 2947 2948 /* trim unused caps to reduce MDS's cache rejoin time */ 2949 if (mdsc->fsc->sb->s_root) 2950 shrink_dcache_parent(mdsc->fsc->sb->s_root); 2951 2952 ceph_con_close(&session->s_con); 2953 ceph_con_open(&session->s_con, 2954 CEPH_ENTITY_TYPE_MDS, mds, 2955 ceph_mdsmap_get_addr(mdsc->mdsmap, mds)); 2956 2957 /* replay unsafe requests */ 2958 replay_unsafe_requests(mdsc, session); 2959 2960 down_read(&mdsc->snap_rwsem); 2961 2962 /* traverse this session's caps */ 2963 s_nr_caps = session->s_nr_caps; 2964 err = ceph_pagelist_encode_32(pagelist, s_nr_caps); 2965 if (err) 2966 goto fail; 2967 2968 recon_state.nr_caps = 0; 2969 recon_state.pagelist = pagelist; 2970 if (session->s_con.peer_features & CEPH_FEATURE_MDSENC) 2971 recon_state.msg_version = 3; 2972 else if (session->s_con.peer_features & CEPH_FEATURE_FLOCK) 2973 recon_state.msg_version = 2; 2974 else 2975 recon_state.msg_version = 1; 2976 err = iterate_session_caps(session, encode_caps_cb, &recon_state); 2977 if (err < 0) 2978 goto fail; 2979 2980 spin_lock(&session->s_cap_lock); 2981 session->s_cap_reconnect = 0; 2982 spin_unlock(&session->s_cap_lock); 2983 2984 /* 2985 * snaprealms. we provide mds with the ino, seq (version), and 2986 * parent for all of our realms. If the mds has any newer info, 2987 * it will tell us. 2988 */ 2989 for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) { 2990 struct ceph_snap_realm *realm = 2991 rb_entry(p, struct ceph_snap_realm, node); 2992 struct ceph_mds_snaprealm_reconnect sr_rec; 2993 2994 dout(" adding snap realm %llx seq %lld parent %llx\n", 2995 realm->ino, realm->seq, realm->parent_ino); 2996 sr_rec.ino = cpu_to_le64(realm->ino); 2997 sr_rec.seq = cpu_to_le64(realm->seq); 2998 sr_rec.parent = cpu_to_le64(realm->parent_ino); 2999 err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec)); 3000 if (err) 3001 goto fail; 3002 } 3003 3004 reply->hdr.version = cpu_to_le16(recon_state.msg_version); 3005 3006 /* raced with cap release? */ 3007 if (s_nr_caps != recon_state.nr_caps) { 3008 struct page *page = list_first_entry(&pagelist->head, 3009 struct page, lru); 3010 __le32 *addr = kmap_atomic(page); 3011 *addr = cpu_to_le32(recon_state.nr_caps); 3012 kunmap_atomic(addr); 3013 } 3014 3015 reply->hdr.data_len = cpu_to_le32(pagelist->length); 3016 ceph_msg_data_add_pagelist(reply, pagelist); 3017 3018 ceph_early_kick_flushing_caps(mdsc, session); 3019 3020 ceph_con_send(&session->s_con, reply); 3021 3022 mutex_unlock(&session->s_mutex); 3023 3024 mutex_lock(&mdsc->mutex); 3025 __wake_requests(mdsc, &session->s_waiting); 3026 mutex_unlock(&mdsc->mutex); 3027 3028 up_read(&mdsc->snap_rwsem); 3029 return; 3030 3031 fail: 3032 ceph_msg_put(reply); 3033 up_read(&mdsc->snap_rwsem); 3034 mutex_unlock(&session->s_mutex); 3035 fail_nomsg: 3036 ceph_pagelist_release(pagelist); 3037 fail_nopagelist: 3038 pr_err("error %d preparing reconnect for mds%d\n", err, mds); 3039 return; 3040 } 3041 3042 3043 /* 3044 * compare old and new mdsmaps, kicking requests 3045 * and closing out old connections as necessary 3046 * 3047 * called under mdsc->mutex. 3048 */ 3049 static void check_new_map(struct ceph_mds_client *mdsc, 3050 struct ceph_mdsmap *newmap, 3051 struct ceph_mdsmap *oldmap) 3052 { 3053 int i; 3054 int oldstate, newstate; 3055 struct ceph_mds_session *s; 3056 3057 dout("check_new_map new %u old %u\n", 3058 newmap->m_epoch, oldmap->m_epoch); 3059 3060 for (i = 0; i < oldmap->m_max_mds && i < mdsc->max_sessions; i++) { 3061 if (mdsc->sessions[i] == NULL) 3062 continue; 3063 s = mdsc->sessions[i]; 3064 oldstate = ceph_mdsmap_get_state(oldmap, i); 3065 newstate = ceph_mdsmap_get_state(newmap, i); 3066 3067 dout("check_new_map mds%d state %s%s -> %s%s (session %s)\n", 3068 i, ceph_mds_state_name(oldstate), 3069 ceph_mdsmap_is_laggy(oldmap, i) ? " (laggy)" : "", 3070 ceph_mds_state_name(newstate), 3071 ceph_mdsmap_is_laggy(newmap, i) ? " (laggy)" : "", 3072 ceph_session_state_name(s->s_state)); 3073 3074 if (i >= newmap->m_max_mds || 3075 memcmp(ceph_mdsmap_get_addr(oldmap, i), 3076 ceph_mdsmap_get_addr(newmap, i), 3077 sizeof(struct ceph_entity_addr))) { 3078 if (s->s_state == CEPH_MDS_SESSION_OPENING) { 3079 /* the session never opened, just close it 3080 * out now */ 3081 __wake_requests(mdsc, &s->s_waiting); 3082 __unregister_session(mdsc, s); 3083 } else { 3084 /* just close it */ 3085 mutex_unlock(&mdsc->mutex); 3086 mutex_lock(&s->s_mutex); 3087 mutex_lock(&mdsc->mutex); 3088 ceph_con_close(&s->s_con); 3089 mutex_unlock(&s->s_mutex); 3090 s->s_state = CEPH_MDS_SESSION_RESTARTING; 3091 } 3092 } else if (oldstate == newstate) { 3093 continue; /* nothing new with this mds */ 3094 } 3095 3096 /* 3097 * send reconnect? 3098 */ 3099 if (s->s_state == CEPH_MDS_SESSION_RESTARTING && 3100 newstate >= CEPH_MDS_STATE_RECONNECT) { 3101 mutex_unlock(&mdsc->mutex); 3102 send_mds_reconnect(mdsc, s); 3103 mutex_lock(&mdsc->mutex); 3104 } 3105 3106 /* 3107 * kick request on any mds that has gone active. 3108 */ 3109 if (oldstate < CEPH_MDS_STATE_ACTIVE && 3110 newstate >= CEPH_MDS_STATE_ACTIVE) { 3111 if (oldstate != CEPH_MDS_STATE_CREATING && 3112 oldstate != CEPH_MDS_STATE_STARTING) 3113 pr_info("mds%d recovery completed\n", s->s_mds); 3114 kick_requests(mdsc, i); 3115 ceph_kick_flushing_caps(mdsc, s); 3116 wake_up_session_caps(s, 1); 3117 } 3118 } 3119 3120 for (i = 0; i < newmap->m_max_mds && i < mdsc->max_sessions; i++) { 3121 s = mdsc->sessions[i]; 3122 if (!s) 3123 continue; 3124 if (!ceph_mdsmap_is_laggy(newmap, i)) 3125 continue; 3126 if (s->s_state == CEPH_MDS_SESSION_OPEN || 3127 s->s_state == CEPH_MDS_SESSION_HUNG || 3128 s->s_state == CEPH_MDS_SESSION_CLOSING) { 3129 dout(" connecting to export targets of laggy mds%d\n", 3130 i); 3131 __open_export_target_sessions(mdsc, s); 3132 } 3133 } 3134 } 3135 3136 3137 3138 /* 3139 * leases 3140 */ 3141 3142 /* 3143 * caller must hold session s_mutex, dentry->d_lock 3144 */ 3145 void __ceph_mdsc_drop_dentry_lease(struct dentry *dentry) 3146 { 3147 struct ceph_dentry_info *di = ceph_dentry(dentry); 3148 3149 ceph_put_mds_session(di->lease_session); 3150 di->lease_session = NULL; 3151 } 3152 3153 static void handle_lease(struct ceph_mds_client *mdsc, 3154 struct ceph_mds_session *session, 3155 struct ceph_msg *msg) 3156 { 3157 struct super_block *sb = mdsc->fsc->sb; 3158 struct inode *inode; 3159 struct dentry *parent, *dentry; 3160 struct ceph_dentry_info *di; 3161 int mds = session->s_mds; 3162 struct ceph_mds_lease *h = msg->front.iov_base; 3163 u32 seq; 3164 struct ceph_vino vino; 3165 struct qstr dname; 3166 int release = 0; 3167 3168 dout("handle_lease from mds%d\n", mds); 3169 3170 /* decode */ 3171 if (msg->front.iov_len < sizeof(*h) + sizeof(u32)) 3172 goto bad; 3173 vino.ino = le64_to_cpu(h->ino); 3174 vino.snap = CEPH_NOSNAP; 3175 seq = le32_to_cpu(h->seq); 3176 dname.name = (void *)h + sizeof(*h) + sizeof(u32); 3177 dname.len = msg->front.iov_len - sizeof(*h) - sizeof(u32); 3178 if (dname.len != get_unaligned_le32(h+1)) 3179 goto bad; 3180 3181 /* lookup inode */ 3182 inode = ceph_find_inode(sb, vino); 3183 dout("handle_lease %s, ino %llx %p %.*s\n", 3184 ceph_lease_op_name(h->action), vino.ino, inode, 3185 dname.len, dname.name); 3186 3187 mutex_lock(&session->s_mutex); 3188 session->s_seq++; 3189 3190 if (inode == NULL) { 3191 dout("handle_lease no inode %llx\n", vino.ino); 3192 goto release; 3193 } 3194 3195 /* dentry */ 3196 parent = d_find_alias(inode); 3197 if (!parent) { 3198 dout("no parent dentry on inode %p\n", inode); 3199 WARN_ON(1); 3200 goto release; /* hrm... */ 3201 } 3202 dname.hash = full_name_hash(parent, dname.name, dname.len); 3203 dentry = d_lookup(parent, &dname); 3204 dput(parent); 3205 if (!dentry) 3206 goto release; 3207 3208 spin_lock(&dentry->d_lock); 3209 di = ceph_dentry(dentry); 3210 switch (h->action) { 3211 case CEPH_MDS_LEASE_REVOKE: 3212 if (di->lease_session == session) { 3213 if (ceph_seq_cmp(di->lease_seq, seq) > 0) 3214 h->seq = cpu_to_le32(di->lease_seq); 3215 __ceph_mdsc_drop_dentry_lease(dentry); 3216 } 3217 release = 1; 3218 break; 3219 3220 case CEPH_MDS_LEASE_RENEW: 3221 if (di->lease_session == session && 3222 di->lease_gen == session->s_cap_gen && 3223 di->lease_renew_from && 3224 di->lease_renew_after == 0) { 3225 unsigned long duration = 3226 msecs_to_jiffies(le32_to_cpu(h->duration_ms)); 3227 3228 di->lease_seq = seq; 3229 di->time = di->lease_renew_from + duration; 3230 di->lease_renew_after = di->lease_renew_from + 3231 (duration >> 1); 3232 di->lease_renew_from = 0; 3233 } 3234 break; 3235 } 3236 spin_unlock(&dentry->d_lock); 3237 dput(dentry); 3238 3239 if (!release) 3240 goto out; 3241 3242 release: 3243 /* let's just reuse the same message */ 3244 h->action = CEPH_MDS_LEASE_REVOKE_ACK; 3245 ceph_msg_get(msg); 3246 ceph_con_send(&session->s_con, msg); 3247 3248 out: 3249 iput(inode); 3250 mutex_unlock(&session->s_mutex); 3251 return; 3252 3253 bad: 3254 pr_err("corrupt lease message\n"); 3255 ceph_msg_dump(msg); 3256 } 3257 3258 void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session, 3259 struct inode *inode, 3260 struct dentry *dentry, char action, 3261 u32 seq) 3262 { 3263 struct ceph_msg *msg; 3264 struct ceph_mds_lease *lease; 3265 int len = sizeof(*lease) + sizeof(u32); 3266 int dnamelen = 0; 3267 3268 dout("lease_send_msg inode %p dentry %p %s to mds%d\n", 3269 inode, dentry, ceph_lease_op_name(action), session->s_mds); 3270 dnamelen = dentry->d_name.len; 3271 len += dnamelen; 3272 3273 msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS, false); 3274 if (!msg) 3275 return; 3276 lease = msg->front.iov_base; 3277 lease->action = action; 3278 lease->ino = cpu_to_le64(ceph_vino(inode).ino); 3279 lease->first = lease->last = cpu_to_le64(ceph_vino(inode).snap); 3280 lease->seq = cpu_to_le32(seq); 3281 put_unaligned_le32(dnamelen, lease + 1); 3282 memcpy((void *)(lease + 1) + 4, dentry->d_name.name, dnamelen); 3283 3284 /* 3285 * if this is a preemptive lease RELEASE, no need to 3286 * flush request stream, since the actual request will 3287 * soon follow. 3288 */ 3289 msg->more_to_follow = (action == CEPH_MDS_LEASE_RELEASE); 3290 3291 ceph_con_send(&session->s_con, msg); 3292 } 3293 3294 /* 3295 * drop all leases (and dentry refs) in preparation for umount 3296 */ 3297 static void drop_leases(struct ceph_mds_client *mdsc) 3298 { 3299 int i; 3300 3301 dout("drop_leases\n"); 3302 mutex_lock(&mdsc->mutex); 3303 for (i = 0; i < mdsc->max_sessions; i++) { 3304 struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i); 3305 if (!s) 3306 continue; 3307 mutex_unlock(&mdsc->mutex); 3308 mutex_lock(&s->s_mutex); 3309 mutex_unlock(&s->s_mutex); 3310 ceph_put_mds_session(s); 3311 mutex_lock(&mdsc->mutex); 3312 } 3313 mutex_unlock(&mdsc->mutex); 3314 } 3315 3316 3317 3318 /* 3319 * delayed work -- periodically trim expired leases, renew caps with mds 3320 */ 3321 static void schedule_delayed(struct ceph_mds_client *mdsc) 3322 { 3323 int delay = 5; 3324 unsigned hz = round_jiffies_relative(HZ * delay); 3325 schedule_delayed_work(&mdsc->delayed_work, hz); 3326 } 3327 3328 static void delayed_work(struct work_struct *work) 3329 { 3330 int i; 3331 struct ceph_mds_client *mdsc = 3332 container_of(work, struct ceph_mds_client, delayed_work.work); 3333 int renew_interval; 3334 int renew_caps; 3335 3336 dout("mdsc delayed_work\n"); 3337 ceph_check_delayed_caps(mdsc); 3338 3339 mutex_lock(&mdsc->mutex); 3340 renew_interval = mdsc->mdsmap->m_session_timeout >> 2; 3341 renew_caps = time_after_eq(jiffies, HZ*renew_interval + 3342 mdsc->last_renew_caps); 3343 if (renew_caps) 3344 mdsc->last_renew_caps = jiffies; 3345 3346 for (i = 0; i < mdsc->max_sessions; i++) { 3347 struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i); 3348 if (s == NULL) 3349 continue; 3350 if (s->s_state == CEPH_MDS_SESSION_CLOSING) { 3351 dout("resending session close request for mds%d\n", 3352 s->s_mds); 3353 request_close_session(mdsc, s); 3354 ceph_put_mds_session(s); 3355 continue; 3356 } 3357 if (s->s_ttl && time_after(jiffies, s->s_ttl)) { 3358 if (s->s_state == CEPH_MDS_SESSION_OPEN) { 3359 s->s_state = CEPH_MDS_SESSION_HUNG; 3360 pr_info("mds%d hung\n", s->s_mds); 3361 } 3362 } 3363 if (s->s_state < CEPH_MDS_SESSION_OPEN) { 3364 /* this mds is failed or recovering, just wait */ 3365 ceph_put_mds_session(s); 3366 continue; 3367 } 3368 mutex_unlock(&mdsc->mutex); 3369 3370 mutex_lock(&s->s_mutex); 3371 if (renew_caps) 3372 send_renew_caps(mdsc, s); 3373 else 3374 ceph_con_keepalive(&s->s_con); 3375 if (s->s_state == CEPH_MDS_SESSION_OPEN || 3376 s->s_state == CEPH_MDS_SESSION_HUNG) 3377 ceph_send_cap_releases(mdsc, s); 3378 mutex_unlock(&s->s_mutex); 3379 ceph_put_mds_session(s); 3380 3381 mutex_lock(&mdsc->mutex); 3382 } 3383 mutex_unlock(&mdsc->mutex); 3384 3385 schedule_delayed(mdsc); 3386 } 3387 3388 int ceph_mdsc_init(struct ceph_fs_client *fsc) 3389 3390 { 3391 struct ceph_mds_client *mdsc; 3392 3393 mdsc = kzalloc(sizeof(struct ceph_mds_client), GFP_NOFS); 3394 if (!mdsc) 3395 return -ENOMEM; 3396 mdsc->fsc = fsc; 3397 fsc->mdsc = mdsc; 3398 mutex_init(&mdsc->mutex); 3399 mdsc->mdsmap = kzalloc(sizeof(*mdsc->mdsmap), GFP_NOFS); 3400 if (mdsc->mdsmap == NULL) { 3401 kfree(mdsc); 3402 return -ENOMEM; 3403 } 3404 3405 init_completion(&mdsc->safe_umount_waiters); 3406 init_waitqueue_head(&mdsc->session_close_wq); 3407 INIT_LIST_HEAD(&mdsc->waiting_for_map); 3408 mdsc->sessions = NULL; 3409 atomic_set(&mdsc->num_sessions, 0); 3410 mdsc->max_sessions = 0; 3411 mdsc->stopping = 0; 3412 mdsc->last_snap_seq = 0; 3413 init_rwsem(&mdsc->snap_rwsem); 3414 mdsc->snap_realms = RB_ROOT; 3415 INIT_LIST_HEAD(&mdsc->snap_empty); 3416 spin_lock_init(&mdsc->snap_empty_lock); 3417 mdsc->last_tid = 0; 3418 mdsc->oldest_tid = 0; 3419 mdsc->request_tree = RB_ROOT; 3420 INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work); 3421 mdsc->last_renew_caps = jiffies; 3422 INIT_LIST_HEAD(&mdsc->cap_delay_list); 3423 spin_lock_init(&mdsc->cap_delay_lock); 3424 INIT_LIST_HEAD(&mdsc->snap_flush_list); 3425 spin_lock_init(&mdsc->snap_flush_lock); 3426 mdsc->last_cap_flush_tid = 1; 3427 INIT_LIST_HEAD(&mdsc->cap_flush_list); 3428 INIT_LIST_HEAD(&mdsc->cap_dirty); 3429 INIT_LIST_HEAD(&mdsc->cap_dirty_migrating); 3430 mdsc->num_cap_flushing = 0; 3431 spin_lock_init(&mdsc->cap_dirty_lock); 3432 init_waitqueue_head(&mdsc->cap_flushing_wq); 3433 spin_lock_init(&mdsc->dentry_lru_lock); 3434 INIT_LIST_HEAD(&mdsc->dentry_lru); 3435 3436 ceph_caps_init(mdsc); 3437 ceph_adjust_min_caps(mdsc, fsc->min_caps); 3438 3439 init_rwsem(&mdsc->pool_perm_rwsem); 3440 mdsc->pool_perm_tree = RB_ROOT; 3441 3442 return 0; 3443 } 3444 3445 /* 3446 * Wait for safe replies on open mds requests. If we time out, drop 3447 * all requests from the tree to avoid dangling dentry refs. 3448 */ 3449 static void wait_requests(struct ceph_mds_client *mdsc) 3450 { 3451 struct ceph_options *opts = mdsc->fsc->client->options; 3452 struct ceph_mds_request *req; 3453 3454 mutex_lock(&mdsc->mutex); 3455 if (__get_oldest_req(mdsc)) { 3456 mutex_unlock(&mdsc->mutex); 3457 3458 dout("wait_requests waiting for requests\n"); 3459 wait_for_completion_timeout(&mdsc->safe_umount_waiters, 3460 ceph_timeout_jiffies(opts->mount_timeout)); 3461 3462 /* tear down remaining requests */ 3463 mutex_lock(&mdsc->mutex); 3464 while ((req = __get_oldest_req(mdsc))) { 3465 dout("wait_requests timed out on tid %llu\n", 3466 req->r_tid); 3467 __unregister_request(mdsc, req); 3468 } 3469 } 3470 mutex_unlock(&mdsc->mutex); 3471 dout("wait_requests done\n"); 3472 } 3473 3474 /* 3475 * called before mount is ro, and before dentries are torn down. 3476 * (hmm, does this still race with new lookups?) 3477 */ 3478 void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc) 3479 { 3480 dout("pre_umount\n"); 3481 mdsc->stopping = 1; 3482 3483 drop_leases(mdsc); 3484 ceph_flush_dirty_caps(mdsc); 3485 wait_requests(mdsc); 3486 3487 /* 3488 * wait for reply handlers to drop their request refs and 3489 * their inode/dcache refs 3490 */ 3491 ceph_msgr_flush(); 3492 } 3493 3494 /* 3495 * wait for all write mds requests to flush. 3496 */ 3497 static void wait_unsafe_requests(struct ceph_mds_client *mdsc, u64 want_tid) 3498 { 3499 struct ceph_mds_request *req = NULL, *nextreq; 3500 struct rb_node *n; 3501 3502 mutex_lock(&mdsc->mutex); 3503 dout("wait_unsafe_requests want %lld\n", want_tid); 3504 restart: 3505 req = __get_oldest_req(mdsc); 3506 while (req && req->r_tid <= want_tid) { 3507 /* find next request */ 3508 n = rb_next(&req->r_node); 3509 if (n) 3510 nextreq = rb_entry(n, struct ceph_mds_request, r_node); 3511 else 3512 nextreq = NULL; 3513 if (req->r_op != CEPH_MDS_OP_SETFILELOCK && 3514 (req->r_op & CEPH_MDS_OP_WRITE)) { 3515 /* write op */ 3516 ceph_mdsc_get_request(req); 3517 if (nextreq) 3518 ceph_mdsc_get_request(nextreq); 3519 mutex_unlock(&mdsc->mutex); 3520 dout("wait_unsafe_requests wait on %llu (want %llu)\n", 3521 req->r_tid, want_tid); 3522 wait_for_completion(&req->r_safe_completion); 3523 mutex_lock(&mdsc->mutex); 3524 ceph_mdsc_put_request(req); 3525 if (!nextreq) 3526 break; /* next dne before, so we're done! */ 3527 if (RB_EMPTY_NODE(&nextreq->r_node)) { 3528 /* next request was removed from tree */ 3529 ceph_mdsc_put_request(nextreq); 3530 goto restart; 3531 } 3532 ceph_mdsc_put_request(nextreq); /* won't go away */ 3533 } 3534 req = nextreq; 3535 } 3536 mutex_unlock(&mdsc->mutex); 3537 dout("wait_unsafe_requests done\n"); 3538 } 3539 3540 void ceph_mdsc_sync(struct ceph_mds_client *mdsc) 3541 { 3542 u64 want_tid, want_flush; 3543 3544 if (ACCESS_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) 3545 return; 3546 3547 dout("sync\n"); 3548 mutex_lock(&mdsc->mutex); 3549 want_tid = mdsc->last_tid; 3550 mutex_unlock(&mdsc->mutex); 3551 3552 ceph_flush_dirty_caps(mdsc); 3553 spin_lock(&mdsc->cap_dirty_lock); 3554 want_flush = mdsc->last_cap_flush_tid; 3555 if (!list_empty(&mdsc->cap_flush_list)) { 3556 struct ceph_cap_flush *cf = 3557 list_last_entry(&mdsc->cap_flush_list, 3558 struct ceph_cap_flush, g_list); 3559 cf->wake = true; 3560 } 3561 spin_unlock(&mdsc->cap_dirty_lock); 3562 3563 dout("sync want tid %lld flush_seq %lld\n", 3564 want_tid, want_flush); 3565 3566 wait_unsafe_requests(mdsc, want_tid); 3567 wait_caps_flush(mdsc, want_flush); 3568 } 3569 3570 /* 3571 * true if all sessions are closed, or we force unmount 3572 */ 3573 static bool done_closing_sessions(struct ceph_mds_client *mdsc, int skipped) 3574 { 3575 if (ACCESS_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) 3576 return true; 3577 return atomic_read(&mdsc->num_sessions) <= skipped; 3578 } 3579 3580 /* 3581 * called after sb is ro. 3582 */ 3583 void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc) 3584 { 3585 struct ceph_options *opts = mdsc->fsc->client->options; 3586 struct ceph_mds_session *session; 3587 int i; 3588 int skipped = 0; 3589 3590 dout("close_sessions\n"); 3591 3592 /* close sessions */ 3593 mutex_lock(&mdsc->mutex); 3594 for (i = 0; i < mdsc->max_sessions; i++) { 3595 session = __ceph_lookup_mds_session(mdsc, i); 3596 if (!session) 3597 continue; 3598 mutex_unlock(&mdsc->mutex); 3599 mutex_lock(&session->s_mutex); 3600 if (__close_session(mdsc, session) <= 0) 3601 skipped++; 3602 mutex_unlock(&session->s_mutex); 3603 ceph_put_mds_session(session); 3604 mutex_lock(&mdsc->mutex); 3605 } 3606 mutex_unlock(&mdsc->mutex); 3607 3608 dout("waiting for sessions to close\n"); 3609 wait_event_timeout(mdsc->session_close_wq, 3610 done_closing_sessions(mdsc, skipped), 3611 ceph_timeout_jiffies(opts->mount_timeout)); 3612 3613 /* tear down remaining sessions */ 3614 mutex_lock(&mdsc->mutex); 3615 for (i = 0; i < mdsc->max_sessions; i++) { 3616 if (mdsc->sessions[i]) { 3617 session = get_session(mdsc->sessions[i]); 3618 __unregister_session(mdsc, session); 3619 mutex_unlock(&mdsc->mutex); 3620 mutex_lock(&session->s_mutex); 3621 remove_session_caps(session); 3622 mutex_unlock(&session->s_mutex); 3623 ceph_put_mds_session(session); 3624 mutex_lock(&mdsc->mutex); 3625 } 3626 } 3627 WARN_ON(!list_empty(&mdsc->cap_delay_list)); 3628 mutex_unlock(&mdsc->mutex); 3629 3630 ceph_cleanup_empty_realms(mdsc); 3631 3632 cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */ 3633 3634 dout("stopped\n"); 3635 } 3636 3637 void ceph_mdsc_force_umount(struct ceph_mds_client *mdsc) 3638 { 3639 struct ceph_mds_session *session; 3640 int mds; 3641 3642 dout("force umount\n"); 3643 3644 mutex_lock(&mdsc->mutex); 3645 for (mds = 0; mds < mdsc->max_sessions; mds++) { 3646 session = __ceph_lookup_mds_session(mdsc, mds); 3647 if (!session) 3648 continue; 3649 mutex_unlock(&mdsc->mutex); 3650 mutex_lock(&session->s_mutex); 3651 __close_session(mdsc, session); 3652 if (session->s_state == CEPH_MDS_SESSION_CLOSING) { 3653 cleanup_session_requests(mdsc, session); 3654 remove_session_caps(session); 3655 } 3656 mutex_unlock(&session->s_mutex); 3657 ceph_put_mds_session(session); 3658 mutex_lock(&mdsc->mutex); 3659 kick_requests(mdsc, mds); 3660 } 3661 __wake_requests(mdsc, &mdsc->waiting_for_map); 3662 mutex_unlock(&mdsc->mutex); 3663 } 3664 3665 static void ceph_mdsc_stop(struct ceph_mds_client *mdsc) 3666 { 3667 dout("stop\n"); 3668 cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */ 3669 if (mdsc->mdsmap) 3670 ceph_mdsmap_destroy(mdsc->mdsmap); 3671 kfree(mdsc->sessions); 3672 ceph_caps_finalize(mdsc); 3673 ceph_pool_perm_destroy(mdsc); 3674 } 3675 3676 void ceph_mdsc_destroy(struct ceph_fs_client *fsc) 3677 { 3678 struct ceph_mds_client *mdsc = fsc->mdsc; 3679 3680 dout("mdsc_destroy %p\n", mdsc); 3681 ceph_mdsc_stop(mdsc); 3682 3683 /* flush out any connection work with references to us */ 3684 ceph_msgr_flush(); 3685 3686 fsc->mdsc = NULL; 3687 kfree(mdsc); 3688 dout("mdsc_destroy %p done\n", mdsc); 3689 } 3690 3691 void ceph_mdsc_handle_fsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg) 3692 { 3693 struct ceph_fs_client *fsc = mdsc->fsc; 3694 const char *mds_namespace = fsc->mount_options->mds_namespace; 3695 void *p = msg->front.iov_base; 3696 void *end = p + msg->front.iov_len; 3697 u32 epoch; 3698 u32 map_len; 3699 u32 num_fs; 3700 u32 mount_fscid = (u32)-1; 3701 u8 struct_v, struct_cv; 3702 int err = -EINVAL; 3703 3704 ceph_decode_need(&p, end, sizeof(u32), bad); 3705 epoch = ceph_decode_32(&p); 3706 3707 dout("handle_fsmap epoch %u\n", epoch); 3708 3709 ceph_decode_need(&p, end, 2 + sizeof(u32), bad); 3710 struct_v = ceph_decode_8(&p); 3711 struct_cv = ceph_decode_8(&p); 3712 map_len = ceph_decode_32(&p); 3713 3714 ceph_decode_need(&p, end, sizeof(u32) * 3, bad); 3715 p += sizeof(u32) * 2; /* skip epoch and legacy_client_fscid */ 3716 3717 num_fs = ceph_decode_32(&p); 3718 while (num_fs-- > 0) { 3719 void *info_p, *info_end; 3720 u32 info_len; 3721 u8 info_v, info_cv; 3722 u32 fscid, namelen; 3723 3724 ceph_decode_need(&p, end, 2 + sizeof(u32), bad); 3725 info_v = ceph_decode_8(&p); 3726 info_cv = ceph_decode_8(&p); 3727 info_len = ceph_decode_32(&p); 3728 ceph_decode_need(&p, end, info_len, bad); 3729 info_p = p; 3730 info_end = p + info_len; 3731 p = info_end; 3732 3733 ceph_decode_need(&info_p, info_end, sizeof(u32) * 2, bad); 3734 fscid = ceph_decode_32(&info_p); 3735 namelen = ceph_decode_32(&info_p); 3736 ceph_decode_need(&info_p, info_end, namelen, bad); 3737 3738 if (mds_namespace && 3739 strlen(mds_namespace) == namelen && 3740 !strncmp(mds_namespace, (char *)info_p, namelen)) { 3741 mount_fscid = fscid; 3742 break; 3743 } 3744 } 3745 3746 ceph_monc_got_map(&fsc->client->monc, CEPH_SUB_FSMAP, epoch); 3747 if (mount_fscid != (u32)-1) { 3748 fsc->client->monc.fs_cluster_id = mount_fscid; 3749 ceph_monc_want_map(&fsc->client->monc, CEPH_SUB_MDSMAP, 3750 0, true); 3751 ceph_monc_renew_subs(&fsc->client->monc); 3752 } else { 3753 err = -ENOENT; 3754 goto err_out; 3755 } 3756 return; 3757 bad: 3758 pr_err("error decoding fsmap\n"); 3759 err_out: 3760 mutex_lock(&mdsc->mutex); 3761 mdsc->mdsmap_err = -ENOENT; 3762 __wake_requests(mdsc, &mdsc->waiting_for_map); 3763 mutex_unlock(&mdsc->mutex); 3764 return; 3765 } 3766 3767 /* 3768 * handle mds map update. 3769 */ 3770 void ceph_mdsc_handle_mdsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg) 3771 { 3772 u32 epoch; 3773 u32 maplen; 3774 void *p = msg->front.iov_base; 3775 void *end = p + msg->front.iov_len; 3776 struct ceph_mdsmap *newmap, *oldmap; 3777 struct ceph_fsid fsid; 3778 int err = -EINVAL; 3779 3780 ceph_decode_need(&p, end, sizeof(fsid)+2*sizeof(u32), bad); 3781 ceph_decode_copy(&p, &fsid, sizeof(fsid)); 3782 if (ceph_check_fsid(mdsc->fsc->client, &fsid) < 0) 3783 return; 3784 epoch = ceph_decode_32(&p); 3785 maplen = ceph_decode_32(&p); 3786 dout("handle_map epoch %u len %d\n", epoch, (int)maplen); 3787 3788 /* do we need it? */ 3789 mutex_lock(&mdsc->mutex); 3790 if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) { 3791 dout("handle_map epoch %u <= our %u\n", 3792 epoch, mdsc->mdsmap->m_epoch); 3793 mutex_unlock(&mdsc->mutex); 3794 return; 3795 } 3796 3797 newmap = ceph_mdsmap_decode(&p, end); 3798 if (IS_ERR(newmap)) { 3799 err = PTR_ERR(newmap); 3800 goto bad_unlock; 3801 } 3802 3803 /* swap into place */ 3804 if (mdsc->mdsmap) { 3805 oldmap = mdsc->mdsmap; 3806 mdsc->mdsmap = newmap; 3807 check_new_map(mdsc, newmap, oldmap); 3808 ceph_mdsmap_destroy(oldmap); 3809 } else { 3810 mdsc->mdsmap = newmap; /* first mds map */ 3811 } 3812 mdsc->fsc->sb->s_maxbytes = mdsc->mdsmap->m_max_file_size; 3813 3814 __wake_requests(mdsc, &mdsc->waiting_for_map); 3815 ceph_monc_got_map(&mdsc->fsc->client->monc, CEPH_SUB_MDSMAP, 3816 mdsc->mdsmap->m_epoch); 3817 3818 mutex_unlock(&mdsc->mutex); 3819 schedule_delayed(mdsc); 3820 return; 3821 3822 bad_unlock: 3823 mutex_unlock(&mdsc->mutex); 3824 bad: 3825 pr_err("error decoding mdsmap %d\n", err); 3826 return; 3827 } 3828 3829 static struct ceph_connection *con_get(struct ceph_connection *con) 3830 { 3831 struct ceph_mds_session *s = con->private; 3832 3833 if (get_session(s)) { 3834 dout("mdsc con_get %p ok (%d)\n", s, atomic_read(&s->s_ref)); 3835 return con; 3836 } 3837 dout("mdsc con_get %p FAIL\n", s); 3838 return NULL; 3839 } 3840 3841 static void con_put(struct ceph_connection *con) 3842 { 3843 struct ceph_mds_session *s = con->private; 3844 3845 dout("mdsc con_put %p (%d)\n", s, atomic_read(&s->s_ref) - 1); 3846 ceph_put_mds_session(s); 3847 } 3848 3849 /* 3850 * if the client is unresponsive for long enough, the mds will kill 3851 * the session entirely. 3852 */ 3853 static void peer_reset(struct ceph_connection *con) 3854 { 3855 struct ceph_mds_session *s = con->private; 3856 struct ceph_mds_client *mdsc = s->s_mdsc; 3857 3858 pr_warn("mds%d closed our session\n", s->s_mds); 3859 send_mds_reconnect(mdsc, s); 3860 } 3861 3862 static void dispatch(struct ceph_connection *con, struct ceph_msg *msg) 3863 { 3864 struct ceph_mds_session *s = con->private; 3865 struct ceph_mds_client *mdsc = s->s_mdsc; 3866 int type = le16_to_cpu(msg->hdr.type); 3867 3868 mutex_lock(&mdsc->mutex); 3869 if (__verify_registered_session(mdsc, s) < 0) { 3870 mutex_unlock(&mdsc->mutex); 3871 goto out; 3872 } 3873 mutex_unlock(&mdsc->mutex); 3874 3875 switch (type) { 3876 case CEPH_MSG_MDS_MAP: 3877 ceph_mdsc_handle_mdsmap(mdsc, msg); 3878 break; 3879 case CEPH_MSG_FS_MAP_USER: 3880 ceph_mdsc_handle_fsmap(mdsc, msg); 3881 break; 3882 case CEPH_MSG_CLIENT_SESSION: 3883 handle_session(s, msg); 3884 break; 3885 case CEPH_MSG_CLIENT_REPLY: 3886 handle_reply(s, msg); 3887 break; 3888 case CEPH_MSG_CLIENT_REQUEST_FORWARD: 3889 handle_forward(mdsc, s, msg); 3890 break; 3891 case CEPH_MSG_CLIENT_CAPS: 3892 ceph_handle_caps(s, msg); 3893 break; 3894 case CEPH_MSG_CLIENT_SNAP: 3895 ceph_handle_snap(mdsc, s, msg); 3896 break; 3897 case CEPH_MSG_CLIENT_LEASE: 3898 handle_lease(mdsc, s, msg); 3899 break; 3900 3901 default: 3902 pr_err("received unknown message type %d %s\n", type, 3903 ceph_msg_type_name(type)); 3904 } 3905 out: 3906 ceph_msg_put(msg); 3907 } 3908 3909 /* 3910 * authentication 3911 */ 3912 3913 /* 3914 * Note: returned pointer is the address of a structure that's 3915 * managed separately. Caller must *not* attempt to free it. 3916 */ 3917 static struct ceph_auth_handshake *get_authorizer(struct ceph_connection *con, 3918 int *proto, int force_new) 3919 { 3920 struct ceph_mds_session *s = con->private; 3921 struct ceph_mds_client *mdsc = s->s_mdsc; 3922 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth; 3923 struct ceph_auth_handshake *auth = &s->s_auth; 3924 3925 if (force_new && auth->authorizer) { 3926 ceph_auth_destroy_authorizer(auth->authorizer); 3927 auth->authorizer = NULL; 3928 } 3929 if (!auth->authorizer) { 3930 int ret = ceph_auth_create_authorizer(ac, CEPH_ENTITY_TYPE_MDS, 3931 auth); 3932 if (ret) 3933 return ERR_PTR(ret); 3934 } else { 3935 int ret = ceph_auth_update_authorizer(ac, CEPH_ENTITY_TYPE_MDS, 3936 auth); 3937 if (ret) 3938 return ERR_PTR(ret); 3939 } 3940 *proto = ac->protocol; 3941 3942 return auth; 3943 } 3944 3945 3946 static int verify_authorizer_reply(struct ceph_connection *con, int len) 3947 { 3948 struct ceph_mds_session *s = con->private; 3949 struct ceph_mds_client *mdsc = s->s_mdsc; 3950 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth; 3951 3952 return ceph_auth_verify_authorizer_reply(ac, s->s_auth.authorizer, len); 3953 } 3954 3955 static int invalidate_authorizer(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 ceph_auth_invalidate_authorizer(ac, CEPH_ENTITY_TYPE_MDS); 3962 3963 return ceph_monc_validate_auth(&mdsc->fsc->client->monc); 3964 } 3965 3966 static struct ceph_msg *mds_alloc_msg(struct ceph_connection *con, 3967 struct ceph_msg_header *hdr, int *skip) 3968 { 3969 struct ceph_msg *msg; 3970 int type = (int) le16_to_cpu(hdr->type); 3971 int front_len = (int) le32_to_cpu(hdr->front_len); 3972 3973 if (con->in_msg) 3974 return con->in_msg; 3975 3976 *skip = 0; 3977 msg = ceph_msg_new(type, front_len, GFP_NOFS, false); 3978 if (!msg) { 3979 pr_err("unable to allocate msg type %d len %d\n", 3980 type, front_len); 3981 return NULL; 3982 } 3983 3984 return msg; 3985 } 3986 3987 static int mds_sign_message(struct ceph_msg *msg) 3988 { 3989 struct ceph_mds_session *s = msg->con->private; 3990 struct ceph_auth_handshake *auth = &s->s_auth; 3991 3992 return ceph_auth_sign_message(auth, msg); 3993 } 3994 3995 static int mds_check_message_signature(struct ceph_msg *msg) 3996 { 3997 struct ceph_mds_session *s = msg->con->private; 3998 struct ceph_auth_handshake *auth = &s->s_auth; 3999 4000 return ceph_auth_check_message_signature(auth, msg); 4001 } 4002 4003 static const struct ceph_connection_operations mds_con_ops = { 4004 .get = con_get, 4005 .put = con_put, 4006 .dispatch = dispatch, 4007 .get_authorizer = get_authorizer, 4008 .verify_authorizer_reply = verify_authorizer_reply, 4009 .invalidate_authorizer = invalidate_authorizer, 4010 .peer_reset = peer_reset, 4011 .alloc_msg = mds_alloc_msg, 4012 .sign_message = mds_sign_message, 4013 .check_message_signature = mds_check_message_signature, 4014 }; 4015 4016 /* eof */ 4017