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