1 // SPDX-License-Identifier: GPL-2.0 2 #include <linux/ceph/ceph_debug.h> 3 4 #include <linux/fs.h> 5 #include <linux/wait.h> 6 #include <linux/slab.h> 7 #include <linux/gfp.h> 8 #include <linux/sched.h> 9 #include <linux/sched/mm.h> 10 #include <linux/delay.h> 11 #include <linux/debugfs.h> 12 #include <linux/seq_file.h> 13 #include <linux/ratelimit.h> 14 #include <linux/bits.h> 15 #include <linux/ktime.h> 16 #include <linux/bitmap.h> 17 #include <linux/mnt_idmapping.h> 18 19 #include "super.h" 20 #include "mds_client.h" 21 #include "crypto.h" 22 23 #include <linux/ceph/ceph_features.h> 24 #include <linux/ceph/messenger.h> 25 #include <linux/ceph/decode.h> 26 #include <linux/ceph/pagelist.h> 27 #include <linux/ceph/auth.h> 28 #include <linux/ceph/debugfs.h> 29 #include <trace/events/ceph.h> 30 31 #define RECONNECT_MAX_SIZE (INT_MAX - PAGE_SIZE) 32 33 /* 34 * A cluster of MDS (metadata server) daemons is responsible for 35 * managing the file system namespace (the directory hierarchy and 36 * inodes) and for coordinating shared access to storage. Metadata is 37 * partitioning hierarchically across a number of servers, and that 38 * partition varies over time as the cluster adjusts the distribution 39 * in order to balance load. 40 * 41 * The MDS client is primarily responsible to managing synchronous 42 * metadata requests for operations like open, unlink, and so forth. 43 * If there is a MDS failure, we find out about it when we (possibly 44 * request and) receive a new MDS map, and can resubmit affected 45 * requests. 46 * 47 * For the most part, though, we take advantage of a lossless 48 * communications channel to the MDS, and do not need to worry about 49 * timing out or resubmitting requests. 50 * 51 * We maintain a stateful "session" with each MDS we interact with. 52 * Within each session, we sent periodic heartbeat messages to ensure 53 * any capabilities or leases we have been issues remain valid. If 54 * the session times out and goes stale, our leases and capabilities 55 * are no longer valid. 56 */ 57 58 struct ceph_reconnect_state { 59 struct ceph_mds_session *session; 60 int nr_caps, nr_realms; 61 struct ceph_pagelist *pagelist; 62 unsigned msg_version; 63 bool allow_multi; 64 }; 65 66 static void __wake_requests(struct ceph_mds_client *mdsc, 67 struct list_head *head); 68 static void ceph_cap_release_work(struct work_struct *work); 69 static void ceph_cap_reclaim_work(struct work_struct *work); 70 static void ceph_mdsc_reset_workfn(struct work_struct *work); 71 72 static const struct ceph_connection_operations mds_con_ops; 73 74 static void ceph_metric_bind_session(struct ceph_mds_client *mdsc, 75 struct ceph_mds_session *session) 76 { 77 struct ceph_mds_session *old; 78 79 if (!mdsc || !session || disable_send_metrics) 80 return; 81 82 old = mdsc->metric.session; 83 mdsc->metric.session = ceph_get_mds_session(session); 84 if (old) 85 ceph_put_mds_session(old); 86 87 metric_schedule_delayed(&mdsc->metric); 88 } 89 90 /* 91 * mds reply parsing 92 */ 93 94 static int parse_reply_info_quota(void **p, void *end, 95 struct ceph_mds_reply_info_in *info) 96 { 97 u8 struct_v, struct_compat; 98 u32 struct_len; 99 100 ceph_decode_8_safe(p, end, struct_v, bad); 101 ceph_decode_8_safe(p, end, struct_compat, bad); 102 /* struct_v is expected to be >= 1. we only 103 * understand encoding with struct_compat == 1. */ 104 if (!struct_v || struct_compat != 1) 105 goto bad; 106 ceph_decode_32_safe(p, end, struct_len, bad); 107 ceph_decode_need(p, end, struct_len, bad); 108 end = *p + struct_len; 109 ceph_decode_64_safe(p, end, info->max_bytes, bad); 110 ceph_decode_64_safe(p, end, info->max_files, bad); 111 *p = end; 112 return 0; 113 bad: 114 return -EIO; 115 } 116 117 static int parse_reply_info_in(void **p, void *end, 118 struct ceph_mds_reply_info_in *info, 119 u64 features, 120 struct ceph_mds_client *mdsc) 121 { 122 int err = 0; 123 u8 struct_v = 0; 124 u8 struct_compat = 0; 125 u32 struct_len = 0; 126 127 info->subvolume_id = CEPH_SUBVOLUME_ID_NONE; 128 129 if (features == (u64)-1) { 130 ceph_decode_8_safe(p, end, struct_v, bad); 131 ceph_decode_8_safe(p, end, struct_compat, bad); 132 /* struct_v is expected to be >= 1. we only understand 133 * encoding with struct_compat == 1. */ 134 if (!struct_v || struct_compat != 1) 135 goto bad; 136 ceph_decode_32_safe(p, end, struct_len, bad); 137 ceph_decode_need(p, end, struct_len, bad); 138 end = *p + struct_len; 139 } 140 141 ceph_decode_need(p, end, sizeof(struct ceph_mds_reply_inode), bad); 142 info->in = *p; 143 *p += sizeof(struct ceph_mds_reply_inode) + 144 sizeof(*info->in->fragtree.splits) * 145 le32_to_cpu(info->in->fragtree.nsplits); 146 147 ceph_decode_32_safe(p, end, info->symlink_len, bad); 148 ceph_decode_need(p, end, info->symlink_len, bad); 149 info->symlink = *p; 150 *p += info->symlink_len; 151 152 ceph_decode_copy_safe(p, end, &info->dir_layout, 153 sizeof(info->dir_layout), bad); 154 ceph_decode_32_safe(p, end, info->xattr_len, bad); 155 ceph_decode_need(p, end, info->xattr_len, bad); 156 info->xattr_data = *p; 157 *p += info->xattr_len; 158 159 if (features == (u64)-1) { 160 /* inline data */ 161 ceph_decode_64_safe(p, end, info->inline_version, bad); 162 ceph_decode_32_safe(p, end, info->inline_len, bad); 163 ceph_decode_need(p, end, info->inline_len, bad); 164 info->inline_data = *p; 165 *p += info->inline_len; 166 /* quota */ 167 err = parse_reply_info_quota(p, end, info); 168 if (err < 0) 169 goto out_bad; 170 /* pool namespace */ 171 ceph_decode_32_safe(p, end, info->pool_ns_len, bad); 172 if (info->pool_ns_len > 0) { 173 ceph_decode_need(p, end, info->pool_ns_len, bad); 174 info->pool_ns_data = *p; 175 *p += info->pool_ns_len; 176 } 177 178 /* btime */ 179 ceph_decode_need(p, end, sizeof(info->btime), bad); 180 ceph_decode_copy(p, &info->btime, sizeof(info->btime)); 181 182 /* change attribute */ 183 ceph_decode_64_safe(p, end, info->change_attr, bad); 184 185 /* dir pin */ 186 if (struct_v >= 2) { 187 ceph_decode_32_safe(p, end, info->dir_pin, bad); 188 } else { 189 info->dir_pin = -ENODATA; 190 } 191 192 /* snapshot birth time, remains zero for v<=2 */ 193 if (struct_v >= 3) { 194 ceph_decode_need(p, end, sizeof(info->snap_btime), bad); 195 ceph_decode_copy(p, &info->snap_btime, 196 sizeof(info->snap_btime)); 197 } else { 198 memset(&info->snap_btime, 0, sizeof(info->snap_btime)); 199 } 200 201 /* snapshot count, remains zero for v<=3 */ 202 if (struct_v >= 4) { 203 ceph_decode_64_safe(p, end, info->rsnaps, bad); 204 } else { 205 info->rsnaps = 0; 206 } 207 208 if (struct_v >= 5) { 209 u32 alen; 210 211 ceph_decode_32_safe(p, end, alen, bad); 212 213 while (alen--) { 214 u32 len; 215 216 /* key */ 217 ceph_decode_32_safe(p, end, len, bad); 218 ceph_decode_skip_n(p, end, len, bad); 219 /* value */ 220 ceph_decode_32_safe(p, end, len, bad); 221 ceph_decode_skip_n(p, end, len, bad); 222 } 223 } 224 225 /* fscrypt flag -- ignore */ 226 if (struct_v >= 6) 227 ceph_decode_skip_8(p, end, bad); 228 229 info->fscrypt_auth = NULL; 230 info->fscrypt_auth_len = 0; 231 info->fscrypt_file = NULL; 232 info->fscrypt_file_len = 0; 233 if (struct_v >= 7) { 234 ceph_decode_32_safe(p, end, info->fscrypt_auth_len, bad); 235 if (info->fscrypt_auth_len) { 236 info->fscrypt_auth = kmalloc(info->fscrypt_auth_len, 237 GFP_KERNEL); 238 if (!info->fscrypt_auth) 239 return -ENOMEM; 240 ceph_decode_copy_safe(p, end, info->fscrypt_auth, 241 info->fscrypt_auth_len, bad); 242 } 243 ceph_decode_32_safe(p, end, info->fscrypt_file_len, bad); 244 if (info->fscrypt_file_len) { 245 info->fscrypt_file = kmalloc(info->fscrypt_file_len, 246 GFP_KERNEL); 247 if (!info->fscrypt_file) 248 return -ENOMEM; 249 ceph_decode_copy_safe(p, end, info->fscrypt_file, 250 info->fscrypt_file_len, bad); 251 } 252 } 253 254 /* 255 * InodeStat encoding versions: 256 * v1-v7: various fields added over time 257 * v8: added optmetadata (versioned sub-structure containing 258 * optional inode metadata like charmap for case-insensitive 259 * filesystems). The kernel client doesn't support 260 * case-insensitive lookups, so we skip this field. 261 * v9: added subvolume_id (parsed below) 262 */ 263 if (struct_v >= 8) { 264 u32 v8_struct_len; 265 266 /* skip optmetadata versioned sub-structure */ 267 ceph_decode_skip_8(p, end, bad); /* struct_v */ 268 ceph_decode_skip_8(p, end, bad); /* struct_compat */ 269 ceph_decode_32_safe(p, end, v8_struct_len, bad); 270 ceph_decode_skip_n(p, end, v8_struct_len, bad); 271 } 272 273 /* struct_v 9 added subvolume_id */ 274 if (struct_v >= 9) 275 ceph_decode_64_safe(p, end, info->subvolume_id, bad); 276 277 *p = end; 278 } else { 279 /* legacy (unversioned) struct */ 280 if (features & CEPH_FEATURE_MDS_INLINE_DATA) { 281 ceph_decode_64_safe(p, end, info->inline_version, bad); 282 ceph_decode_32_safe(p, end, info->inline_len, bad); 283 ceph_decode_need(p, end, info->inline_len, bad); 284 info->inline_data = *p; 285 *p += info->inline_len; 286 } else 287 info->inline_version = CEPH_INLINE_NONE; 288 289 if (features & CEPH_FEATURE_MDS_QUOTA) { 290 err = parse_reply_info_quota(p, end, info); 291 if (err < 0) 292 goto out_bad; 293 } else { 294 info->max_bytes = 0; 295 info->max_files = 0; 296 } 297 298 info->pool_ns_len = 0; 299 info->pool_ns_data = NULL; 300 if (features & CEPH_FEATURE_FS_FILE_LAYOUT_V2) { 301 ceph_decode_32_safe(p, end, info->pool_ns_len, bad); 302 if (info->pool_ns_len > 0) { 303 ceph_decode_need(p, end, info->pool_ns_len, bad); 304 info->pool_ns_data = *p; 305 *p += info->pool_ns_len; 306 } 307 } 308 309 if (features & CEPH_FEATURE_FS_BTIME) { 310 ceph_decode_need(p, end, sizeof(info->btime), bad); 311 ceph_decode_copy(p, &info->btime, sizeof(info->btime)); 312 ceph_decode_64_safe(p, end, info->change_attr, bad); 313 } 314 315 info->dir_pin = -ENODATA; 316 /* info->snap_btime and info->rsnaps remain zero */ 317 } 318 return 0; 319 bad: 320 err = -EIO; 321 out_bad: 322 return err; 323 } 324 325 static int parse_reply_info_dir(void **p, void *end, 326 struct ceph_mds_reply_dirfrag **dirfrag, 327 u64 features) 328 { 329 if (features == (u64)-1) { 330 u8 struct_v, struct_compat; 331 u32 struct_len; 332 ceph_decode_8_safe(p, end, struct_v, bad); 333 ceph_decode_8_safe(p, end, struct_compat, bad); 334 /* struct_v is expected to be >= 1. we only understand 335 * encoding whose struct_compat == 1. */ 336 if (!struct_v || struct_compat != 1) 337 goto bad; 338 ceph_decode_32_safe(p, end, struct_len, bad); 339 ceph_decode_need(p, end, struct_len, bad); 340 end = *p + struct_len; 341 } 342 343 ceph_decode_need(p, end, sizeof(**dirfrag), bad); 344 *dirfrag = *p; 345 *p += sizeof(**dirfrag) + sizeof(u32) * le32_to_cpu((*dirfrag)->ndist); 346 if (unlikely(*p > end)) 347 goto bad; 348 if (features == (u64)-1) 349 *p = end; 350 return 0; 351 bad: 352 return -EIO; 353 } 354 355 static int parse_reply_info_lease(void **p, void *end, 356 struct ceph_mds_reply_lease **lease, 357 u64 features, u32 *altname_len, u8 **altname) 358 { 359 u8 struct_v; 360 u32 struct_len; 361 void *lend; 362 363 if (features == (u64)-1) { 364 u8 struct_compat; 365 366 ceph_decode_8_safe(p, end, struct_v, bad); 367 ceph_decode_8_safe(p, end, struct_compat, bad); 368 369 /* struct_v is expected to be >= 1. we only understand 370 * encoding whose struct_compat == 1. */ 371 if (!struct_v || struct_compat != 1) 372 goto bad; 373 374 ceph_decode_32_safe(p, end, struct_len, bad); 375 } else { 376 struct_len = sizeof(**lease); 377 *altname_len = 0; 378 *altname = NULL; 379 } 380 381 lend = *p + struct_len; 382 ceph_decode_need(p, end, struct_len, bad); 383 *lease = *p; 384 *p += sizeof(**lease); 385 386 if (features == (u64)-1) { 387 if (struct_v >= 2) { 388 ceph_decode_32_safe(p, end, *altname_len, bad); 389 ceph_decode_need(p, end, *altname_len, bad); 390 *altname = *p; 391 *p += *altname_len; 392 } else { 393 *altname = NULL; 394 *altname_len = 0; 395 } 396 } 397 *p = lend; 398 return 0; 399 bad: 400 return -EIO; 401 } 402 403 /* 404 * parse a normal reply, which may contain a (dir+)dentry and/or a 405 * target inode. 406 */ 407 static int parse_reply_info_trace(void **p, void *end, 408 struct ceph_mds_reply_info_parsed *info, 409 u64 features, 410 struct ceph_mds_client *mdsc) 411 { 412 int err; 413 414 if (info->head->is_dentry) { 415 err = parse_reply_info_in(p, end, &info->diri, features, mdsc); 416 if (err < 0) 417 goto out_bad; 418 419 err = parse_reply_info_dir(p, end, &info->dirfrag, features); 420 if (err < 0) 421 goto out_bad; 422 423 ceph_decode_32_safe(p, end, info->dname_len, bad); 424 ceph_decode_need(p, end, info->dname_len, bad); 425 info->dname = *p; 426 *p += info->dname_len; 427 428 err = parse_reply_info_lease(p, end, &info->dlease, features, 429 &info->altname_len, &info->altname); 430 if (err < 0) 431 goto out_bad; 432 } 433 434 if (info->head->is_target) { 435 err = parse_reply_info_in(p, end, &info->targeti, features, 436 mdsc); 437 if (err < 0) 438 goto out_bad; 439 } 440 441 if (unlikely(*p != end)) 442 goto bad; 443 return 0; 444 445 bad: 446 err = -EIO; 447 out_bad: 448 pr_err("problem parsing mds trace %d\n", err); 449 return err; 450 } 451 452 /* 453 * parse readdir results 454 */ 455 static int parse_reply_info_readdir(void **p, void *end, 456 struct ceph_mds_request *req, 457 u64 features, 458 struct ceph_mds_client *mdsc) 459 { 460 struct ceph_mds_reply_info_parsed *info = &req->r_reply_info; 461 struct ceph_client *cl = req->r_mdsc->fsc->client; 462 u32 num, i = 0; 463 int err; 464 465 err = parse_reply_info_dir(p, end, &info->dir_dir, features); 466 if (err < 0) 467 goto out_bad; 468 469 ceph_decode_need(p, end, sizeof(num) + 2, bad); 470 num = ceph_decode_32(p); 471 { 472 u16 flags = ceph_decode_16(p); 473 info->dir_end = !!(flags & CEPH_READDIR_FRAG_END); 474 info->dir_complete = !!(flags & CEPH_READDIR_FRAG_COMPLETE); 475 info->hash_order = !!(flags & CEPH_READDIR_HASH_ORDER); 476 info->offset_hash = !!(flags & CEPH_READDIR_OFFSET_HASH); 477 } 478 if (num == 0) 479 goto done; 480 481 BUG_ON(!info->dir_entries); 482 if ((unsigned long)(info->dir_entries + num) > 483 (unsigned long)info->dir_entries + info->dir_buf_size) { 484 pr_err_client(cl, "dir contents are larger than expected\n"); 485 WARN_ON(1); 486 goto bad; 487 } 488 489 info->dir_nr = num; 490 while (num) { 491 struct inode *inode = d_inode(req->r_dentry); 492 struct ceph_inode_info *ci = ceph_inode(inode); 493 struct ceph_mds_reply_dir_entry *rde = info->dir_entries + i; 494 struct fscrypt_str tname = FSTR_INIT(NULL, 0); 495 struct fscrypt_str oname = FSTR_INIT(NULL, 0); 496 struct ceph_fname fname; 497 u32 altname_len, _name_len; 498 u8 *altname, *_name; 499 500 /* dentry */ 501 ceph_decode_32_safe(p, end, _name_len, bad); 502 ceph_decode_need(p, end, _name_len, bad); 503 _name = *p; 504 *p += _name_len; 505 doutc(cl, "parsed dir dname '%.*s'\n", _name_len, _name); 506 507 if (info->hash_order) 508 rde->raw_hash = ceph_str_hash(ci->i_dir_layout.dl_dir_hash, 509 _name, _name_len); 510 511 /* dentry lease */ 512 err = parse_reply_info_lease(p, end, &rde->lease, features, 513 &altname_len, &altname); 514 if (err) 515 goto out_bad; 516 517 /* 518 * Try to dencrypt the dentry names and update them 519 * in the ceph_mds_reply_dir_entry struct. 520 */ 521 fname.dir = inode; 522 fname.name = _name; 523 fname.name_len = _name_len; 524 fname.ctext = altname; 525 fname.ctext_len = altname_len; 526 /* 527 * The _name_len maybe larger than altname_len, such as 528 * when the human readable name length is in range of 529 * (CEPH_NOHASH_NAME_MAX, CEPH_NOHASH_NAME_MAX + SHA256_DIGEST_SIZE), 530 * then the copy in ceph_fname_to_usr will corrupt the 531 * data if there has no encryption key. 532 * 533 * Just set the no_copy flag and then if there has no 534 * encryption key the oname.name will be assigned to 535 * _name always. 536 */ 537 fname.no_copy = true; 538 if (altname_len == 0) { 539 /* 540 * Set tname to _name, and this will be used 541 * to do the base64_decode in-place. It's 542 * safe because the decoded string should 543 * always be shorter, which is 3/4 of origin 544 * string. 545 */ 546 tname.name = _name; 547 548 /* 549 * Set oname to _name too, and this will be 550 * used to do the dencryption in-place. 551 */ 552 oname.name = _name; 553 oname.len = _name_len; 554 } else { 555 /* 556 * This will do the decryption only in-place 557 * from altname cryptext directly. 558 */ 559 oname.name = altname; 560 oname.len = altname_len; 561 } 562 rde->is_nokey = false; 563 err = ceph_fname_to_usr(&fname, &tname, &oname, &rde->is_nokey); 564 if (err) { 565 pr_err_client(cl, "unable to decode %.*s, got %d\n", 566 _name_len, _name, err); 567 goto out_bad; 568 } 569 rde->name = oname.name; 570 rde->name_len = oname.len; 571 572 /* inode */ 573 err = parse_reply_info_in(p, end, &rde->inode, features, mdsc); 574 if (err < 0) 575 goto out_bad; 576 /* ceph_readdir_prepopulate() will update it */ 577 rde->offset = 0; 578 i++; 579 num--; 580 } 581 582 done: 583 /* Skip over any unrecognized fields */ 584 *p = end; 585 return 0; 586 587 bad: 588 err = -EIO; 589 out_bad: 590 pr_err_client(cl, "problem parsing dir contents %d\n", err); 591 return err; 592 } 593 594 /* 595 * parse fcntl F_GETLK results 596 */ 597 static int parse_reply_info_filelock(void **p, void *end, 598 struct ceph_mds_reply_info_parsed *info, 599 u64 features) 600 { 601 if (*p + sizeof(*info->filelock_reply) > end) 602 goto bad; 603 604 info->filelock_reply = *p; 605 606 /* Skip over any unrecognized fields */ 607 *p = end; 608 return 0; 609 bad: 610 return -EIO; 611 } 612 613 614 #if BITS_PER_LONG == 64 615 616 #define DELEGATED_INO_AVAILABLE xa_mk_value(1) 617 618 static int ceph_parse_deleg_inos(void **p, void *end, 619 struct ceph_mds_session *s) 620 { 621 struct ceph_client *cl = s->s_mdsc->fsc->client; 622 u32 sets; 623 624 ceph_decode_32_safe(p, end, sets, bad); 625 doutc(cl, "got %u sets of delegated inodes\n", sets); 626 while (sets--) { 627 u64 start, len; 628 629 ceph_decode_64_safe(p, end, start, bad); 630 ceph_decode_64_safe(p, end, len, bad); 631 632 /* Don't accept a delegation of system inodes */ 633 if (start < CEPH_INO_SYSTEM_BASE) { 634 pr_warn_ratelimited_client(cl, 635 "ignoring reserved inode range delegation (start=0x%llx len=0x%llx)\n", 636 start, len); 637 continue; 638 } 639 while (len--) { 640 int err = xa_insert(&s->s_delegated_inos, start++, 641 DELEGATED_INO_AVAILABLE, 642 GFP_KERNEL); 643 if (!err) { 644 doutc(cl, "added delegated inode 0x%llx\n", start - 1); 645 } else if (err == -EBUSY) { 646 pr_warn_client(cl, 647 "MDS delegated inode 0x%llx more than once.\n", 648 start - 1); 649 } else { 650 return err; 651 } 652 } 653 } 654 return 0; 655 bad: 656 return -EIO; 657 } 658 659 u64 ceph_get_deleg_ino(struct ceph_mds_session *s) 660 { 661 unsigned long ino; 662 void *val; 663 664 xa_for_each(&s->s_delegated_inos, ino, val) { 665 val = xa_erase(&s->s_delegated_inos, ino); 666 if (val == DELEGATED_INO_AVAILABLE) 667 return ino; 668 } 669 return 0; 670 } 671 672 int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino) 673 { 674 return xa_insert(&s->s_delegated_inos, ino, DELEGATED_INO_AVAILABLE, 675 GFP_KERNEL); 676 } 677 #else /* BITS_PER_LONG == 64 */ 678 /* 679 * FIXME: xarrays can't handle 64-bit indexes on a 32-bit arch. For now, just 680 * ignore delegated_inos on 32 bit arch. Maybe eventually add xarrays for top 681 * and bottom words? 682 */ 683 static int ceph_parse_deleg_inos(void **p, void *end, 684 struct ceph_mds_session *s) 685 { 686 u32 sets; 687 688 ceph_decode_32_safe(p, end, sets, bad); 689 if (sets) 690 ceph_decode_skip_n(p, end, sets * 2 * sizeof(__le64), bad); 691 return 0; 692 bad: 693 return -EIO; 694 } 695 696 u64 ceph_get_deleg_ino(struct ceph_mds_session *s) 697 { 698 return 0; 699 } 700 701 int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino) 702 { 703 return 0; 704 } 705 #endif /* BITS_PER_LONG == 64 */ 706 707 /* 708 * parse create results 709 */ 710 static int parse_reply_info_create(void **p, void *end, 711 struct ceph_mds_reply_info_parsed *info, 712 u64 features, struct ceph_mds_session *s) 713 { 714 int ret; 715 716 if (features == (u64)-1 || 717 (features & CEPH_FEATURE_REPLY_CREATE_INODE)) { 718 if (*p == end) { 719 /* Malformed reply? */ 720 info->has_create_ino = false; 721 } else if (test_bit(CEPHFS_FEATURE_DELEG_INO, &s->s_features)) { 722 info->has_create_ino = true; 723 /* struct_v, struct_compat, and len */ 724 ceph_decode_skip_n(p, end, 2 + sizeof(u32), bad); 725 ceph_decode_64_safe(p, end, info->ino, bad); 726 ret = ceph_parse_deleg_inos(p, end, s); 727 if (ret) 728 return ret; 729 } else { 730 /* legacy */ 731 ceph_decode_64_safe(p, end, info->ino, bad); 732 info->has_create_ino = true; 733 } 734 } else { 735 if (*p != end) 736 goto bad; 737 } 738 739 /* Skip over any unrecognized fields */ 740 *p = end; 741 return 0; 742 bad: 743 return -EIO; 744 } 745 746 static int parse_reply_info_getvxattr(void **p, void *end, 747 struct ceph_mds_reply_info_parsed *info, 748 u64 features) 749 { 750 u32 value_len; 751 752 ceph_decode_skip_8(p, end, bad); /* skip current version: 1 */ 753 ceph_decode_skip_8(p, end, bad); /* skip first version: 1 */ 754 ceph_decode_skip_32(p, end, bad); /* skip payload length */ 755 756 ceph_decode_32_safe(p, end, value_len, bad); 757 758 if (value_len == end - *p) { 759 info->xattr_info.xattr_value = *p; 760 info->xattr_info.xattr_value_len = value_len; 761 *p = end; 762 return value_len; 763 } 764 bad: 765 return -EIO; 766 } 767 768 /* 769 * parse extra results 770 */ 771 static int parse_reply_info_extra(void **p, void *end, 772 struct ceph_mds_request *req, 773 u64 features, struct ceph_mds_session *s) 774 { 775 struct ceph_mds_reply_info_parsed *info = &req->r_reply_info; 776 u32 op = le32_to_cpu(info->head->op); 777 778 if (op == CEPH_MDS_OP_GETFILELOCK) 779 return parse_reply_info_filelock(p, end, info, features); 780 else if (op == CEPH_MDS_OP_READDIR || op == CEPH_MDS_OP_LSSNAP) 781 return parse_reply_info_readdir(p, end, req, features, 782 req->r_mdsc); 783 else if (op == CEPH_MDS_OP_CREATE) 784 return parse_reply_info_create(p, end, info, features, s); 785 else if (op == CEPH_MDS_OP_GETVXATTR) 786 return parse_reply_info_getvxattr(p, end, info, features); 787 else 788 return -EIO; 789 } 790 791 /* 792 * parse entire mds reply 793 */ 794 static int parse_reply_info(struct ceph_mds_session *s, struct ceph_msg *msg, 795 struct ceph_mds_request *req, u64 features) 796 { 797 struct ceph_mds_reply_info_parsed *info = &req->r_reply_info; 798 struct ceph_client *cl = s->s_mdsc->fsc->client; 799 void *p, *end; 800 u32 len; 801 int err; 802 803 info->head = msg->front.iov_base; 804 p = msg->front.iov_base + sizeof(struct ceph_mds_reply_head); 805 end = p + msg->front.iov_len - sizeof(struct ceph_mds_reply_head); 806 807 /* trace */ 808 ceph_decode_32_safe(&p, end, len, bad); 809 if (len > 0) { 810 ceph_decode_need(&p, end, len, bad); 811 err = parse_reply_info_trace(&p, p + len, info, features, 812 s->s_mdsc); 813 if (err < 0) 814 goto out_bad; 815 } 816 817 /* extra */ 818 ceph_decode_32_safe(&p, end, len, bad); 819 if (len > 0) { 820 ceph_decode_need(&p, end, len, bad); 821 err = parse_reply_info_extra(&p, p + len, req, features, s); 822 if (err < 0) 823 goto out_bad; 824 } 825 826 /* snap blob */ 827 ceph_decode_32_safe(&p, end, len, bad); 828 info->snapblob_len = len; 829 info->snapblob = p; 830 p += len; 831 832 if (p != end) 833 goto bad; 834 return 0; 835 836 bad: 837 err = -EIO; 838 out_bad: 839 pr_err_client(cl, "mds parse_reply err %d\n", err); 840 ceph_msg_dump(msg); 841 return err; 842 } 843 844 static void destroy_reply_info(struct ceph_mds_reply_info_parsed *info) 845 { 846 int i; 847 848 kfree(info->diri.fscrypt_auth); 849 kfree(info->diri.fscrypt_file); 850 kfree(info->targeti.fscrypt_auth); 851 kfree(info->targeti.fscrypt_file); 852 if (!info->dir_entries) 853 return; 854 855 for (i = 0; i < info->dir_nr; i++) { 856 struct ceph_mds_reply_dir_entry *rde = info->dir_entries + i; 857 858 kfree(rde->inode.fscrypt_auth); 859 kfree(rde->inode.fscrypt_file); 860 } 861 free_pages((unsigned long)info->dir_entries, get_order(info->dir_buf_size)); 862 } 863 864 /* 865 * In async unlink case the kclient won't wait for the first reply 866 * from MDS and just drop all the links and unhash the dentry and then 867 * succeeds immediately. 868 * 869 * For any new create/link/rename,etc requests followed by using the 870 * same file names we must wait for the first reply of the inflight 871 * unlink request, or the MDS possibly will fail these following 872 * requests with -EEXIST if the inflight async unlink request was 873 * delayed for some reasons. 874 * 875 * And the worst case is that for the none async openc request it will 876 * successfully open the file if the CDentry hasn't been unlinked yet, 877 * but later the previous delayed async unlink request will remove the 878 * CDentry. That means the just created file is possibly deleted later 879 * by accident. 880 * 881 * We need to wait for the inflight async unlink requests to finish 882 * when creating new files/directories by using the same file names. 883 */ 884 int ceph_wait_on_conflict_unlink(struct dentry *dentry) 885 { 886 struct ceph_fs_client *fsc = ceph_sb_to_fs_client(dentry->d_sb); 887 struct ceph_client *cl = fsc->client; 888 struct dentry *pdentry = dentry->d_parent; 889 struct dentry *udentry, *found = NULL; 890 struct ceph_dentry_info *di; 891 struct qstr dname; 892 u32 hash = dentry->d_name.hash; 893 int err; 894 895 dname.name = dentry->d_name.name; 896 dname.len = dentry->d_name.len; 897 898 rcu_read_lock(); 899 hash_for_each_possible_rcu(fsc->async_unlink_conflict, di, 900 hnode, hash) { 901 udentry = di->dentry; 902 903 spin_lock(&udentry->d_lock); 904 if (udentry->d_name.hash != hash) 905 goto next; 906 if (unlikely(udentry->d_parent != pdentry)) 907 goto next; 908 if (!hash_hashed(&di->hnode)) 909 goto next; 910 911 if (!test_bit(CEPH_DENTRY_ASYNC_UNLINK_BIT, &di->flags)) 912 pr_warn_client(cl, "dentry %p:%pd async unlink bit is not set\n", 913 dentry, dentry); 914 915 if (!d_same_name(udentry, pdentry, &dname)) 916 goto next; 917 918 found = dget_dlock(udentry); 919 spin_unlock(&udentry->d_lock); 920 break; 921 next: 922 spin_unlock(&udentry->d_lock); 923 } 924 rcu_read_unlock(); 925 926 if (likely(!found)) 927 return 0; 928 929 doutc(cl, "dentry %p:%pd conflict with old %p:%pd\n", dentry, dentry, 930 found, found); 931 932 err = wait_on_bit(&di->flags, CEPH_DENTRY_ASYNC_UNLINK_BIT, 933 TASK_KILLABLE); 934 dput(found); 935 return err; 936 } 937 938 939 /* 940 * sessions 941 */ 942 const char *ceph_session_state_name(int s) 943 { 944 switch (s) { 945 case CEPH_MDS_SESSION_NEW: return "new"; 946 case CEPH_MDS_SESSION_OPENING: return "opening"; 947 case CEPH_MDS_SESSION_OPEN: return "open"; 948 case CEPH_MDS_SESSION_HUNG: return "hung"; 949 case CEPH_MDS_SESSION_CLOSING: return "closing"; 950 case CEPH_MDS_SESSION_CLOSED: return "closed"; 951 case CEPH_MDS_SESSION_RESTARTING: return "restarting"; 952 case CEPH_MDS_SESSION_RECONNECTING: return "reconnecting"; 953 case CEPH_MDS_SESSION_REJECTED: return "rejected"; 954 default: return "???"; 955 } 956 } 957 958 struct ceph_mds_session *ceph_get_mds_session(struct ceph_mds_session *s) 959 { 960 if (refcount_inc_not_zero(&s->s_ref)) 961 return s; 962 return NULL; 963 } 964 965 void ceph_put_mds_session(struct ceph_mds_session *s) 966 { 967 if (IS_ERR_OR_NULL(s)) 968 return; 969 970 if (refcount_dec_and_test(&s->s_ref)) { 971 if (s->s_auth.authorizer) 972 ceph_auth_destroy_authorizer(s->s_auth.authorizer); 973 WARN_ON(mutex_is_locked(&s->s_mutex)); 974 xa_destroy(&s->s_delegated_inos); 975 kfree(s); 976 } 977 } 978 979 /* 980 * called under mdsc->mutex 981 */ 982 struct ceph_mds_session *__ceph_lookup_mds_session(struct ceph_mds_client *mdsc, 983 int mds) 984 { 985 if (mds >= mdsc->max_sessions || !mdsc->sessions[mds]) 986 return NULL; 987 return ceph_get_mds_session(mdsc->sessions[mds]); 988 } 989 990 static bool __have_session(struct ceph_mds_client *mdsc, int mds) 991 { 992 if (mds >= mdsc->max_sessions || !mdsc->sessions[mds]) 993 return false; 994 else 995 return true; 996 } 997 998 static int __verify_registered_session(struct ceph_mds_client *mdsc, 999 struct ceph_mds_session *s) 1000 { 1001 if (s->s_mds >= mdsc->max_sessions || 1002 mdsc->sessions[s->s_mds] != s) 1003 return -ENOENT; 1004 return 0; 1005 } 1006 1007 /* 1008 * create+register a new session for given mds. 1009 * called under mdsc->mutex. 1010 */ 1011 static struct ceph_mds_session *register_session(struct ceph_mds_client *mdsc, 1012 int mds) 1013 { 1014 struct ceph_client *cl = mdsc->fsc->client; 1015 struct ceph_mds_session *s; 1016 1017 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO) 1018 return ERR_PTR(-EIO); 1019 1020 if (mds >= mdsc->mdsmap->possible_max_rank) 1021 return ERR_PTR(-EINVAL); 1022 1023 s = kzalloc_obj(*s, GFP_NOFS); 1024 if (!s) 1025 return ERR_PTR(-ENOMEM); 1026 1027 if (mds >= mdsc->max_sessions) { 1028 int newmax = 1 << get_count_order(mds + 1); 1029 struct ceph_mds_session **sa; 1030 size_t ptr_size = sizeof(struct ceph_mds_session *); 1031 1032 doutc(cl, "realloc to %d\n", newmax); 1033 sa = kcalloc(newmax, ptr_size, GFP_NOFS); 1034 if (!sa) 1035 goto fail_realloc; 1036 if (mdsc->sessions) { 1037 memcpy(sa, mdsc->sessions, 1038 mdsc->max_sessions * ptr_size); 1039 kfree(mdsc->sessions); 1040 } 1041 mdsc->sessions = sa; 1042 mdsc->max_sessions = newmax; 1043 } 1044 1045 doutc(cl, "mds%d\n", mds); 1046 s->s_mdsc = mdsc; 1047 s->s_mds = mds; 1048 s->s_state = CEPH_MDS_SESSION_NEW; 1049 mutex_init(&s->s_mutex); 1050 1051 ceph_con_init(&s->s_con, s, &mds_con_ops, &mdsc->fsc->client->msgr); 1052 1053 atomic_set(&s->s_cap_gen, 1); 1054 s->s_cap_ttl = jiffies - 1; 1055 1056 spin_lock_init(&s->s_cap_lock); 1057 INIT_LIST_HEAD(&s->s_caps); 1058 refcount_set(&s->s_ref, 1); 1059 INIT_LIST_HEAD(&s->s_waiting); 1060 INIT_LIST_HEAD(&s->s_unsafe); 1061 xa_init(&s->s_delegated_inos); 1062 INIT_LIST_HEAD(&s->s_cap_releases); 1063 INIT_WORK(&s->s_cap_release_work, ceph_cap_release_work); 1064 1065 INIT_LIST_HEAD(&s->s_cap_dirty); 1066 INIT_LIST_HEAD(&s->s_cap_flushing); 1067 1068 mdsc->sessions[mds] = s; 1069 atomic_inc(&mdsc->num_sessions); 1070 refcount_inc(&s->s_ref); /* one ref to sessions[], one to caller */ 1071 1072 ceph_con_open(&s->s_con, CEPH_ENTITY_TYPE_MDS, mds, 1073 ceph_mdsmap_get_addr(mdsc->mdsmap, mds)); 1074 1075 return s; 1076 1077 fail_realloc: 1078 kfree(s); 1079 return ERR_PTR(-ENOMEM); 1080 } 1081 1082 /* 1083 * called under mdsc->mutex 1084 */ 1085 static void __unregister_session(struct ceph_mds_client *mdsc, 1086 struct ceph_mds_session *s) 1087 { 1088 doutc(mdsc->fsc->client, "mds%d %p\n", s->s_mds, s); 1089 BUG_ON(mdsc->sessions[s->s_mds] != s); 1090 mdsc->sessions[s->s_mds] = NULL; 1091 ceph_con_close(&s->s_con); 1092 ceph_put_mds_session(s); 1093 atomic_dec(&mdsc->num_sessions); 1094 } 1095 1096 /* 1097 * drop session refs in request. 1098 * 1099 * should be last request ref, or hold mdsc->mutex 1100 */ 1101 static void put_request_session(struct ceph_mds_request *req) 1102 { 1103 if (req->r_session) { 1104 ceph_put_mds_session(req->r_session); 1105 req->r_session = NULL; 1106 } 1107 } 1108 1109 void ceph_mdsc_iterate_sessions(struct ceph_mds_client *mdsc, 1110 void (*cb)(struct ceph_mds_session *), 1111 bool check_state) 1112 { 1113 int mds; 1114 1115 mutex_lock(&mdsc->mutex); 1116 for (mds = 0; mds < mdsc->max_sessions; ++mds) { 1117 struct ceph_mds_session *s; 1118 1119 s = __ceph_lookup_mds_session(mdsc, mds); 1120 if (!s) 1121 continue; 1122 1123 if (check_state && !check_session_state(s)) { 1124 ceph_put_mds_session(s); 1125 continue; 1126 } 1127 1128 mutex_unlock(&mdsc->mutex); 1129 cb(s); 1130 ceph_put_mds_session(s); 1131 mutex_lock(&mdsc->mutex); 1132 } 1133 mutex_unlock(&mdsc->mutex); 1134 } 1135 1136 void ceph_mdsc_release_request(struct kref *kref) 1137 { 1138 struct ceph_mds_request *req = container_of(kref, 1139 struct ceph_mds_request, 1140 r_kref); 1141 ceph_mdsc_release_dir_caps_async(req); 1142 destroy_reply_info(&req->r_reply_info); 1143 if (req->r_request) 1144 ceph_msg_put(req->r_request); 1145 if (req->r_reply) 1146 ceph_msg_put(req->r_reply); 1147 if (req->r_inode) { 1148 ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN); 1149 iput(req->r_inode); 1150 } 1151 if (req->r_parent) { 1152 ceph_put_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN); 1153 iput(req->r_parent); 1154 } 1155 iput(req->r_target_inode); 1156 iput(req->r_new_inode); 1157 if (req->r_dentry) 1158 dput(req->r_dentry); 1159 if (req->r_old_dentry) 1160 dput(req->r_old_dentry); 1161 if (req->r_old_dentry_dir) { 1162 /* 1163 * track (and drop pins for) r_old_dentry_dir 1164 * separately, since r_old_dentry's d_parent may have 1165 * changed between the dir mutex being dropped and 1166 * this request being freed. 1167 */ 1168 ceph_put_cap_refs(ceph_inode(req->r_old_dentry_dir), 1169 CEPH_CAP_PIN); 1170 iput(req->r_old_dentry_dir); 1171 } 1172 kfree(req->r_path1); 1173 kfree(req->r_path2); 1174 put_cred(req->r_cred); 1175 if (req->r_mnt_idmap) 1176 mnt_idmap_put(req->r_mnt_idmap); 1177 if (req->r_pagelist) 1178 ceph_pagelist_release(req->r_pagelist); 1179 kfree(req->r_fscrypt_auth); 1180 kfree(req->r_altname); 1181 put_request_session(req); 1182 ceph_unreserve_caps(req->r_mdsc, &req->r_caps_reservation); 1183 WARN_ON_ONCE(!list_empty(&req->r_wait)); 1184 kmem_cache_free(ceph_mds_request_cachep, req); 1185 } 1186 1187 DEFINE_RB_FUNCS(request, struct ceph_mds_request, r_tid, r_node) 1188 1189 /* 1190 * lookup session, bump ref if found. 1191 * 1192 * called under mdsc->mutex. 1193 */ 1194 static struct ceph_mds_request * 1195 lookup_get_request(struct ceph_mds_client *mdsc, u64 tid) 1196 { 1197 struct ceph_mds_request *req; 1198 1199 req = lookup_request(&mdsc->request_tree, tid); 1200 if (req) 1201 ceph_mdsc_get_request(req); 1202 1203 return req; 1204 } 1205 1206 /* 1207 * Register an in-flight request, and assign a tid. Link to directory 1208 * are modifying (if any). 1209 * 1210 * Called under mdsc->mutex. 1211 */ 1212 static void __register_request(struct ceph_mds_client *mdsc, 1213 struct ceph_mds_request *req, 1214 struct inode *dir) 1215 { 1216 struct ceph_client *cl = mdsc->fsc->client; 1217 int ret = 0; 1218 1219 req->r_tid = ++mdsc->last_tid; 1220 if (req->r_num_caps) { 1221 ret = ceph_reserve_caps(mdsc, &req->r_caps_reservation, 1222 req->r_num_caps); 1223 if (ret < 0) { 1224 pr_err_client(cl, "%p failed to reserve caps: %d\n", 1225 req, ret); 1226 /* set req->r_err to fail early from __do_request */ 1227 req->r_err = ret; 1228 return; 1229 } 1230 } 1231 doutc(cl, "%p tid %lld\n", req, req->r_tid); 1232 ceph_mdsc_get_request(req); 1233 insert_request(&mdsc->request_tree, req); 1234 1235 req->r_cred = get_current_cred(); 1236 if (!req->r_mnt_idmap) 1237 req->r_mnt_idmap = &nop_mnt_idmap; 1238 1239 if (mdsc->oldest_tid == 0 && req->r_op != CEPH_MDS_OP_SETFILELOCK) 1240 mdsc->oldest_tid = req->r_tid; 1241 1242 if (dir) { 1243 struct ceph_inode_info *ci = ceph_inode(dir); 1244 1245 ihold(dir); 1246 req->r_unsafe_dir = dir; 1247 spin_lock(&ci->i_unsafe_lock); 1248 list_add_tail(&req->r_unsafe_dir_item, &ci->i_unsafe_dirops); 1249 spin_unlock(&ci->i_unsafe_lock); 1250 } 1251 } 1252 1253 static void __unregister_request(struct ceph_mds_client *mdsc, 1254 struct ceph_mds_request *req) 1255 { 1256 doutc(mdsc->fsc->client, "%p tid %lld\n", req, req->r_tid); 1257 1258 /* Never leave an unregistered request on an unsafe list! */ 1259 list_del_init(&req->r_unsafe_item); 1260 1261 if (req->r_tid == mdsc->oldest_tid) { 1262 struct rb_node *p = rb_next(&req->r_node); 1263 mdsc->oldest_tid = 0; 1264 while (p) { 1265 struct ceph_mds_request *next_req = 1266 rb_entry(p, struct ceph_mds_request, r_node); 1267 if (next_req->r_op != CEPH_MDS_OP_SETFILELOCK) { 1268 mdsc->oldest_tid = next_req->r_tid; 1269 break; 1270 } 1271 p = rb_next(p); 1272 } 1273 } 1274 1275 erase_request(&mdsc->request_tree, req); 1276 1277 if (req->r_unsafe_dir) { 1278 struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir); 1279 spin_lock(&ci->i_unsafe_lock); 1280 list_del_init(&req->r_unsafe_dir_item); 1281 spin_unlock(&ci->i_unsafe_lock); 1282 } 1283 if (req->r_target_inode && 1284 test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) { 1285 struct ceph_inode_info *ci = ceph_inode(req->r_target_inode); 1286 spin_lock(&ci->i_unsafe_lock); 1287 list_del_init(&req->r_unsafe_target_item); 1288 spin_unlock(&ci->i_unsafe_lock); 1289 } 1290 1291 if (req->r_unsafe_dir) { 1292 iput(req->r_unsafe_dir); 1293 req->r_unsafe_dir = NULL; 1294 } 1295 1296 complete_all(&req->r_safe_completion); 1297 1298 ceph_mdsc_put_request(req); 1299 } 1300 1301 /* 1302 * Walk back up the dentry tree until we hit a dentry representing a 1303 * non-snapshot inode. We do this using the rcu_read_lock (which must be held 1304 * when calling this) to ensure that the objects won't disappear while we're 1305 * working with them. Once we hit a candidate dentry, we attempt to take a 1306 * reference to it, and return that as the result. 1307 */ 1308 static struct inode *get_nonsnap_parent(struct dentry *dentry) 1309 { 1310 struct inode *inode = NULL; 1311 1312 while (dentry && !IS_ROOT(dentry)) { 1313 inode = d_inode_rcu(dentry); 1314 if (!inode || ceph_snap(inode) == CEPH_NOSNAP) 1315 break; 1316 dentry = dentry->d_parent; 1317 } 1318 if (inode) 1319 inode = igrab(inode); 1320 return inode; 1321 } 1322 1323 /* 1324 * Choose mds to send request to next. If there is a hint set in the 1325 * request (e.g., due to a prior forward hint from the mds), use that. 1326 * Otherwise, consult frag tree and/or caps to identify the 1327 * appropriate mds. If all else fails, choose randomly. 1328 * 1329 * Called under mdsc->mutex. 1330 */ 1331 static int __choose_mds(struct ceph_mds_client *mdsc, 1332 struct ceph_mds_request *req, 1333 bool *random) 1334 { 1335 struct inode *inode; 1336 struct ceph_inode_info *ci; 1337 struct ceph_cap *cap; 1338 int mode = req->r_direct_mode; 1339 int mds = -1; 1340 u32 hash = req->r_direct_hash; 1341 bool is_hash = test_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags); 1342 struct ceph_client *cl = mdsc->fsc->client; 1343 1344 if (random) 1345 *random = false; 1346 1347 /* 1348 * is there a specific mds we should try? ignore hint if we have 1349 * no session and the mds is not up (active or recovering). 1350 */ 1351 if (req->r_resend_mds >= 0 && 1352 (__have_session(mdsc, req->r_resend_mds) || 1353 ceph_mdsmap_get_state(mdsc->mdsmap, req->r_resend_mds) > 0)) { 1354 doutc(cl, "using resend_mds mds%d\n", req->r_resend_mds); 1355 return req->r_resend_mds; 1356 } 1357 1358 if (mode == USE_RANDOM_MDS) 1359 goto random; 1360 1361 inode = NULL; 1362 if (req->r_inode) { 1363 if (ceph_snap(req->r_inode) != CEPH_SNAPDIR) { 1364 inode = req->r_inode; 1365 ihold(inode); 1366 } else { 1367 /* req->r_dentry is non-null for LSSNAP request */ 1368 rcu_read_lock(); 1369 inode = get_nonsnap_parent(req->r_dentry); 1370 rcu_read_unlock(); 1371 doutc(cl, "using snapdir's parent %p %llx.%llx\n", 1372 inode, ceph_vinop(inode)); 1373 } 1374 } else if (req->r_dentry) { 1375 /* ignore race with rename; old or new d_parent is okay */ 1376 struct dentry *parent; 1377 struct inode *dir; 1378 1379 rcu_read_lock(); 1380 parent = READ_ONCE(req->r_dentry->d_parent); 1381 dir = req->r_parent ? : d_inode_rcu(parent); 1382 1383 if (!dir || dir->i_sb != mdsc->fsc->sb) { 1384 /* not this fs or parent went negative */ 1385 inode = d_inode(req->r_dentry); 1386 if (inode) 1387 ihold(inode); 1388 } else if (ceph_snap(dir) != CEPH_NOSNAP) { 1389 /* direct snapped/virtual snapdir requests 1390 * based on parent dir inode */ 1391 inode = get_nonsnap_parent(parent); 1392 doutc(cl, "using nonsnap parent %p %llx.%llx\n", 1393 inode, ceph_vinop(inode)); 1394 } else { 1395 /* dentry target */ 1396 inode = d_inode(req->r_dentry); 1397 if (!inode || mode == USE_AUTH_MDS) { 1398 /* dir + name */ 1399 inode = igrab(dir); 1400 hash = ceph_dentry_hash(dir, req->r_dentry); 1401 is_hash = true; 1402 } else { 1403 ihold(inode); 1404 } 1405 } 1406 rcu_read_unlock(); 1407 } 1408 1409 if (!inode) 1410 goto random; 1411 1412 doutc(cl, "%p %llx.%llx is_hash=%d (0x%x) mode %d\n", inode, 1413 ceph_vinop(inode), (int)is_hash, hash, mode); 1414 ci = ceph_inode(inode); 1415 1416 if (is_hash && S_ISDIR(inode->i_mode)) { 1417 struct ceph_inode_frag frag; 1418 int found; 1419 1420 ceph_choose_frag(ci, hash, &frag, &found); 1421 if (found) { 1422 if (mode == USE_ANY_MDS && frag.ndist > 0) { 1423 u8 r; 1424 1425 /* choose a random replica */ 1426 get_random_bytes(&r, 1); 1427 r %= frag.ndist; 1428 mds = frag.dist[r]; 1429 doutc(cl, "%p %llx.%llx frag %u mds%d (%d/%d)\n", 1430 inode, ceph_vinop(inode), frag.frag, 1431 mds, (int)r, frag.ndist); 1432 if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >= 1433 CEPH_MDS_STATE_ACTIVE && 1434 !ceph_mdsmap_is_laggy(mdsc->mdsmap, mds)) 1435 goto out; 1436 } 1437 1438 /* since this file/dir wasn't known to be 1439 * replicated, then we want to look for the 1440 * authoritative mds. */ 1441 if (frag.mds >= 0) { 1442 /* choose auth mds */ 1443 mds = frag.mds; 1444 doutc(cl, "%p %llx.%llx frag %u mds%d (auth)\n", 1445 inode, ceph_vinop(inode), frag.frag, mds); 1446 if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >= 1447 CEPH_MDS_STATE_ACTIVE) { 1448 if (!ceph_mdsmap_is_laggy(mdsc->mdsmap, 1449 mds)) 1450 goto out; 1451 } 1452 } 1453 mode = USE_AUTH_MDS; 1454 } 1455 } 1456 1457 spin_lock(&ci->i_ceph_lock); 1458 cap = NULL; 1459 if (mode == USE_AUTH_MDS) 1460 cap = ci->i_auth_cap; 1461 if (!cap && !RB_EMPTY_ROOT(&ci->i_caps)) 1462 cap = rb_entry(rb_first(&ci->i_caps), struct ceph_cap, ci_node); 1463 if (!cap) { 1464 spin_unlock(&ci->i_ceph_lock); 1465 iput(inode); 1466 goto random; 1467 } 1468 mds = cap->session->s_mds; 1469 doutc(cl, "%p %llx.%llx mds%d (%scap %p)\n", inode, 1470 ceph_vinop(inode), mds, 1471 cap == ci->i_auth_cap ? "auth " : "", cap); 1472 spin_unlock(&ci->i_ceph_lock); 1473 out: 1474 iput(inode); 1475 return mds; 1476 1477 random: 1478 if (random) 1479 *random = true; 1480 1481 mds = ceph_mdsmap_get_random_mds(mdsc->mdsmap); 1482 doutc(cl, "chose random mds%d\n", mds); 1483 return mds; 1484 } 1485 1486 1487 /* 1488 * session messages 1489 */ 1490 struct ceph_msg *ceph_create_session_msg(u32 op, u64 seq) 1491 { 1492 struct ceph_msg *msg; 1493 struct ceph_mds_session_head *h; 1494 1495 msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h), GFP_NOFS, 1496 false); 1497 if (!msg) { 1498 pr_err("ENOMEM creating session %s msg\n", 1499 ceph_session_op_name(op)); 1500 return NULL; 1501 } 1502 h = msg->front.iov_base; 1503 h->op = cpu_to_le32(op); 1504 h->seq = cpu_to_le64(seq); 1505 1506 return msg; 1507 } 1508 1509 static const unsigned char feature_bits[] = CEPHFS_FEATURES_CLIENT_SUPPORTED; 1510 #define FEATURE_BYTES(c) (DIV_ROUND_UP((size_t)feature_bits[c - 1] + 1, 64) * 8) 1511 static int encode_supported_features(void **p, void *end) 1512 { 1513 static const size_t count = ARRAY_SIZE(feature_bits); 1514 1515 if (count > 0) { 1516 size_t i; 1517 size_t size = FEATURE_BYTES(count); 1518 unsigned long bit; 1519 1520 if (WARN_ON_ONCE(*p + 4 + size > end)) 1521 return -ERANGE; 1522 1523 ceph_encode_32(p, size); 1524 memset(*p, 0, size); 1525 for (i = 0; i < count; i++) { 1526 bit = feature_bits[i]; 1527 ((unsigned char *)(*p))[bit / 8] |= BIT(bit % 8); 1528 } 1529 *p += size; 1530 } else { 1531 if (WARN_ON_ONCE(*p + 4 > end)) 1532 return -ERANGE; 1533 1534 ceph_encode_32(p, 0); 1535 } 1536 1537 return 0; 1538 } 1539 1540 static const unsigned char metric_bits[] = CEPHFS_METRIC_SPEC_CLIENT_SUPPORTED; 1541 #define METRIC_BYTES(cnt) (DIV_ROUND_UP((size_t)metric_bits[cnt - 1] + 1, 64) * 8) 1542 static int encode_metric_spec(void **p, void *end) 1543 { 1544 static const size_t count = ARRAY_SIZE(metric_bits); 1545 1546 /* header */ 1547 if (WARN_ON_ONCE(*p + 2 > end)) 1548 return -ERANGE; 1549 1550 ceph_encode_8(p, 1); /* version */ 1551 ceph_encode_8(p, 1); /* compat */ 1552 1553 if (count > 0) { 1554 size_t i; 1555 size_t size = METRIC_BYTES(count); 1556 1557 if (WARN_ON_ONCE(*p + 4 + 4 + size > end)) 1558 return -ERANGE; 1559 1560 /* metric spec info length */ 1561 ceph_encode_32(p, 4 + size); 1562 1563 /* metric spec */ 1564 ceph_encode_32(p, size); 1565 memset(*p, 0, size); 1566 for (i = 0; i < count; i++) 1567 ((unsigned char *)(*p))[i / 8] |= BIT(metric_bits[i] % 8); 1568 *p += size; 1569 } else { 1570 if (WARN_ON_ONCE(*p + 4 + 4 > end)) 1571 return -ERANGE; 1572 1573 /* metric spec info length */ 1574 ceph_encode_32(p, 4); 1575 /* metric spec */ 1576 ceph_encode_32(p, 0); 1577 } 1578 1579 return 0; 1580 } 1581 1582 /* 1583 * session message, specialization for CEPH_SESSION_REQUEST_OPEN 1584 * to include additional client metadata fields. 1585 */ 1586 static struct ceph_msg * 1587 create_session_full_msg(struct ceph_mds_client *mdsc, int op, u64 seq) 1588 { 1589 struct ceph_msg *msg; 1590 struct ceph_mds_session_head *h; 1591 int i; 1592 int extra_bytes = 0; 1593 int metadata_key_count = 0; 1594 struct ceph_options *opt = mdsc->fsc->client->options; 1595 struct ceph_mount_options *fsopt = mdsc->fsc->mount_options; 1596 struct ceph_client *cl = mdsc->fsc->client; 1597 size_t size, count; 1598 void *p, *end; 1599 int ret; 1600 1601 const char* metadata[][2] = { 1602 {"hostname", mdsc->nodename}, 1603 {"kernel_version", init_utsname()->release}, 1604 {"entity_id", opt->name ? : ""}, 1605 {"root", fsopt->server_path ? : "/"}, 1606 {NULL, NULL} 1607 }; 1608 1609 /* Calculate serialized length of metadata */ 1610 extra_bytes = 4; /* map length */ 1611 for (i = 0; metadata[i][0]; ++i) { 1612 extra_bytes += 8 + strlen(metadata[i][0]) + 1613 strlen(metadata[i][1]); 1614 metadata_key_count++; 1615 } 1616 1617 /* supported feature */ 1618 size = 0; 1619 count = ARRAY_SIZE(feature_bits); 1620 if (count > 0) 1621 size = FEATURE_BYTES(count); 1622 extra_bytes += 4 + size; 1623 1624 /* metric spec */ 1625 size = 0; 1626 count = ARRAY_SIZE(metric_bits); 1627 if (count > 0) 1628 size = METRIC_BYTES(count); 1629 extra_bytes += 2 + 4 + 4 + size; 1630 1631 /* flags, mds auth caps and oldest_client_tid */ 1632 extra_bytes += 4 + 4 + 8; 1633 1634 /* Allocate the message */ 1635 msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h) + extra_bytes, 1636 GFP_NOFS, false); 1637 if (!msg) { 1638 pr_err_client(cl, "ENOMEM creating session open msg\n"); 1639 return ERR_PTR(-ENOMEM); 1640 } 1641 p = msg->front.iov_base; 1642 end = p + msg->front.iov_len; 1643 1644 h = p; 1645 h->op = cpu_to_le32(op); 1646 h->seq = cpu_to_le64(seq); 1647 1648 /* 1649 * Serialize client metadata into waiting buffer space, using 1650 * the format that userspace expects for map<string, string> 1651 * 1652 * ClientSession messages with metadata are v7 1653 */ 1654 msg->hdr.version = cpu_to_le16(7); 1655 msg->hdr.compat_version = cpu_to_le16(1); 1656 1657 /* The write pointer, following the session_head structure */ 1658 p += sizeof(*h); 1659 1660 /* Number of entries in the map */ 1661 ceph_encode_32(&p, metadata_key_count); 1662 1663 /* Two length-prefixed strings for each entry in the map */ 1664 for (i = 0; metadata[i][0]; ++i) { 1665 size_t const key_len = strlen(metadata[i][0]); 1666 size_t const val_len = strlen(metadata[i][1]); 1667 1668 ceph_encode_32(&p, key_len); 1669 memcpy(p, metadata[i][0], key_len); 1670 p += key_len; 1671 ceph_encode_32(&p, val_len); 1672 memcpy(p, metadata[i][1], val_len); 1673 p += val_len; 1674 } 1675 1676 ret = encode_supported_features(&p, end); 1677 if (ret) { 1678 pr_err_client(cl, "encode_supported_features failed!\n"); 1679 ceph_msg_put(msg); 1680 return ERR_PTR(ret); 1681 } 1682 1683 ret = encode_metric_spec(&p, end); 1684 if (ret) { 1685 pr_err_client(cl, "encode_metric_spec failed!\n"); 1686 ceph_msg_put(msg); 1687 return ERR_PTR(ret); 1688 } 1689 1690 /* version == 5, flags */ 1691 ceph_encode_32(&p, 0); 1692 1693 /* version == 6, mds auth caps */ 1694 ceph_encode_32(&p, 0); 1695 1696 /* version == 7, oldest_client_tid */ 1697 ceph_encode_64(&p, mdsc->oldest_tid); 1698 1699 msg->front.iov_len = p - msg->front.iov_base; 1700 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len); 1701 1702 return msg; 1703 } 1704 1705 /* 1706 * send session open request. 1707 * 1708 * called under mdsc->mutex 1709 */ 1710 static int __open_session(struct ceph_mds_client *mdsc, 1711 struct ceph_mds_session *session) 1712 { 1713 struct ceph_msg *msg; 1714 int mstate; 1715 int mds = session->s_mds; 1716 1717 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO) 1718 return -EIO; 1719 1720 /* wait for mds to go active? */ 1721 mstate = ceph_mdsmap_get_state(mdsc->mdsmap, mds); 1722 doutc(mdsc->fsc->client, "open_session to mds%d (%s)\n", mds, 1723 ceph_mds_state_name(mstate)); 1724 session->s_state = CEPH_MDS_SESSION_OPENING; 1725 session->s_renew_requested = jiffies; 1726 1727 /* send connect message */ 1728 msg = create_session_full_msg(mdsc, CEPH_SESSION_REQUEST_OPEN, 1729 session->s_seq); 1730 if (IS_ERR(msg)) 1731 return PTR_ERR(msg); 1732 ceph_con_send(&session->s_con, msg); 1733 return 0; 1734 } 1735 1736 /* 1737 * open sessions for any export targets for the given mds 1738 * 1739 * called under mdsc->mutex 1740 */ 1741 static struct ceph_mds_session * 1742 __open_export_target_session(struct ceph_mds_client *mdsc, int target) 1743 { 1744 struct ceph_mds_session *session; 1745 int ret; 1746 1747 session = __ceph_lookup_mds_session(mdsc, target); 1748 if (!session) { 1749 session = register_session(mdsc, target); 1750 if (IS_ERR(session)) 1751 return session; 1752 } 1753 if (session->s_state == CEPH_MDS_SESSION_NEW || 1754 session->s_state == CEPH_MDS_SESSION_CLOSING) { 1755 ret = __open_session(mdsc, session); 1756 if (ret) 1757 return ERR_PTR(ret); 1758 } 1759 1760 return session; 1761 } 1762 1763 struct ceph_mds_session * 1764 ceph_mdsc_open_export_target_session(struct ceph_mds_client *mdsc, int target) 1765 { 1766 struct ceph_mds_session *session; 1767 struct ceph_client *cl = mdsc->fsc->client; 1768 1769 doutc(cl, "to mds%d\n", target); 1770 1771 mutex_lock(&mdsc->mutex); 1772 session = __open_export_target_session(mdsc, target); 1773 mutex_unlock(&mdsc->mutex); 1774 1775 return session; 1776 } 1777 1778 static void __open_export_target_sessions(struct ceph_mds_client *mdsc, 1779 struct ceph_mds_session *session) 1780 { 1781 struct ceph_mds_info *mi; 1782 struct ceph_mds_session *ts; 1783 int i, mds = session->s_mds; 1784 struct ceph_client *cl = mdsc->fsc->client; 1785 1786 if (mds >= mdsc->mdsmap->possible_max_rank) 1787 return; 1788 1789 mi = &mdsc->mdsmap->m_info[mds]; 1790 doutc(cl, "for mds%d (%d targets)\n", session->s_mds, 1791 mi->num_export_targets); 1792 1793 for (i = 0; i < mi->num_export_targets; i++) { 1794 ts = __open_export_target_session(mdsc, mi->export_targets[i]); 1795 ceph_put_mds_session(ts); 1796 } 1797 } 1798 1799 /* 1800 * session caps 1801 */ 1802 1803 static void detach_cap_releases(struct ceph_mds_session *session, 1804 struct list_head *target) 1805 { 1806 struct ceph_client *cl = session->s_mdsc->fsc->client; 1807 1808 lockdep_assert_held(&session->s_cap_lock); 1809 1810 list_splice_init(&session->s_cap_releases, target); 1811 session->s_num_cap_releases = 0; 1812 doutc(cl, "mds%d\n", session->s_mds); 1813 } 1814 1815 static void dispose_cap_releases(struct ceph_mds_client *mdsc, 1816 struct list_head *dispose) 1817 { 1818 while (!list_empty(dispose)) { 1819 struct ceph_cap *cap; 1820 /* zero out the in-progress message */ 1821 cap = list_first_entry(dispose, struct ceph_cap, session_caps); 1822 list_del(&cap->session_caps); 1823 ceph_put_cap(mdsc, cap); 1824 } 1825 } 1826 1827 static void cleanup_session_requests(struct ceph_mds_client *mdsc, 1828 struct ceph_mds_session *session) 1829 { 1830 struct ceph_client *cl = mdsc->fsc->client; 1831 struct ceph_mds_request *req; 1832 struct rb_node *p; 1833 1834 doutc(cl, "mds%d\n", session->s_mds); 1835 mutex_lock(&mdsc->mutex); 1836 while (!list_empty(&session->s_unsafe)) { 1837 req = list_first_entry(&session->s_unsafe, 1838 struct ceph_mds_request, r_unsafe_item); 1839 pr_warn_ratelimited_client(cl, " dropping unsafe request %llu\n", 1840 req->r_tid); 1841 if (req->r_target_inode) 1842 mapping_set_error(req->r_target_inode->i_mapping, -EIO); 1843 if (req->r_unsafe_dir) 1844 mapping_set_error(req->r_unsafe_dir->i_mapping, -EIO); 1845 __unregister_request(mdsc, req); 1846 } 1847 /* zero r_attempts, so kick_requests() will re-send requests */ 1848 p = rb_first(&mdsc->request_tree); 1849 while (p) { 1850 req = rb_entry(p, struct ceph_mds_request, r_node); 1851 p = rb_next(p); 1852 if (req->r_session && 1853 req->r_session->s_mds == session->s_mds) 1854 req->r_attempts = 0; 1855 } 1856 mutex_unlock(&mdsc->mutex); 1857 } 1858 1859 /* 1860 * Helper to safely iterate over all caps associated with a session, with 1861 * special care taken to handle a racing __ceph_remove_cap(). 1862 * 1863 * Caller must hold session s_mutex. 1864 */ 1865 int ceph_iterate_session_caps(struct ceph_mds_session *session, 1866 int (*cb)(struct inode *, int mds, void *), 1867 void *arg) 1868 { 1869 struct ceph_client *cl = session->s_mdsc->fsc->client; 1870 struct list_head *p; 1871 struct ceph_cap *cap; 1872 struct inode *inode, *last_inode = NULL; 1873 struct ceph_cap *old_cap = NULL; 1874 int ret; 1875 1876 doutc(cl, "%p mds%d\n", session, session->s_mds); 1877 spin_lock(&session->s_cap_lock); 1878 p = session->s_caps.next; 1879 while (p != &session->s_caps) { 1880 int mds; 1881 1882 cap = list_entry(p, struct ceph_cap, session_caps); 1883 inode = igrab(&cap->ci->netfs.inode); 1884 if (!inode) { 1885 p = p->next; 1886 continue; 1887 } 1888 session->s_cap_iterator = cap; 1889 mds = cap->mds; 1890 spin_unlock(&session->s_cap_lock); 1891 1892 if (last_inode) { 1893 iput(last_inode); 1894 last_inode = NULL; 1895 } 1896 if (old_cap) { 1897 ceph_put_cap(session->s_mdsc, old_cap); 1898 old_cap = NULL; 1899 } 1900 1901 ret = cb(inode, mds, arg); 1902 last_inode = inode; 1903 1904 spin_lock(&session->s_cap_lock); 1905 p = p->next; 1906 if (!cap->ci) { 1907 doutc(cl, "finishing cap %p removal\n", cap); 1908 BUG_ON(cap->session != session); 1909 cap->session = NULL; 1910 list_del_init(&cap->session_caps); 1911 session->s_nr_caps--; 1912 atomic64_dec(&session->s_mdsc->metric.total_caps); 1913 if (cap->queue_release) 1914 __ceph_queue_cap_release(session, cap); 1915 else 1916 old_cap = cap; /* put_cap it w/o locks held */ 1917 } 1918 if (ret < 0) 1919 goto out; 1920 } 1921 ret = 0; 1922 out: 1923 session->s_cap_iterator = NULL; 1924 spin_unlock(&session->s_cap_lock); 1925 1926 iput(last_inode); 1927 if (old_cap) 1928 ceph_put_cap(session->s_mdsc, old_cap); 1929 1930 return ret; 1931 } 1932 1933 static int remove_session_caps_cb(struct inode *inode, int mds, void *arg) 1934 { 1935 struct ceph_inode_info *ci = ceph_inode(inode); 1936 struct ceph_client *cl = ceph_inode_to_client(inode); 1937 bool invalidate = false; 1938 struct ceph_cap *cap; 1939 int iputs = 0; 1940 1941 spin_lock(&ci->i_ceph_lock); 1942 cap = __get_cap_for_mds(ci, mds); 1943 if (cap) { 1944 doutc(cl, " removing cap %p, ci is %p, inode is %p\n", 1945 cap, ci, &ci->netfs.inode); 1946 1947 iputs = ceph_purge_inode_cap(inode, cap, &invalidate); 1948 } 1949 spin_unlock(&ci->i_ceph_lock); 1950 1951 if (cap) 1952 wake_up_all(&ci->i_cap_wq); 1953 if (invalidate) 1954 ceph_queue_invalidate(inode); 1955 while (iputs--) 1956 iput(inode); 1957 return 0; 1958 } 1959 1960 /* 1961 * caller must hold session s_mutex 1962 */ 1963 static void remove_session_caps(struct ceph_mds_session *session) 1964 { 1965 struct ceph_fs_client *fsc = session->s_mdsc->fsc; 1966 struct super_block *sb = fsc->sb; 1967 LIST_HEAD(dispose); 1968 1969 doutc(fsc->client, "on %p\n", session); 1970 ceph_iterate_session_caps(session, remove_session_caps_cb, fsc); 1971 1972 wake_up_all(&fsc->mdsc->cap_flushing_wq); 1973 1974 spin_lock(&session->s_cap_lock); 1975 if (session->s_nr_caps > 0) { 1976 struct inode *inode; 1977 struct ceph_cap *cap, *prev = NULL; 1978 struct ceph_vino vino; 1979 /* 1980 * iterate_session_caps() skips inodes that are being 1981 * deleted, we need to wait until deletions are complete. 1982 * __wait_on_freeing_inode() is designed for the job, 1983 * but it is not exported, so use lookup inode function 1984 * to access it. 1985 */ 1986 while (!list_empty(&session->s_caps)) { 1987 cap = list_entry(session->s_caps.next, 1988 struct ceph_cap, session_caps); 1989 if (cap == prev) 1990 break; 1991 prev = cap; 1992 vino = cap->ci->i_vino; 1993 spin_unlock(&session->s_cap_lock); 1994 1995 inode = ceph_find_inode(sb, vino); 1996 iput(inode); 1997 1998 spin_lock(&session->s_cap_lock); 1999 } 2000 } 2001 2002 // drop cap expires and unlock s_cap_lock 2003 detach_cap_releases(session, &dispose); 2004 2005 BUG_ON(session->s_nr_caps > 0); 2006 BUG_ON(!list_empty(&session->s_cap_flushing)); 2007 spin_unlock(&session->s_cap_lock); 2008 dispose_cap_releases(session->s_mdsc, &dispose); 2009 } 2010 2011 enum { 2012 RECONNECT, 2013 RENEWCAPS, 2014 FORCE_RO, 2015 }; 2016 2017 /* 2018 * wake up any threads waiting on this session's caps. if the cap is 2019 * old (didn't get renewed on the client reconnect), remove it now. 2020 * 2021 * caller must hold s_mutex. 2022 */ 2023 static int wake_up_session_cb(struct inode *inode, int mds, void *arg) 2024 { 2025 struct ceph_inode_info *ci = ceph_inode(inode); 2026 unsigned long ev = (unsigned long)arg; 2027 2028 if (ev == RECONNECT) { 2029 spin_lock(&ci->i_ceph_lock); 2030 ci->i_wanted_max_size = 0; 2031 ci->i_requested_max_size = 0; 2032 spin_unlock(&ci->i_ceph_lock); 2033 } else if (ev == RENEWCAPS) { 2034 struct ceph_cap *cap; 2035 2036 spin_lock(&ci->i_ceph_lock); 2037 cap = __get_cap_for_mds(ci, mds); 2038 /* mds did not re-issue stale cap */ 2039 if (cap && cap->cap_gen < atomic_read(&cap->session->s_cap_gen)) 2040 cap->issued = cap->implemented = CEPH_CAP_PIN; 2041 spin_unlock(&ci->i_ceph_lock); 2042 } else if (ev == FORCE_RO) { 2043 } 2044 wake_up_all(&ci->i_cap_wq); 2045 return 0; 2046 } 2047 2048 static void wake_up_session_caps(struct ceph_mds_session *session, int ev) 2049 { 2050 struct ceph_client *cl = session->s_mdsc->fsc->client; 2051 2052 doutc(cl, "session %p mds%d\n", session, session->s_mds); 2053 ceph_iterate_session_caps(session, wake_up_session_cb, 2054 (void *)(unsigned long)ev); 2055 } 2056 2057 /* 2058 * Send periodic message to MDS renewing all currently held caps. The 2059 * ack will reset the expiration for all caps from this session. 2060 * 2061 * caller holds s_mutex 2062 */ 2063 static int send_renew_caps(struct ceph_mds_client *mdsc, 2064 struct ceph_mds_session *session) 2065 { 2066 struct ceph_client *cl = mdsc->fsc->client; 2067 struct ceph_msg *msg; 2068 int state; 2069 2070 if (time_after_eq(jiffies, session->s_cap_ttl) && 2071 time_after_eq(session->s_cap_ttl, session->s_renew_requested)) 2072 pr_info_client(cl, "mds%d caps stale\n", session->s_mds); 2073 session->s_renew_requested = jiffies; 2074 2075 /* do not try to renew caps until a recovering mds has reconnected 2076 * with its clients. */ 2077 state = ceph_mdsmap_get_state(mdsc->mdsmap, session->s_mds); 2078 if (state < CEPH_MDS_STATE_RECONNECT) { 2079 doutc(cl, "ignoring mds%d (%s)\n", session->s_mds, 2080 ceph_mds_state_name(state)); 2081 return 0; 2082 } 2083 2084 doutc(cl, "to mds%d (%s)\n", session->s_mds, 2085 ceph_mds_state_name(state)); 2086 msg = create_session_full_msg(mdsc, CEPH_SESSION_REQUEST_RENEWCAPS, 2087 ++session->s_renew_seq); 2088 if (IS_ERR(msg)) 2089 return PTR_ERR(msg); 2090 ceph_con_send(&session->s_con, msg); 2091 return 0; 2092 } 2093 2094 static int send_flushmsg_ack(struct ceph_mds_client *mdsc, 2095 struct ceph_mds_session *session, u64 seq) 2096 { 2097 struct ceph_client *cl = mdsc->fsc->client; 2098 struct ceph_msg *msg; 2099 2100 doutc(cl, "to mds%d (%s)s seq %lld\n", session->s_mds, 2101 ceph_session_state_name(session->s_state), seq); 2102 msg = ceph_create_session_msg(CEPH_SESSION_FLUSHMSG_ACK, seq); 2103 if (!msg) 2104 return -ENOMEM; 2105 ceph_con_send(&session->s_con, msg); 2106 return 0; 2107 } 2108 2109 2110 /* 2111 * Note new cap ttl, and any transition from stale -> not stale (fresh?). 2112 * 2113 * Called under session->s_mutex 2114 */ 2115 static void renewed_caps(struct ceph_mds_client *mdsc, 2116 struct ceph_mds_session *session, int is_renew) 2117 { 2118 struct ceph_client *cl = mdsc->fsc->client; 2119 int was_stale; 2120 int wake = 0; 2121 2122 spin_lock(&session->s_cap_lock); 2123 was_stale = is_renew && time_after_eq(jiffies, session->s_cap_ttl); 2124 2125 session->s_cap_ttl = session->s_renew_requested + 2126 mdsc->mdsmap->m_session_timeout*HZ; 2127 2128 if (was_stale) { 2129 if (time_before(jiffies, session->s_cap_ttl)) { 2130 pr_info_client(cl, "mds%d caps renewed\n", 2131 session->s_mds); 2132 wake = 1; 2133 } else { 2134 pr_info_client(cl, "mds%d caps still stale\n", 2135 session->s_mds); 2136 } 2137 } 2138 doutc(cl, "mds%d ttl now %lu, was %s, now %s\n", session->s_mds, 2139 session->s_cap_ttl, was_stale ? "stale" : "fresh", 2140 time_before(jiffies, session->s_cap_ttl) ? "stale" : "fresh"); 2141 spin_unlock(&session->s_cap_lock); 2142 2143 if (wake) 2144 wake_up_session_caps(session, RENEWCAPS); 2145 } 2146 2147 /* 2148 * send a session close request 2149 */ 2150 static int request_close_session(struct ceph_mds_session *session) 2151 { 2152 struct ceph_client *cl = session->s_mdsc->fsc->client; 2153 struct ceph_msg *msg; 2154 2155 doutc(cl, "mds%d state %s seq %lld\n", session->s_mds, 2156 ceph_session_state_name(session->s_state), session->s_seq); 2157 msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_CLOSE, 2158 session->s_seq); 2159 if (!msg) 2160 return -ENOMEM; 2161 ceph_con_send(&session->s_con, msg); 2162 return 1; 2163 } 2164 2165 /* 2166 * Called with s_mutex held. 2167 */ 2168 static int __close_session(struct ceph_mds_client *mdsc, 2169 struct ceph_mds_session *session) 2170 { 2171 if (session->s_state >= CEPH_MDS_SESSION_CLOSING) 2172 return 0; 2173 session->s_state = CEPH_MDS_SESSION_CLOSING; 2174 return request_close_session(session); 2175 } 2176 2177 static bool drop_negative_children(struct dentry *dentry) 2178 { 2179 struct dentry *child; 2180 bool all_negative = true; 2181 2182 if (!d_is_dir(dentry)) 2183 goto out; 2184 2185 spin_lock(&dentry->d_lock); 2186 hlist_for_each_entry(child, &dentry->d_children, d_sib) { 2187 if (d_really_is_positive(child)) { 2188 all_negative = false; 2189 break; 2190 } 2191 } 2192 spin_unlock(&dentry->d_lock); 2193 2194 if (all_negative) 2195 shrink_dcache_parent(dentry); 2196 out: 2197 return all_negative; 2198 } 2199 2200 /* 2201 * Trim old(er) caps. 2202 * 2203 * Because we can't cache an inode without one or more caps, we do 2204 * this indirectly: if a cap is unused, we prune its aliases, at which 2205 * point the inode will hopefully get dropped to. 2206 * 2207 * Yes, this is a bit sloppy. Our only real goal here is to respond to 2208 * memory pressure from the MDS, though, so it needn't be perfect. 2209 */ 2210 static int trim_caps_cb(struct inode *inode, int mds, void *arg) 2211 { 2212 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb); 2213 struct ceph_client *cl = mdsc->fsc->client; 2214 int *remaining = arg; 2215 struct ceph_inode_info *ci = ceph_inode(inode); 2216 int used, wanted, oissued, mine; 2217 struct ceph_cap *cap; 2218 2219 if (*remaining <= 0) 2220 return -1; 2221 2222 spin_lock(&ci->i_ceph_lock); 2223 cap = __get_cap_for_mds(ci, mds); 2224 if (!cap) { 2225 spin_unlock(&ci->i_ceph_lock); 2226 return 0; 2227 } 2228 mine = cap->issued | cap->implemented; 2229 used = __ceph_caps_used(ci); 2230 wanted = __ceph_caps_file_wanted(ci); 2231 oissued = __ceph_caps_issued_other(ci, cap); 2232 2233 doutc(cl, "%p %llx.%llx cap %p mine %s oissued %s used %s wanted %s\n", 2234 inode, ceph_vinop(inode), cap, ceph_cap_string(mine), 2235 ceph_cap_string(oissued), ceph_cap_string(used), 2236 ceph_cap_string(wanted)); 2237 if (cap == ci->i_auth_cap) { 2238 if (ci->i_dirty_caps || ci->i_flushing_caps || 2239 !list_empty(&ci->i_cap_snaps)) 2240 goto out; 2241 if ((used | wanted) & CEPH_CAP_ANY_WR) 2242 goto out; 2243 /* Note: it's possible that i_filelock_ref becomes non-zero 2244 * after dropping auth caps. It doesn't hurt because reply 2245 * of lock mds request will re-add auth caps. */ 2246 if (atomic_read(&ci->i_filelock_ref) > 0) 2247 goto out; 2248 } 2249 /* The inode has cached pages, but it's no longer used. 2250 * we can safely drop it */ 2251 if (S_ISREG(inode->i_mode) && 2252 wanted == 0 && used == CEPH_CAP_FILE_CACHE && 2253 !(oissued & CEPH_CAP_FILE_CACHE)) { 2254 used = 0; 2255 oissued = 0; 2256 } 2257 if ((used | wanted) & ~oissued & mine) 2258 goto out; /* we need these caps */ 2259 2260 if (oissued) { 2261 /* we aren't the only cap.. just remove us */ 2262 ceph_remove_cap(mdsc, cap, true); 2263 (*remaining)--; 2264 } else { 2265 struct dentry *dentry; 2266 /* try dropping referring dentries */ 2267 spin_unlock(&ci->i_ceph_lock); 2268 dentry = d_find_any_alias(inode); 2269 if (dentry && drop_negative_children(dentry)) { 2270 int count; 2271 dput(dentry); 2272 d_prune_aliases(inode); 2273 count = icount_read_once(inode); 2274 if (count == 1) 2275 (*remaining)--; 2276 doutc(cl, "%p %llx.%llx cap %p pruned, count now %d\n", 2277 inode, ceph_vinop(inode), cap, count); 2278 } else { 2279 dput(dentry); 2280 } 2281 return 0; 2282 } 2283 2284 out: 2285 spin_unlock(&ci->i_ceph_lock); 2286 return 0; 2287 } 2288 2289 /* 2290 * Trim session cap count down to some max number. 2291 */ 2292 int ceph_trim_caps(struct ceph_mds_client *mdsc, 2293 struct ceph_mds_session *session, 2294 int max_caps) 2295 { 2296 struct ceph_client *cl = mdsc->fsc->client; 2297 int trim_caps = session->s_nr_caps - max_caps; 2298 2299 doutc(cl, "mds%d start: %d / %d, trim %d\n", session->s_mds, 2300 session->s_nr_caps, max_caps, trim_caps); 2301 if (trim_caps > 0) { 2302 int remaining = trim_caps; 2303 2304 ceph_iterate_session_caps(session, trim_caps_cb, &remaining); 2305 doutc(cl, "mds%d done: %d / %d, trimmed %d\n", 2306 session->s_mds, session->s_nr_caps, max_caps, 2307 trim_caps - remaining); 2308 } 2309 2310 ceph_flush_session_cap_releases(mdsc, session); 2311 return 0; 2312 } 2313 2314 static int check_caps_flush(struct ceph_mds_client *mdsc, 2315 u64 want_flush_tid) 2316 { 2317 struct ceph_client *cl = mdsc->fsc->client; 2318 int ret = 1; 2319 2320 spin_lock(&mdsc->cap_dirty_lock); 2321 if (!list_empty(&mdsc->cap_flush_list)) { 2322 struct ceph_cap_flush *cf = 2323 list_first_entry(&mdsc->cap_flush_list, 2324 struct ceph_cap_flush, g_list); 2325 if (cf->tid <= want_flush_tid) { 2326 doutc(cl, "still flushing tid %llu <= %llu\n", 2327 cf->tid, want_flush_tid); 2328 ret = 0; 2329 } 2330 } 2331 spin_unlock(&mdsc->cap_dirty_lock); 2332 return ret; 2333 } 2334 2335 /* 2336 * Snapshot of a single cap_flush entry for diagnostic dump. 2337 * Collected under cap_dirty_lock, printed after releasing it. 2338 */ 2339 struct flush_dump_entry { 2340 u64 ino; /* inode number */ 2341 u64 snap; /* snap id */ 2342 int caps; /* dirty cap bits */ 2343 u64 tid; /* flush transaction id */ 2344 u64 last_ack; /* most recent ack tid for this inode */ 2345 bool wake; /* whether completion was requested */ 2346 bool is_capsnap; /* true if this is a cap snap flush */ 2347 bool ci_null; /* true if cf->ci was unexpectedly NULL */ 2348 }; 2349 2350 /* 2351 * Dump pending cap flushes for diagnostic purposes. 2352 * 2353 * cf->ci is safe to dereference here: cap_flush entries hold a 2354 * reference on the inode (via the cap), and entries are removed from 2355 * cap_flush_list under cap_dirty_lock before the cap (and thus the 2356 * inode reference) is released. Holding cap_dirty_lock therefore 2357 * guarantees the inode remains valid for the lifetime of the scan. 2358 */ 2359 2360 static void dump_cap_flushes(struct ceph_mds_client *mdsc, u64 want_tid) 2361 { 2362 struct ceph_client *cl = mdsc->fsc->client; 2363 struct flush_dump_entry entries[CEPH_CAP_FLUSH_MAX_DUMP_ENTRIES]; 2364 struct ceph_cap_flush *cf; 2365 int n = 0, remaining = 0; 2366 int i; 2367 2368 spin_lock(&mdsc->cap_dirty_lock); 2369 list_for_each_entry(cf, &mdsc->cap_flush_list, g_list) { 2370 if (cf->tid > want_tid) 2371 break; 2372 if (n < CEPH_CAP_FLUSH_MAX_DUMP_ENTRIES) { 2373 struct flush_dump_entry *e = &entries[n++]; 2374 2375 e->ci_null = WARN_ON_ONCE(!cf->ci); 2376 if (!e->ci_null) { 2377 e->ino = ceph_ino(&cf->ci->netfs.inode); 2378 e->snap = ceph_snap(&cf->ci->netfs.inode); 2379 e->last_ack = READ_ONCE(cf->ci->i_last_cap_flush_ack); 2380 } 2381 e->caps = cf->caps; 2382 e->tid = cf->tid; 2383 e->wake = cf->wake; 2384 e->is_capsnap = cf->is_capsnap; 2385 } else { 2386 remaining++; 2387 } 2388 } 2389 spin_unlock(&mdsc->cap_dirty_lock); 2390 2391 pr_info_client(cl, "still waiting for cap flushes through %llu:\n", 2392 want_tid); 2393 for (i = 0; i < n; i++) { 2394 struct flush_dump_entry *e = &entries[i]; 2395 2396 if (e->ci_null) 2397 pr_info_client(cl, 2398 " (null ci) %s tid=%llu wake=%d%s\n", 2399 ceph_cap_string(e->caps), e->tid, 2400 e->wake, 2401 e->is_capsnap ? " is_capsnap" : ""); 2402 else 2403 pr_info_client(cl, 2404 " %llx.%llx %s tid=%llu last_ack=%llu wake=%d%s\n", 2405 e->ino, e->snap, 2406 ceph_cap_string(e->caps), e->tid, 2407 e->last_ack, e->wake, 2408 e->is_capsnap ? " is_capsnap" : ""); 2409 } 2410 if (remaining) 2411 pr_info_client(cl, " ... and %d more pending flushes\n", 2412 remaining); 2413 } 2414 2415 /* 2416 * Wait for all cap flushes through @want_flush_tid to complete. 2417 * Periodically dumps pending cap flush state for diagnostics. 2418 */ 2419 static void wait_caps_flush(struct ceph_mds_client *mdsc, 2420 u64 want_flush_tid) 2421 { 2422 struct ceph_client *cl = mdsc->fsc->client; 2423 int i = 0; 2424 long ret; 2425 2426 doutc(cl, "want %llu\n", want_flush_tid); 2427 2428 do { 2429 /* 60 * HZ fits in a long on all supported architectures. */ 2430 ret = wait_event_timeout(mdsc->cap_flushing_wq, 2431 check_caps_flush(mdsc, want_flush_tid), 2432 CEPH_CAP_FLUSH_WAIT_TIMEOUT_SEC * HZ); 2433 if (ret == 0) { 2434 if (i < CEPH_CAP_FLUSH_MAX_DUMP_ITERS) 2435 dump_cap_flushes(mdsc, want_flush_tid); 2436 else if (i == CEPH_CAP_FLUSH_MAX_DUMP_ITERS) 2437 pr_info_client(cl, 2438 "still waiting for cap flushes; suppressing further dumps\n"); 2439 i++; 2440 } 2441 } while (ret == 0); 2442 2443 doutc(cl, "ok, flushed thru %llu\n", want_flush_tid); 2444 } 2445 2446 /* 2447 * called under s_mutex 2448 */ 2449 static void ceph_send_cap_releases(struct ceph_mds_client *mdsc, 2450 struct ceph_mds_session *session) 2451 { 2452 struct ceph_client *cl = mdsc->fsc->client; 2453 struct ceph_msg *msg = NULL; 2454 struct ceph_mds_cap_release *head; 2455 struct ceph_mds_cap_item *item; 2456 struct ceph_osd_client *osdc = &mdsc->fsc->client->osdc; 2457 struct ceph_cap *cap; 2458 LIST_HEAD(tmp_list); 2459 int num_cap_releases; 2460 __le32 barrier, *cap_barrier; 2461 2462 down_read(&osdc->lock); 2463 barrier = cpu_to_le32(osdc->epoch_barrier); 2464 up_read(&osdc->lock); 2465 2466 spin_lock(&session->s_cap_lock); 2467 again: 2468 list_splice_init(&session->s_cap_releases, &tmp_list); 2469 num_cap_releases = session->s_num_cap_releases; 2470 session->s_num_cap_releases = 0; 2471 spin_unlock(&session->s_cap_lock); 2472 2473 while (!list_empty(&tmp_list)) { 2474 if (!msg) { 2475 msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPRELEASE, 2476 PAGE_SIZE, GFP_NOFS, false); 2477 if (!msg) 2478 goto out_err; 2479 head = msg->front.iov_base; 2480 head->num = cpu_to_le32(0); 2481 msg->front.iov_len = sizeof(*head); 2482 2483 msg->hdr.version = cpu_to_le16(2); 2484 msg->hdr.compat_version = cpu_to_le16(1); 2485 } 2486 2487 cap = list_first_entry(&tmp_list, struct ceph_cap, 2488 session_caps); 2489 list_del(&cap->session_caps); 2490 num_cap_releases--; 2491 2492 head = msg->front.iov_base; 2493 put_unaligned_le32(get_unaligned_le32(&head->num) + 1, 2494 &head->num); 2495 item = msg->front.iov_base + msg->front.iov_len; 2496 item->ino = cpu_to_le64(cap->cap_ino); 2497 item->cap_id = cpu_to_le64(cap->cap_id); 2498 item->migrate_seq = cpu_to_le32(cap->mseq); 2499 item->issue_seq = cpu_to_le32(cap->issue_seq); 2500 msg->front.iov_len += sizeof(*item); 2501 2502 ceph_put_cap(mdsc, cap); 2503 2504 if (le32_to_cpu(head->num) == CEPH_CAPS_PER_RELEASE) { 2505 // Append cap_barrier field 2506 cap_barrier = msg->front.iov_base + msg->front.iov_len; 2507 *cap_barrier = barrier; 2508 msg->front.iov_len += sizeof(*cap_barrier); 2509 2510 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len); 2511 doutc(cl, "mds%d %p\n", session->s_mds, msg); 2512 ceph_con_send(&session->s_con, msg); 2513 msg = NULL; 2514 } 2515 } 2516 2517 BUG_ON(num_cap_releases != 0); 2518 2519 spin_lock(&session->s_cap_lock); 2520 if (!list_empty(&session->s_cap_releases)) 2521 goto again; 2522 spin_unlock(&session->s_cap_lock); 2523 2524 if (msg) { 2525 // Append cap_barrier field 2526 cap_barrier = msg->front.iov_base + msg->front.iov_len; 2527 *cap_barrier = barrier; 2528 msg->front.iov_len += sizeof(*cap_barrier); 2529 2530 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len); 2531 doutc(cl, "mds%d %p\n", session->s_mds, msg); 2532 ceph_con_send(&session->s_con, msg); 2533 } 2534 return; 2535 out_err: 2536 pr_err_client(cl, "mds%d, failed to allocate message\n", 2537 session->s_mds); 2538 spin_lock(&session->s_cap_lock); 2539 list_splice(&tmp_list, &session->s_cap_releases); 2540 session->s_num_cap_releases += num_cap_releases; 2541 spin_unlock(&session->s_cap_lock); 2542 } 2543 2544 static void ceph_cap_release_work(struct work_struct *work) 2545 { 2546 struct ceph_mds_session *session = 2547 container_of(work, struct ceph_mds_session, s_cap_release_work); 2548 2549 mutex_lock(&session->s_mutex); 2550 if (session->s_state == CEPH_MDS_SESSION_OPEN || 2551 session->s_state == CEPH_MDS_SESSION_HUNG) 2552 ceph_send_cap_releases(session->s_mdsc, session); 2553 mutex_unlock(&session->s_mutex); 2554 ceph_put_mds_session(session); 2555 } 2556 2557 void ceph_flush_session_cap_releases(struct ceph_mds_client *mdsc, 2558 struct ceph_mds_session *session) 2559 { 2560 struct ceph_client *cl = mdsc->fsc->client; 2561 if (mdsc->stopping) 2562 return; 2563 2564 ceph_get_mds_session(session); 2565 if (queue_work(mdsc->fsc->cap_wq, 2566 &session->s_cap_release_work)) { 2567 doutc(cl, "cap release work queued\n"); 2568 } else { 2569 ceph_put_mds_session(session); 2570 doutc(cl, "failed to queue cap release work\n"); 2571 } 2572 } 2573 2574 /* 2575 * caller holds session->s_cap_lock 2576 */ 2577 void __ceph_queue_cap_release(struct ceph_mds_session *session, 2578 struct ceph_cap *cap) 2579 { 2580 list_add_tail(&cap->session_caps, &session->s_cap_releases); 2581 session->s_num_cap_releases++; 2582 2583 if (!(session->s_num_cap_releases % CEPH_CAPS_PER_RELEASE)) 2584 ceph_flush_session_cap_releases(session->s_mdsc, session); 2585 } 2586 2587 static void ceph_cap_reclaim_work(struct work_struct *work) 2588 { 2589 struct ceph_mds_client *mdsc = 2590 container_of(work, struct ceph_mds_client, cap_reclaim_work); 2591 int ret = ceph_trim_dentries(mdsc); 2592 if (ret == -EAGAIN) 2593 ceph_queue_cap_reclaim_work(mdsc); 2594 } 2595 2596 void ceph_queue_cap_reclaim_work(struct ceph_mds_client *mdsc) 2597 { 2598 struct ceph_client *cl = mdsc->fsc->client; 2599 if (mdsc->stopping) 2600 return; 2601 2602 if (queue_work(mdsc->fsc->cap_wq, &mdsc->cap_reclaim_work)) { 2603 doutc(cl, "caps reclaim work queued\n"); 2604 } else { 2605 doutc(cl, "failed to queue caps release work\n"); 2606 } 2607 } 2608 2609 void ceph_reclaim_caps_nr(struct ceph_mds_client *mdsc, int nr) 2610 { 2611 int val; 2612 if (!nr) 2613 return; 2614 val = atomic_add_return(nr, &mdsc->cap_reclaim_pending); 2615 if ((val % CEPH_CAPS_PER_RELEASE) < nr) { 2616 atomic_set(&mdsc->cap_reclaim_pending, 0); 2617 ceph_queue_cap_reclaim_work(mdsc); 2618 } 2619 } 2620 2621 void ceph_queue_cap_unlink_work(struct ceph_mds_client *mdsc) 2622 { 2623 struct ceph_client *cl = mdsc->fsc->client; 2624 if (mdsc->stopping) 2625 return; 2626 2627 if (queue_work(mdsc->fsc->cap_wq, &mdsc->cap_unlink_work)) { 2628 doutc(cl, "caps unlink work queued\n"); 2629 } else { 2630 doutc(cl, "failed to queue caps unlink work\n"); 2631 } 2632 } 2633 2634 static void ceph_cap_unlink_work(struct work_struct *work) 2635 { 2636 struct ceph_mds_client *mdsc = 2637 container_of(work, struct ceph_mds_client, cap_unlink_work); 2638 struct ceph_client *cl = mdsc->fsc->client; 2639 2640 doutc(cl, "begin\n"); 2641 spin_lock(&mdsc->cap_delay_lock); 2642 while (!list_empty(&mdsc->cap_unlink_delay_list)) { 2643 struct ceph_inode_info *ci; 2644 struct inode *inode; 2645 2646 ci = list_first_entry(&mdsc->cap_unlink_delay_list, 2647 struct ceph_inode_info, 2648 i_cap_delay_list); 2649 list_del_init(&ci->i_cap_delay_list); 2650 2651 inode = igrab(&ci->netfs.inode); 2652 if (inode) { 2653 spin_unlock(&mdsc->cap_delay_lock); 2654 doutc(cl, "on %p %llx.%llx\n", inode, 2655 ceph_vinop(inode)); 2656 ceph_check_caps(ci, CHECK_CAPS_FLUSH); 2657 iput(inode); 2658 spin_lock(&mdsc->cap_delay_lock); 2659 } 2660 } 2661 spin_unlock(&mdsc->cap_delay_lock); 2662 doutc(cl, "done\n"); 2663 } 2664 2665 /* 2666 * requests 2667 */ 2668 2669 int ceph_alloc_readdir_reply_buffer(struct ceph_mds_request *req, 2670 struct inode *dir) 2671 { 2672 struct ceph_inode_info *ci = ceph_inode(dir); 2673 struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info; 2674 struct ceph_mount_options *opt = req->r_mdsc->fsc->mount_options; 2675 size_t size = sizeof(struct ceph_mds_reply_dir_entry); 2676 unsigned int num_entries; 2677 u64 bytes_count; 2678 int order; 2679 2680 spin_lock(&ci->i_ceph_lock); 2681 num_entries = ci->i_files + ci->i_subdirs; 2682 spin_unlock(&ci->i_ceph_lock); 2683 num_entries = max(num_entries, 1U); 2684 num_entries = min(num_entries, opt->max_readdir); 2685 2686 bytes_count = (u64)size * num_entries; 2687 if (unlikely(bytes_count > ULONG_MAX)) 2688 bytes_count = ULONG_MAX; 2689 2690 order = get_order((unsigned long)bytes_count); 2691 while (order >= 0) { 2692 rinfo->dir_entries = (void*)__get_free_pages(GFP_KERNEL | 2693 __GFP_NOWARN | 2694 __GFP_ZERO, 2695 order); 2696 if (rinfo->dir_entries) 2697 break; 2698 order--; 2699 } 2700 if (!rinfo->dir_entries || unlikely(order < 0)) 2701 return -ENOMEM; 2702 2703 num_entries = (PAGE_SIZE << order) / size; 2704 num_entries = min(num_entries, opt->max_readdir); 2705 2706 rinfo->dir_buf_size = PAGE_SIZE << order; 2707 req->r_num_caps = num_entries + 1; 2708 req->r_args.readdir.max_entries = cpu_to_le32(num_entries); 2709 req->r_args.readdir.max_bytes = cpu_to_le32(opt->max_readdir_bytes); 2710 return 0; 2711 } 2712 2713 /* 2714 * Create an mds request. 2715 */ 2716 struct ceph_mds_request * 2717 ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode) 2718 { 2719 struct ceph_mds_request *req; 2720 2721 req = kmem_cache_zalloc(ceph_mds_request_cachep, GFP_NOFS); 2722 if (!req) 2723 return ERR_PTR(-ENOMEM); 2724 2725 mutex_init(&req->r_fill_mutex); 2726 req->r_mdsc = mdsc; 2727 req->r_started = jiffies; 2728 req->r_start_latency = ktime_get(); 2729 req->r_resend_mds = -1; 2730 INIT_LIST_HEAD(&req->r_unsafe_dir_item); 2731 INIT_LIST_HEAD(&req->r_unsafe_target_item); 2732 req->r_fmode = -1; 2733 req->r_feature_needed = -1; 2734 kref_init(&req->r_kref); 2735 RB_CLEAR_NODE(&req->r_node); 2736 INIT_LIST_HEAD(&req->r_wait); 2737 init_completion(&req->r_completion); 2738 init_completion(&req->r_safe_completion); 2739 INIT_LIST_HEAD(&req->r_unsafe_item); 2740 2741 ktime_get_coarse_real_ts64(&req->r_stamp); 2742 2743 req->r_op = op; 2744 req->r_direct_mode = mode; 2745 return req; 2746 } 2747 2748 /* 2749 * return oldest (lowest) request, tid in request tree, 0 if none. 2750 * 2751 * called under mdsc->mutex. 2752 */ 2753 static struct ceph_mds_request *__get_oldest_req(struct ceph_mds_client *mdsc) 2754 { 2755 if (RB_EMPTY_ROOT(&mdsc->request_tree)) 2756 return NULL; 2757 return rb_entry(rb_first(&mdsc->request_tree), 2758 struct ceph_mds_request, r_node); 2759 } 2760 2761 static inline u64 __get_oldest_tid(struct ceph_mds_client *mdsc) 2762 { 2763 return mdsc->oldest_tid; 2764 } 2765 2766 #if IS_ENABLED(CONFIG_FS_ENCRYPTION) 2767 static u8 *get_fscrypt_altname(const struct ceph_mds_request *req, u32 *plen) 2768 { 2769 struct inode *dir = req->r_parent; 2770 struct dentry *dentry = req->r_dentry; 2771 const struct qstr *name = req->r_dname; 2772 u8 *cryptbuf = NULL; 2773 u32 len = 0; 2774 int ret = 0; 2775 2776 /* only encode if we have parent and dentry */ 2777 if (!dir || !dentry) 2778 goto success; 2779 2780 /* No-op unless this is encrypted */ 2781 if (!IS_ENCRYPTED(dir)) 2782 goto success; 2783 2784 ret = ceph_fscrypt_prepare_readdir(dir); 2785 if (ret < 0) 2786 return ERR_PTR(ret); 2787 2788 /* No key? Just ignore it. */ 2789 if (!fscrypt_has_encryption_key(dir)) 2790 goto success; 2791 2792 if (!name) 2793 name = &dentry->d_name; 2794 2795 if (!fscrypt_fname_encrypted_size(dir, name->len, NAME_MAX, &len)) { 2796 WARN_ON_ONCE(1); 2797 return ERR_PTR(-ENAMETOOLONG); 2798 } 2799 2800 /* No need to append altname if name is short enough */ 2801 if (len <= CEPH_NOHASH_NAME_MAX) { 2802 len = 0; 2803 goto success; 2804 } 2805 2806 cryptbuf = kmalloc(len, GFP_KERNEL); 2807 if (!cryptbuf) 2808 return ERR_PTR(-ENOMEM); 2809 2810 ret = fscrypt_fname_encrypt(dir, name, cryptbuf, len); 2811 if (ret) { 2812 kfree(cryptbuf); 2813 return ERR_PTR(ret); 2814 } 2815 success: 2816 *plen = len; 2817 return cryptbuf; 2818 } 2819 #else 2820 static u8 *get_fscrypt_altname(const struct ceph_mds_request *req, u32 *plen) 2821 { 2822 *plen = 0; 2823 return NULL; 2824 } 2825 #endif 2826 2827 /** 2828 * ceph_mdsc_build_path - build a path string to a given dentry 2829 * @mdsc: mds client 2830 * @dentry: dentry to which path should be built 2831 * @path_info: output path, length, base ino+snap, and freepath ownership flag 2832 * @for_wire: is this path going to be sent to the MDS? 2833 * 2834 * Build a string that represents the path to the dentry. This is mostly called 2835 * for two different purposes: 2836 * 2837 * 1) we need to build a path string to send to the MDS (for_wire == true) 2838 * 2) we need a path string for local presentation (e.g. debugfs) 2839 * (for_wire == false) 2840 * 2841 * The path is built in reverse, starting with the dentry. Walk back up toward 2842 * the root, building the path until the first non-snapped inode is reached 2843 * (for_wire) or the root inode is reached (!for_wire). 2844 * 2845 * Encode hidden .snap dirs as a double /, i.e. 2846 * foo/.snap/bar -> foo//bar 2847 */ 2848 char *ceph_mdsc_build_path(struct ceph_mds_client *mdsc, struct dentry *dentry, 2849 struct ceph_path_info *path_info, int for_wire) 2850 { 2851 struct ceph_client *cl = mdsc->fsc->client; 2852 struct dentry *cur; 2853 struct inode *inode; 2854 char *path; 2855 int pos; 2856 unsigned seq; 2857 u64 base; 2858 2859 if (!dentry) 2860 return ERR_PTR(-EINVAL); 2861 2862 path = __getname(); 2863 if (!path) 2864 return ERR_PTR(-ENOMEM); 2865 retry: 2866 pos = PATH_MAX - 1; 2867 path[pos] = '\0'; 2868 2869 seq = read_seqbegin(&rename_lock); 2870 cur = dget(dentry); 2871 for (;;) { 2872 struct dentry *parent; 2873 2874 spin_lock(&cur->d_lock); 2875 inode = d_inode(cur); 2876 if (inode && ceph_snap(inode) == CEPH_SNAPDIR) { 2877 doutc(cl, "path+%d: %p SNAPDIR\n", pos, cur); 2878 spin_unlock(&cur->d_lock); 2879 parent = dget_parent(cur); 2880 } else if (for_wire && inode && dentry != cur && 2881 ceph_snap(inode) == CEPH_NOSNAP) { 2882 spin_unlock(&cur->d_lock); 2883 pos++; /* get rid of any prepended '/' */ 2884 break; 2885 } else if (!for_wire || !IS_ENCRYPTED(d_inode(cur->d_parent))) { 2886 pos -= cur->d_name.len; 2887 if (pos < 0) { 2888 spin_unlock(&cur->d_lock); 2889 break; 2890 } 2891 memcpy(path + pos, cur->d_name.name, cur->d_name.len); 2892 spin_unlock(&cur->d_lock); 2893 parent = dget_parent(cur); 2894 } else { 2895 int len, ret; 2896 char buf[NAME_MAX]; 2897 2898 /* 2899 * Proactively copy name into buf, in case we need to 2900 * present it as-is. 2901 */ 2902 memcpy(buf, cur->d_name.name, cur->d_name.len); 2903 len = cur->d_name.len; 2904 spin_unlock(&cur->d_lock); 2905 parent = dget_parent(cur); 2906 2907 ret = ceph_fscrypt_prepare_readdir(d_inode(parent)); 2908 if (ret < 0) { 2909 dput(parent); 2910 dput(cur); 2911 __putname(path); 2912 return ERR_PTR(ret); 2913 } 2914 2915 if (fscrypt_has_encryption_key(d_inode(parent))) { 2916 len = ceph_encode_encrypted_dname(d_inode(parent), 2917 buf, len); 2918 if (len < 0) { 2919 dput(parent); 2920 dput(cur); 2921 __putname(path); 2922 return ERR_PTR(len); 2923 } 2924 } 2925 pos -= len; 2926 if (pos < 0) { 2927 dput(parent); 2928 break; 2929 } 2930 memcpy(path + pos, buf, len); 2931 } 2932 dput(cur); 2933 cur = parent; 2934 2935 /* Are we at the root? */ 2936 if (IS_ROOT(cur)) 2937 break; 2938 2939 /* Are we out of buffer? */ 2940 if (--pos < 0) 2941 break; 2942 2943 path[pos] = '/'; 2944 } 2945 inode = d_inode(cur); 2946 base = inode ? ceph_ino(inode) : 0; 2947 dput(cur); 2948 2949 if (read_seqretry(&rename_lock, seq)) 2950 goto retry; 2951 2952 if (pos < 0) { 2953 /* 2954 * The path is longer than PATH_MAX and this function 2955 * cannot ever succeed. Creating paths that long is 2956 * possible with Ceph, but Linux cannot use them. 2957 */ 2958 __putname(path); 2959 return ERR_PTR(-ENAMETOOLONG); 2960 } 2961 2962 /* Initialize the output structure */ 2963 memset(path_info, 0, sizeof(*path_info)); 2964 2965 path_info->vino.ino = base; 2966 path_info->pathlen = PATH_MAX - 1 - pos; 2967 path_info->path = path + pos; 2968 path_info->freepath = true; 2969 2970 /* Set snap from dentry if available */ 2971 if (d_inode(dentry)) 2972 path_info->vino.snap = ceph_snap(d_inode(dentry)); 2973 else 2974 path_info->vino.snap = CEPH_NOSNAP; 2975 2976 doutc(cl, "on %p %d built %llx '%.*s'\n", dentry, d_count(dentry), 2977 base, PATH_MAX - 1 - pos, path + pos); 2978 return path + pos; 2979 } 2980 2981 static int build_dentry_path(struct ceph_mds_client *mdsc, struct dentry *dentry, 2982 struct inode *dir, struct ceph_path_info *path_info, 2983 bool parent_locked) 2984 { 2985 char *path; 2986 2987 rcu_read_lock(); 2988 if (!dir) 2989 dir = d_inode_rcu(dentry->d_parent); 2990 if (dir && parent_locked && ceph_snap(dir) == CEPH_NOSNAP && 2991 !IS_ENCRYPTED(dir)) { 2992 path_info->vino.ino = ceph_ino(dir); 2993 path_info->vino.snap = ceph_snap(dir); 2994 rcu_read_unlock(); 2995 path_info->path = dentry->d_name.name; 2996 path_info->pathlen = dentry->d_name.len; 2997 path_info->freepath = false; 2998 return 0; 2999 } 3000 rcu_read_unlock(); 3001 path = ceph_mdsc_build_path(mdsc, dentry, path_info, 1); 3002 if (IS_ERR(path)) 3003 return PTR_ERR(path); 3004 /* 3005 * ceph_mdsc_build_path already fills path_info, including snap handling. 3006 */ 3007 return 0; 3008 } 3009 3010 static int build_inode_path(struct inode *inode, struct ceph_path_info *path_info) 3011 { 3012 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb); 3013 struct dentry *dentry; 3014 char *path; 3015 3016 if (ceph_snap(inode) == CEPH_NOSNAP) { 3017 path_info->vino.ino = ceph_ino(inode); 3018 path_info->vino.snap = ceph_snap(inode); 3019 path_info->pathlen = 0; 3020 path_info->freepath = false; 3021 return 0; 3022 } 3023 dentry = d_find_alias(inode); 3024 path = ceph_mdsc_build_path(mdsc, dentry, path_info, 1); 3025 dput(dentry); 3026 if (IS_ERR(path)) 3027 return PTR_ERR(path); 3028 /* 3029 * ceph_mdsc_build_path already fills path_info, including snap from dentry. 3030 * Override with inode's snap since that's what this function is for. 3031 */ 3032 path_info->vino.snap = ceph_snap(inode); 3033 return 0; 3034 } 3035 3036 /* 3037 * request arguments may be specified via an inode *, a dentry *, or 3038 * an explicit ino+path. 3039 */ 3040 static int set_request_path_attr(struct ceph_mds_client *mdsc, struct inode *rinode, 3041 struct dentry *rdentry, struct inode *rdiri, 3042 const char *rpath, u64 rino, 3043 struct ceph_path_info *path_info, 3044 bool parent_locked) 3045 { 3046 struct ceph_client *cl = mdsc->fsc->client; 3047 int r = 0; 3048 3049 /* Initialize the output structure */ 3050 memset(path_info, 0, sizeof(*path_info)); 3051 3052 if (rinode) { 3053 r = build_inode_path(rinode, path_info); 3054 doutc(cl, " inode %p %llx.%llx\n", rinode, ceph_ino(rinode), 3055 ceph_snap(rinode)); 3056 } else if (rdentry) { 3057 r = build_dentry_path(mdsc, rdentry, rdiri, path_info, parent_locked); 3058 doutc(cl, " dentry %p %llx/%.*s\n", rdentry, path_info->vino.ino, 3059 path_info->pathlen, path_info->path); 3060 } else if (rpath || rino) { 3061 path_info->vino.ino = rino; 3062 path_info->vino.snap = CEPH_NOSNAP; 3063 path_info->path = rpath; 3064 path_info->pathlen = rpath ? strlen(rpath) : 0; 3065 path_info->freepath = false; 3066 3067 doutc(cl, " path %.*s\n", path_info->pathlen, rpath); 3068 } 3069 3070 return r; 3071 } 3072 3073 static void encode_mclientrequest_tail(void **p, 3074 const struct ceph_mds_request *req) 3075 { 3076 struct ceph_timespec ts; 3077 int i; 3078 3079 ceph_encode_timespec64(&ts, &req->r_stamp); 3080 ceph_encode_copy(p, &ts, sizeof(ts)); 3081 3082 /* v4: gid_list */ 3083 ceph_encode_32(p, req->r_cred->group_info->ngroups); 3084 for (i = 0; i < req->r_cred->group_info->ngroups; i++) 3085 ceph_encode_64(p, from_kgid(&init_user_ns, 3086 req->r_cred->group_info->gid[i])); 3087 3088 /* v5: altname */ 3089 ceph_encode_32(p, req->r_altname_len); 3090 ceph_encode_copy(p, req->r_altname, req->r_altname_len); 3091 3092 /* v6: fscrypt_auth and fscrypt_file */ 3093 if (req->r_fscrypt_auth) { 3094 u32 authlen = ceph_fscrypt_auth_len(req->r_fscrypt_auth); 3095 3096 ceph_encode_32(p, authlen); 3097 ceph_encode_copy(p, req->r_fscrypt_auth, authlen); 3098 } else { 3099 ceph_encode_32(p, 0); 3100 } 3101 if (test_bit(CEPH_MDS_R_FSCRYPT_FILE, &req->r_req_flags)) { 3102 ceph_encode_32(p, sizeof(__le64)); 3103 ceph_encode_64(p, req->r_fscrypt_file); 3104 } else { 3105 ceph_encode_32(p, 0); 3106 } 3107 } 3108 3109 static inline u16 mds_supported_head_version(struct ceph_mds_session *session) 3110 { 3111 if (!test_bit(CEPHFS_FEATURE_32BITS_RETRY_FWD, &session->s_features)) 3112 return 1; 3113 3114 if (!test_bit(CEPHFS_FEATURE_HAS_OWNER_UIDGID, &session->s_features)) 3115 return 2; 3116 3117 return CEPH_MDS_REQUEST_HEAD_VERSION; 3118 } 3119 3120 static struct ceph_mds_request_head_legacy * 3121 find_legacy_request_head(void *p, u64 features) 3122 { 3123 bool legacy = !(features & CEPH_FEATURE_FS_BTIME); 3124 struct ceph_mds_request_head *head; 3125 3126 if (legacy) 3127 return (struct ceph_mds_request_head_legacy *)p; 3128 head = (struct ceph_mds_request_head *)p; 3129 return (struct ceph_mds_request_head_legacy *)&head->oldest_client_tid; 3130 } 3131 3132 /* 3133 * called under mdsc->mutex 3134 */ 3135 static struct ceph_msg *create_request_message(struct ceph_mds_session *session, 3136 struct ceph_mds_request *req, 3137 bool drop_cap_releases) 3138 { 3139 int mds = session->s_mds; 3140 struct ceph_mds_client *mdsc = session->s_mdsc; 3141 struct ceph_client *cl = mdsc->fsc->client; 3142 struct ceph_msg *msg; 3143 struct ceph_mds_request_head_legacy *lhead; 3144 struct ceph_path_info path_info1 = {0}; 3145 struct ceph_path_info path_info2 = {0}; 3146 struct dentry *old_dentry = NULL; 3147 int len; 3148 u16 releases; 3149 void *p, *end; 3150 int ret; 3151 bool legacy = !(session->s_con.peer_features & CEPH_FEATURE_FS_BTIME); 3152 u16 request_head_version = mds_supported_head_version(session); 3153 kuid_t caller_fsuid = req->r_cred->fsuid; 3154 kgid_t caller_fsgid = req->r_cred->fsgid; 3155 bool parent_locked = test_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags); 3156 3157 ret = set_request_path_attr(mdsc, req->r_inode, req->r_dentry, 3158 req->r_parent, req->r_path1, req->r_ino1.ino, 3159 &path_info1, parent_locked); 3160 if (ret < 0) { 3161 msg = ERR_PTR(ret); 3162 goto out; 3163 } 3164 3165 /* 3166 * When the parent directory's i_rwsem is *not* locked, req->r_parent may 3167 * have become stale (e.g. after a concurrent rename) between the time the 3168 * dentry was looked up and now. If we detect that the stored r_parent 3169 * does not match the inode number we just encoded for the request, switch 3170 * to the correct inode so that the MDS receives a valid parent reference. 3171 */ 3172 if (!parent_locked && req->r_parent && path_info1.vino.ino && 3173 ceph_ino(req->r_parent) != path_info1.vino.ino) { 3174 struct inode *old_parent = req->r_parent; 3175 struct inode *correct_dir = ceph_get_inode(mdsc->fsc->sb, path_info1.vino, NULL); 3176 if (!IS_ERR(correct_dir)) { 3177 WARN_ONCE(1, "ceph: r_parent mismatch (had %llx wanted %llx) - updating\n", 3178 ceph_ino(old_parent), path_info1.vino.ino); 3179 /* 3180 * Transfer CEPH_CAP_PIN from the old parent to the new one. 3181 * The pin was taken earlier in ceph_mdsc_submit_request(). 3182 */ 3183 ceph_put_cap_refs(ceph_inode(old_parent), CEPH_CAP_PIN); 3184 iput(old_parent); 3185 req->r_parent = correct_dir; 3186 ceph_get_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN); 3187 } 3188 } 3189 3190 /* If r_old_dentry is set, then assume that its parent is locked */ 3191 if (req->r_old_dentry && 3192 !(req->r_old_dentry->d_flags & DCACHE_DISCONNECTED)) 3193 old_dentry = req->r_old_dentry; 3194 ret = set_request_path_attr(mdsc, NULL, old_dentry, 3195 req->r_old_dentry_dir, 3196 req->r_path2, req->r_ino2.ino, 3197 &path_info2, true); 3198 if (ret < 0) { 3199 msg = ERR_PTR(ret); 3200 goto out_free1; 3201 } 3202 3203 req->r_altname = get_fscrypt_altname(req, &req->r_altname_len); 3204 if (IS_ERR(req->r_altname)) { 3205 msg = ERR_CAST(req->r_altname); 3206 req->r_altname = NULL; 3207 goto out_free2; 3208 } 3209 3210 /* 3211 * For old cephs without supporting the 32bit retry/fwd feature 3212 * it will copy the raw memories directly when decoding the 3213 * requests. While new cephs will decode the head depending the 3214 * version member, so we need to make sure it will be compatible 3215 * with them both. 3216 */ 3217 if (legacy) 3218 len = sizeof(struct ceph_mds_request_head_legacy); 3219 else if (request_head_version == 1) 3220 len = offsetofend(struct ceph_mds_request_head, args); 3221 else if (request_head_version == 2) 3222 len = offsetofend(struct ceph_mds_request_head, ext_num_fwd); 3223 else 3224 len = sizeof(struct ceph_mds_request_head); 3225 3226 /* filepaths */ 3227 len += 2 * (1 + sizeof(u32) + sizeof(u64)); 3228 len += path_info1.pathlen + path_info2.pathlen; 3229 3230 /* cap releases */ 3231 len += sizeof(struct ceph_mds_request_release) * 3232 (!!req->r_inode_drop + !!req->r_dentry_drop + 3233 !!req->r_old_inode_drop + !!req->r_old_dentry_drop); 3234 3235 if (req->r_dentry_drop) 3236 len += path_info1.pathlen; 3237 if (req->r_old_dentry_drop) 3238 len += path_info2.pathlen; 3239 3240 /* MClientRequest tail */ 3241 3242 /* req->r_stamp */ 3243 len += sizeof(struct ceph_timespec); 3244 3245 /* gid list */ 3246 len += sizeof(u32) + (sizeof(u64) * req->r_cred->group_info->ngroups); 3247 3248 /* alternate name */ 3249 len += sizeof(u32) + req->r_altname_len; 3250 3251 /* fscrypt_auth */ 3252 len += sizeof(u32); // fscrypt_auth 3253 if (req->r_fscrypt_auth) 3254 len += ceph_fscrypt_auth_len(req->r_fscrypt_auth); 3255 3256 /* fscrypt_file */ 3257 len += sizeof(u32); 3258 if (test_bit(CEPH_MDS_R_FSCRYPT_FILE, &req->r_req_flags)) 3259 len += sizeof(__le64); 3260 3261 msg = ceph_msg_new2(CEPH_MSG_CLIENT_REQUEST, len, 1, GFP_NOFS, false); 3262 if (!msg) { 3263 msg = ERR_PTR(-ENOMEM); 3264 goto out_free2; 3265 } 3266 3267 msg->hdr.tid = cpu_to_le64(req->r_tid); 3268 3269 lhead = find_legacy_request_head(msg->front.iov_base, 3270 session->s_con.peer_features); 3271 3272 if ((req->r_mnt_idmap != &nop_mnt_idmap) && 3273 !test_bit(CEPHFS_FEATURE_HAS_OWNER_UIDGID, &session->s_features)) { 3274 WARN_ON_ONCE(!IS_CEPH_MDS_OP_NEWINODE(req->r_op)); 3275 3276 if (enable_unsafe_idmap) { 3277 pr_warn_once_client(cl, 3278 "idmapped mount is used and CEPHFS_FEATURE_HAS_OWNER_UIDGID" 3279 " is not supported by MDS. UID/GID-based restrictions may" 3280 " not work properly.\n"); 3281 3282 caller_fsuid = from_vfsuid(req->r_mnt_idmap, &init_user_ns, 3283 VFSUIDT_INIT(req->r_cred->fsuid)); 3284 caller_fsgid = from_vfsgid(req->r_mnt_idmap, &init_user_ns, 3285 VFSGIDT_INIT(req->r_cred->fsgid)); 3286 } else { 3287 pr_err_ratelimited_client(cl, 3288 "idmapped mount is used and CEPHFS_FEATURE_HAS_OWNER_UIDGID" 3289 " is not supported by MDS. Fail request with -EIO.\n"); 3290 3291 ret = -EIO; 3292 goto out_err; 3293 } 3294 } 3295 3296 /* 3297 * The ceph_mds_request_head_legacy didn't contain a version field, and 3298 * one was added when we moved the message version from 3->4. 3299 */ 3300 if (legacy) { 3301 msg->hdr.version = cpu_to_le16(3); 3302 p = msg->front.iov_base + sizeof(*lhead); 3303 } else if (request_head_version == 1) { 3304 struct ceph_mds_request_head *nhead = msg->front.iov_base; 3305 3306 msg->hdr.version = cpu_to_le16(4); 3307 nhead->version = cpu_to_le16(1); 3308 p = msg->front.iov_base + offsetofend(struct ceph_mds_request_head, args); 3309 } else if (request_head_version == 2) { 3310 struct ceph_mds_request_head *nhead = msg->front.iov_base; 3311 3312 msg->hdr.version = cpu_to_le16(6); 3313 nhead->version = cpu_to_le16(2); 3314 3315 p = msg->front.iov_base + offsetofend(struct ceph_mds_request_head, ext_num_fwd); 3316 } else { 3317 struct ceph_mds_request_head *nhead = msg->front.iov_base; 3318 kuid_t owner_fsuid; 3319 kgid_t owner_fsgid; 3320 3321 msg->hdr.version = cpu_to_le16(6); 3322 nhead->version = cpu_to_le16(CEPH_MDS_REQUEST_HEAD_VERSION); 3323 nhead->struct_len = cpu_to_le32(sizeof(struct ceph_mds_request_head)); 3324 3325 if (IS_CEPH_MDS_OP_NEWINODE(req->r_op)) { 3326 owner_fsuid = from_vfsuid(req->r_mnt_idmap, &init_user_ns, 3327 VFSUIDT_INIT(req->r_cred->fsuid)); 3328 owner_fsgid = from_vfsgid(req->r_mnt_idmap, &init_user_ns, 3329 VFSGIDT_INIT(req->r_cred->fsgid)); 3330 nhead->owner_uid = cpu_to_le32(from_kuid(&init_user_ns, owner_fsuid)); 3331 nhead->owner_gid = cpu_to_le32(from_kgid(&init_user_ns, owner_fsgid)); 3332 } else { 3333 nhead->owner_uid = cpu_to_le32(-1); 3334 nhead->owner_gid = cpu_to_le32(-1); 3335 } 3336 3337 p = msg->front.iov_base + sizeof(*nhead); 3338 } 3339 3340 end = msg->front.iov_base + msg->front.iov_len; 3341 3342 lhead->mdsmap_epoch = cpu_to_le32(mdsc->mdsmap->m_epoch); 3343 lhead->op = cpu_to_le32(req->r_op); 3344 lhead->caller_uid = cpu_to_le32(from_kuid(&init_user_ns, 3345 caller_fsuid)); 3346 lhead->caller_gid = cpu_to_le32(from_kgid(&init_user_ns, 3347 caller_fsgid)); 3348 lhead->ino = cpu_to_le64(req->r_deleg_ino); 3349 lhead->args = req->r_args; 3350 3351 ceph_encode_filepath(&p, end, path_info1.vino.ino, path_info1.path); 3352 ceph_encode_filepath(&p, end, path_info2.vino.ino, path_info2.path); 3353 3354 /* make note of release offset, in case we need to replay */ 3355 req->r_request_release_offset = p - msg->front.iov_base; 3356 3357 /* cap releases */ 3358 releases = 0; 3359 if (req->r_inode_drop) 3360 releases += ceph_encode_inode_release(&p, 3361 req->r_inode ? req->r_inode : d_inode(req->r_dentry), 3362 mds, req->r_inode_drop, req->r_inode_unless, 3363 req->r_op == CEPH_MDS_OP_READDIR); 3364 if (req->r_dentry_drop) { 3365 ret = ceph_encode_dentry_release(&p, req->r_dentry, 3366 req->r_parent, mds, req->r_dentry_drop, 3367 req->r_dentry_unless); 3368 if (ret < 0) 3369 goto out_err; 3370 releases += ret; 3371 } 3372 if (req->r_old_dentry_drop) { 3373 ret = ceph_encode_dentry_release(&p, req->r_old_dentry, 3374 req->r_old_dentry_dir, mds, 3375 req->r_old_dentry_drop, 3376 req->r_old_dentry_unless); 3377 if (ret < 0) 3378 goto out_err; 3379 releases += ret; 3380 } 3381 if (req->r_old_inode_drop) 3382 releases += ceph_encode_inode_release(&p, 3383 d_inode(req->r_old_dentry), 3384 mds, req->r_old_inode_drop, req->r_old_inode_unless, 0); 3385 3386 if (drop_cap_releases) { 3387 releases = 0; 3388 p = msg->front.iov_base + req->r_request_release_offset; 3389 } 3390 3391 lhead->num_releases = cpu_to_le16(releases); 3392 3393 encode_mclientrequest_tail(&p, req); 3394 3395 if (WARN_ON_ONCE(p > end)) { 3396 ceph_msg_put(msg); 3397 msg = ERR_PTR(-ERANGE); 3398 goto out_free2; 3399 } 3400 3401 msg->front.iov_len = p - msg->front.iov_base; 3402 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len); 3403 3404 if (req->r_pagelist) { 3405 struct ceph_pagelist *pagelist = req->r_pagelist; 3406 ceph_msg_data_add_pagelist(msg, pagelist); 3407 msg->hdr.data_len = cpu_to_le32(pagelist->length); 3408 } else { 3409 msg->hdr.data_len = 0; 3410 } 3411 3412 msg->hdr.data_off = cpu_to_le16(0); 3413 3414 out_free2: 3415 ceph_mdsc_free_path_info(&path_info2); 3416 out_free1: 3417 ceph_mdsc_free_path_info(&path_info1); 3418 out: 3419 return msg; 3420 out_err: 3421 ceph_msg_put(msg); 3422 msg = ERR_PTR(ret); 3423 goto out_free2; 3424 } 3425 3426 /* 3427 * called under mdsc->mutex if error, under no mutex if 3428 * success. 3429 */ 3430 static void complete_request(struct ceph_mds_client *mdsc, 3431 struct ceph_mds_request *req) 3432 { 3433 req->r_end_latency = ktime_get(); 3434 3435 trace_ceph_mdsc_complete_request(mdsc, req); 3436 3437 if (req->r_callback) 3438 req->r_callback(mdsc, req); 3439 complete_all(&req->r_completion); 3440 } 3441 3442 /* 3443 * called under mdsc->mutex 3444 */ 3445 static int __prepare_send_request(struct ceph_mds_session *session, 3446 struct ceph_mds_request *req, 3447 bool drop_cap_releases) 3448 { 3449 int mds = session->s_mds; 3450 struct ceph_mds_client *mdsc = session->s_mdsc; 3451 struct ceph_client *cl = mdsc->fsc->client; 3452 struct ceph_mds_request_head_legacy *lhead; 3453 struct ceph_mds_request_head *nhead; 3454 struct ceph_msg *msg; 3455 int flags = 0, old_max_retry; 3456 bool old_version = !test_bit(CEPHFS_FEATURE_32BITS_RETRY_FWD, 3457 &session->s_features); 3458 3459 /* 3460 * Avoid infinite retrying after overflow. The client will 3461 * increase the retry count and if the MDS is old version, 3462 * so we limit to retry at most 256 times. 3463 */ 3464 if (req->r_attempts) { 3465 old_max_retry = sizeof_field(struct ceph_mds_request_head, 3466 num_retry); 3467 old_max_retry = 1 << (old_max_retry * BITS_PER_BYTE); 3468 if ((old_version && req->r_attempts >= old_max_retry) || 3469 ((uint32_t)req->r_attempts >= U32_MAX)) { 3470 pr_warn_ratelimited_client(cl, "request tid %llu seq overflow\n", 3471 req->r_tid); 3472 return -EMULTIHOP; 3473 } 3474 } 3475 3476 req->r_attempts++; 3477 if (req->r_inode) { 3478 struct ceph_cap *cap = 3479 ceph_get_cap_for_mds(ceph_inode(req->r_inode), mds); 3480 3481 if (cap) 3482 req->r_sent_on_mseq = cap->mseq; 3483 else 3484 req->r_sent_on_mseq = -1; 3485 } 3486 doutc(cl, "%p tid %lld %s (attempt %d)\n", req, req->r_tid, 3487 ceph_mds_op_name(req->r_op), req->r_attempts); 3488 3489 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) { 3490 void *p; 3491 3492 /* 3493 * Replay. Do not regenerate message (and rebuild 3494 * paths, etc.); just use the original message. 3495 * Rebuilding paths will break for renames because 3496 * d_move mangles the src name. 3497 */ 3498 msg = req->r_request; 3499 lhead = find_legacy_request_head(msg->front.iov_base, 3500 session->s_con.peer_features); 3501 3502 flags = le32_to_cpu(lhead->flags); 3503 flags |= CEPH_MDS_FLAG_REPLAY; 3504 lhead->flags = cpu_to_le32(flags); 3505 3506 if (req->r_target_inode) 3507 lhead->ino = cpu_to_le64(ceph_ino(req->r_target_inode)); 3508 3509 lhead->num_retry = req->r_attempts - 1; 3510 if (!old_version) { 3511 nhead = (struct ceph_mds_request_head*)msg->front.iov_base; 3512 nhead->ext_num_retry = cpu_to_le32(req->r_attempts - 1); 3513 } 3514 3515 /* remove cap/dentry releases from message */ 3516 lhead->num_releases = 0; 3517 3518 p = msg->front.iov_base + req->r_request_release_offset; 3519 encode_mclientrequest_tail(&p, req); 3520 3521 msg->front.iov_len = p - msg->front.iov_base; 3522 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len); 3523 return 0; 3524 } 3525 3526 if (req->r_request) { 3527 ceph_msg_put(req->r_request); 3528 req->r_request = NULL; 3529 } 3530 msg = create_request_message(session, req, drop_cap_releases); 3531 if (IS_ERR(msg)) { 3532 req->r_err = PTR_ERR(msg); 3533 return PTR_ERR(msg); 3534 } 3535 req->r_request = msg; 3536 3537 lhead = find_legacy_request_head(msg->front.iov_base, 3538 session->s_con.peer_features); 3539 lhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc)); 3540 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) 3541 flags |= CEPH_MDS_FLAG_REPLAY; 3542 if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) 3543 flags |= CEPH_MDS_FLAG_ASYNC; 3544 if (req->r_parent) 3545 flags |= CEPH_MDS_FLAG_WANT_DENTRY; 3546 lhead->flags = cpu_to_le32(flags); 3547 lhead->num_fwd = req->r_num_fwd; 3548 lhead->num_retry = req->r_attempts - 1; 3549 if (!old_version) { 3550 nhead = (struct ceph_mds_request_head*)msg->front.iov_base; 3551 nhead->ext_num_fwd = cpu_to_le32(req->r_num_fwd); 3552 nhead->ext_num_retry = cpu_to_le32(req->r_attempts - 1); 3553 } 3554 3555 doutc(cl, " r_parent = %p\n", req->r_parent); 3556 return 0; 3557 } 3558 3559 /* 3560 * called under mdsc->mutex 3561 */ 3562 static int __send_request(struct ceph_mds_session *session, 3563 struct ceph_mds_request *req, 3564 bool drop_cap_releases) 3565 { 3566 int err; 3567 3568 trace_ceph_mdsc_send_request(session, req); 3569 3570 err = __prepare_send_request(session, req, drop_cap_releases); 3571 if (!err) { 3572 ceph_msg_get(req->r_request); 3573 ceph_con_send(&session->s_con, req->r_request); 3574 } 3575 3576 return err; 3577 } 3578 3579 /* 3580 * send request, or put it on the appropriate wait list. 3581 */ 3582 static void __do_request(struct ceph_mds_client *mdsc, 3583 struct ceph_mds_request *req) 3584 { 3585 struct ceph_client *cl = mdsc->fsc->client; 3586 struct ceph_mds_session *session = NULL; 3587 int mds = -1; 3588 int err = 0; 3589 bool random; 3590 3591 if (req->r_err || test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) { 3592 if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) 3593 __unregister_request(mdsc, req); 3594 return; 3595 } 3596 3597 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO) { 3598 doutc(cl, "metadata corrupted\n"); 3599 err = -EIO; 3600 goto finish; 3601 } 3602 if (req->r_timeout && 3603 time_after_eq(jiffies, req->r_started + req->r_timeout)) { 3604 doutc(cl, "timed out\n"); 3605 err = -ETIMEDOUT; 3606 goto finish; 3607 } 3608 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) { 3609 doutc(cl, "forced umount\n"); 3610 err = -EIO; 3611 goto finish; 3612 } 3613 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_MOUNTING) { 3614 if (mdsc->mdsmap_err) { 3615 err = mdsc->mdsmap_err; 3616 doutc(cl, "mdsmap err %d\n", err); 3617 goto finish; 3618 } 3619 if (mdsc->mdsmap->m_epoch == 0) { 3620 doutc(cl, "no mdsmap, waiting for map\n"); 3621 trace_ceph_mdsc_suspend_request(mdsc, session, req, 3622 ceph_mdsc_suspend_reason_no_mdsmap); 3623 list_add(&req->r_wait, &mdsc->waiting_for_map); 3624 return; 3625 } 3626 if (!(mdsc->fsc->mount_options->flags & 3627 CEPH_MOUNT_OPT_MOUNTWAIT) && 3628 !ceph_mdsmap_is_cluster_available(mdsc->mdsmap)) { 3629 err = -EHOSTUNREACH; 3630 goto finish; 3631 } 3632 } 3633 3634 put_request_session(req); 3635 3636 mds = __choose_mds(mdsc, req, &random); 3637 if (mds < 0 || 3638 ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) { 3639 if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) { 3640 err = -EJUKEBOX; 3641 goto finish; 3642 } 3643 doutc(cl, "no mds or not active, waiting for map\n"); 3644 trace_ceph_mdsc_suspend_request(mdsc, session, req, 3645 ceph_mdsc_suspend_reason_no_active_mds); 3646 list_add(&req->r_wait, &mdsc->waiting_for_map); 3647 return; 3648 } 3649 3650 /* get, open session */ 3651 session = __ceph_lookup_mds_session(mdsc, mds); 3652 if (!session) { 3653 session = register_session(mdsc, mds); 3654 if (IS_ERR(session)) { 3655 err = PTR_ERR(session); 3656 goto finish; 3657 } 3658 } 3659 req->r_session = ceph_get_mds_session(session); 3660 3661 doutc(cl, "mds%d session %p state %s\n", mds, session, 3662 ceph_session_state_name(session->s_state)); 3663 3664 /* 3665 * The old ceph will crash the MDSs when see unknown OPs 3666 */ 3667 if (req->r_feature_needed > 0 && 3668 !test_bit(req->r_feature_needed, &session->s_features)) { 3669 err = -EOPNOTSUPP; 3670 goto out_session; 3671 } 3672 3673 if (session->s_state != CEPH_MDS_SESSION_OPEN && 3674 session->s_state != CEPH_MDS_SESSION_HUNG) { 3675 /* 3676 * We cannot queue async requests since the caps and delegated 3677 * inodes are bound to the session. Just return -EJUKEBOX and 3678 * let the caller retry a sync request in that case. 3679 */ 3680 if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) { 3681 err = -EJUKEBOX; 3682 goto out_session; 3683 } 3684 3685 /* 3686 * If the session has been REJECTED, then return a hard error, 3687 * unless it's a CLEANRECOVER mount, in which case we'll queue 3688 * it to the mdsc queue. 3689 */ 3690 if (session->s_state == CEPH_MDS_SESSION_REJECTED) { 3691 if (ceph_test_mount_opt(mdsc->fsc, CLEANRECOVER)) { 3692 trace_ceph_mdsc_suspend_request(mdsc, session, req, 3693 ceph_mdsc_suspend_reason_rejected); 3694 list_add(&req->r_wait, &mdsc->waiting_for_map); 3695 } else 3696 err = -EACCES; 3697 goto out_session; 3698 } 3699 3700 if (session->s_state == CEPH_MDS_SESSION_NEW || 3701 session->s_state == CEPH_MDS_SESSION_CLOSING) { 3702 err = __open_session(mdsc, session); 3703 if (err) 3704 goto out_session; 3705 /* retry the same mds later */ 3706 if (random) 3707 req->r_resend_mds = mds; 3708 } 3709 trace_ceph_mdsc_suspend_request(mdsc, session, req, 3710 ceph_mdsc_suspend_reason_session); 3711 list_add(&req->r_wait, &session->s_waiting); 3712 goto out_session; 3713 } 3714 3715 /* send request */ 3716 req->r_resend_mds = -1; /* forget any previous mds hint */ 3717 3718 if (req->r_request_started == 0) /* note request start time */ 3719 req->r_request_started = jiffies; 3720 3721 /* 3722 * For async create we will choose the auth MDS of frag in parent 3723 * directory to send the request and usually this works fine, but 3724 * if the migrated the dirtory to another MDS before it could handle 3725 * it the request will be forwarded. 3726 * 3727 * And then the auth cap will be changed. 3728 */ 3729 if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags) && req->r_num_fwd) { 3730 struct ceph_dentry_info *di = ceph_dentry(req->r_dentry); 3731 struct ceph_inode_info *ci; 3732 struct ceph_cap *cap; 3733 3734 /* 3735 * The request maybe handled very fast and the new inode 3736 * hasn't been linked to the dentry yet. We need to wait 3737 * for the ceph_finish_async_create(), which shouldn't be 3738 * stuck too long or fail in thoery, to finish when forwarding 3739 * the request. 3740 */ 3741 if (!d_inode(req->r_dentry)) { 3742 err = wait_on_bit(&di->flags, CEPH_DENTRY_ASYNC_CREATE_BIT, 3743 TASK_KILLABLE); 3744 if (err) { 3745 mutex_lock(&req->r_fill_mutex); 3746 set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags); 3747 mutex_unlock(&req->r_fill_mutex); 3748 goto out_session; 3749 } 3750 } 3751 3752 ci = ceph_inode(d_inode(req->r_dentry)); 3753 3754 spin_lock(&ci->i_ceph_lock); 3755 cap = ci->i_auth_cap; 3756 if (test_bit(CEPH_I_ASYNC_CREATE_BIT, &ci->i_ceph_flags) && 3757 mds != cap->mds) { 3758 doutc(cl, "session changed for auth cap %d -> %d\n", 3759 cap->session->s_mds, session->s_mds); 3760 3761 /* Remove the auth cap from old session */ 3762 spin_lock(&cap->session->s_cap_lock); 3763 cap->session->s_nr_caps--; 3764 list_del_init(&cap->session_caps); 3765 spin_unlock(&cap->session->s_cap_lock); 3766 3767 /* Add the auth cap to the new session */ 3768 cap->mds = mds; 3769 cap->session = session; 3770 spin_lock(&session->s_cap_lock); 3771 session->s_nr_caps++; 3772 list_add_tail(&cap->session_caps, &session->s_caps); 3773 spin_unlock(&session->s_cap_lock); 3774 3775 change_auth_cap_ses(ci, session); 3776 } 3777 spin_unlock(&ci->i_ceph_lock); 3778 } 3779 3780 err = __send_request(session, req, false); 3781 3782 out_session: 3783 ceph_put_mds_session(session); 3784 finish: 3785 if (err) { 3786 doutc(cl, "early error %d\n", err); 3787 req->r_err = err; 3788 complete_request(mdsc, req); 3789 __unregister_request(mdsc, req); 3790 } 3791 return; 3792 } 3793 3794 /* 3795 * called under mdsc->mutex 3796 */ 3797 static void __wake_requests(struct ceph_mds_client *mdsc, 3798 struct list_head *head) 3799 { 3800 struct ceph_client *cl = mdsc->fsc->client; 3801 struct ceph_mds_request *req; 3802 LIST_HEAD(tmp_list); 3803 3804 list_splice_init(head, &tmp_list); 3805 3806 while (!list_empty(&tmp_list)) { 3807 req = list_entry(tmp_list.next, 3808 struct ceph_mds_request, r_wait); 3809 list_del_init(&req->r_wait); 3810 doutc(cl, " wake request %p tid %llu\n", req, 3811 req->r_tid); 3812 trace_ceph_mdsc_resume_request(mdsc, req); 3813 __do_request(mdsc, req); 3814 } 3815 } 3816 3817 /* 3818 * Wake up threads with requests pending for @mds, so that they can 3819 * resubmit their requests to a possibly different mds. 3820 */ 3821 static void kick_requests(struct ceph_mds_client *mdsc, int mds) 3822 { 3823 struct ceph_client *cl = mdsc->fsc->client; 3824 struct ceph_mds_request *req; 3825 struct rb_node *p = rb_first(&mdsc->request_tree); 3826 3827 doutc(cl, "kick_requests mds%d\n", mds); 3828 while (p) { 3829 req = rb_entry(p, struct ceph_mds_request, r_node); 3830 p = rb_next(p); 3831 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) 3832 continue; 3833 if (req->r_attempts > 0) 3834 continue; /* only new requests */ 3835 if (req->r_session && 3836 req->r_session->s_mds == mds) { 3837 doutc(cl, " kicking tid %llu\n", req->r_tid); 3838 list_del_init(&req->r_wait); 3839 trace_ceph_mdsc_resume_request(mdsc, req); 3840 __do_request(mdsc, req); 3841 } 3842 } 3843 } 3844 3845 int ceph_mdsc_submit_request(struct ceph_mds_client *mdsc, struct inode *dir, 3846 struct ceph_mds_request *req) 3847 { 3848 struct ceph_client *cl = mdsc->fsc->client; 3849 int err = 0; 3850 3851 /* 3852 * If a reset is in progress, wait for it to complete. 3853 * 3854 * This is best-effort: a request can pass this check just 3855 * before the phase leaves IDLE and proceed concurrently with 3856 * reset. That is acceptable because (a) such requests will 3857 * either complete normally or fail and be retried by the 3858 * caller, and (b) adding lock serialization here would 3859 * penalize every request for a rare manual operation. 3860 */ 3861 err = ceph_mdsc_wait_for_reset(mdsc); 3862 if (err) { 3863 doutc(cl, "wait_for_reset failed: %d\n", err); 3864 return err; 3865 } 3866 3867 /* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */ 3868 if (req->r_inode) 3869 ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN); 3870 if (req->r_parent) { 3871 struct ceph_inode_info *ci = ceph_inode(req->r_parent); 3872 int fmode = (req->r_op & CEPH_MDS_OP_WRITE) ? 3873 CEPH_FILE_MODE_WR : CEPH_FILE_MODE_RD; 3874 spin_lock(&ci->i_ceph_lock); 3875 ceph_take_cap_refs(ci, CEPH_CAP_PIN, false); 3876 __ceph_touch_fmode(ci, mdsc, fmode); 3877 spin_unlock(&ci->i_ceph_lock); 3878 } 3879 if (req->r_old_dentry_dir) 3880 ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir), 3881 CEPH_CAP_PIN); 3882 3883 if (req->r_inode) { 3884 err = ceph_wait_on_async_create(req->r_inode); 3885 if (err) { 3886 doutc(cl, "wait for async create returned: %d\n", err); 3887 return err; 3888 } 3889 } 3890 3891 if (!err && req->r_old_inode) { 3892 err = ceph_wait_on_async_create(req->r_old_inode); 3893 if (err) { 3894 doutc(cl, "wait for async create returned: %d\n", err); 3895 return err; 3896 } 3897 } 3898 3899 doutc(cl, "submit_request on %p for inode %p\n", req, dir); 3900 mutex_lock(&mdsc->mutex); 3901 __register_request(mdsc, req, dir); 3902 trace_ceph_mdsc_submit_request(mdsc, req); 3903 __do_request(mdsc, req); 3904 err = req->r_err; 3905 mutex_unlock(&mdsc->mutex); 3906 return err; 3907 } 3908 3909 int ceph_mdsc_wait_request(struct ceph_mds_client *mdsc, 3910 struct ceph_mds_request *req, 3911 ceph_mds_request_wait_callback_t wait_func) 3912 { 3913 struct ceph_client *cl = mdsc->fsc->client; 3914 int err; 3915 3916 /* wait */ 3917 doutc(cl, "do_request waiting\n"); 3918 if (wait_func) { 3919 err = wait_func(mdsc, req); 3920 } else { 3921 long timeleft = wait_for_completion_killable_timeout( 3922 &req->r_completion, 3923 ceph_timeout_jiffies(req->r_timeout)); 3924 if (timeleft > 0) 3925 err = 0; 3926 else if (!timeleft) 3927 err = -ETIMEDOUT; /* timed out */ 3928 else 3929 err = timeleft; /* killed */ 3930 } 3931 doutc(cl, "do_request waited, got %d\n", err); 3932 mutex_lock(&mdsc->mutex); 3933 3934 /* only abort if we didn't race with a real reply */ 3935 if (test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) { 3936 err = le32_to_cpu(req->r_reply_info.head->result); 3937 } else if (err < 0) { 3938 doutc(cl, "aborted request %lld with %d\n", req->r_tid, err); 3939 3940 /* 3941 * ensure we aren't running concurrently with 3942 * ceph_fill_trace or ceph_readdir_prepopulate, which 3943 * rely on locks (dir mutex) held by our caller. 3944 */ 3945 mutex_lock(&req->r_fill_mutex); 3946 req->r_err = err; 3947 set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags); 3948 mutex_unlock(&req->r_fill_mutex); 3949 3950 if (req->r_parent && 3951 (req->r_op & CEPH_MDS_OP_WRITE)) 3952 ceph_invalidate_dir_request(req); 3953 } else { 3954 err = req->r_err; 3955 } 3956 3957 mutex_unlock(&mdsc->mutex); 3958 return err; 3959 } 3960 3961 /* 3962 * Synchrously perform an mds request. Take care of all of the 3963 * session setup, forwarding, retry details. 3964 */ 3965 int ceph_mdsc_do_request(struct ceph_mds_client *mdsc, 3966 struct inode *dir, 3967 struct ceph_mds_request *req) 3968 { 3969 struct ceph_client *cl = mdsc->fsc->client; 3970 int err; 3971 3972 doutc(cl, "do_request on %p\n", req); 3973 3974 /* issue */ 3975 err = ceph_mdsc_submit_request(mdsc, dir, req); 3976 if (!err) 3977 err = ceph_mdsc_wait_request(mdsc, req, NULL); 3978 doutc(cl, "do_request %p done, result %d\n", req, err); 3979 return err; 3980 } 3981 3982 /* 3983 * Invalidate dir's completeness, dentry lease state on an aborted MDS 3984 * namespace request. 3985 */ 3986 void ceph_invalidate_dir_request(struct ceph_mds_request *req) 3987 { 3988 struct inode *dir = req->r_parent; 3989 struct inode *old_dir = req->r_old_dentry_dir; 3990 struct ceph_client *cl = req->r_mdsc->fsc->client; 3991 3992 doutc(cl, "invalidate_dir_request %p %p (complete, lease(s))\n", 3993 dir, old_dir); 3994 3995 ceph_dir_clear_complete(dir); 3996 if (old_dir) 3997 ceph_dir_clear_complete(old_dir); 3998 if (req->r_dentry) 3999 ceph_invalidate_dentry_lease(req->r_dentry); 4000 if (req->r_old_dentry) 4001 ceph_invalidate_dentry_lease(req->r_old_dentry); 4002 } 4003 4004 /* 4005 * Handle mds reply. 4006 * 4007 * We take the session mutex and parse and process the reply immediately. 4008 * This preserves the logical ordering of replies, capabilities, etc., sent 4009 * by the MDS as they are applied to our local cache. 4010 */ 4011 static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg) 4012 { 4013 struct ceph_mds_client *mdsc = session->s_mdsc; 4014 struct ceph_client *cl = mdsc->fsc->client; 4015 struct ceph_mds_request *req; 4016 struct ceph_mds_reply_head *head = msg->front.iov_base; 4017 struct ceph_mds_reply_info_parsed *rinfo; /* parsed reply info */ 4018 struct ceph_snap_realm *realm; 4019 unsigned int nofs_flags; 4020 u64 tid; 4021 int err, result; 4022 int mds = session->s_mds; 4023 bool close_sessions = false; 4024 4025 if (msg->front.iov_len < sizeof(*head)) { 4026 pr_err_client(cl, "got corrupt (short) reply\n"); 4027 ceph_msg_dump(msg); 4028 return; 4029 } 4030 4031 /* get request, session */ 4032 tid = le64_to_cpu(msg->hdr.tid); 4033 mutex_lock(&mdsc->mutex); 4034 req = lookup_get_request(mdsc, tid); 4035 if (!req) { 4036 doutc(cl, "on unknown tid %llu\n", tid); 4037 mutex_unlock(&mdsc->mutex); 4038 return; 4039 } 4040 doutc(cl, "handle_reply %p\n", req); 4041 4042 /* correct session? */ 4043 if (req->r_session != session) { 4044 pr_err_client(cl, "got %llu on session mds%d not mds%d\n", 4045 tid, session->s_mds, 4046 req->r_session ? req->r_session->s_mds : -1); 4047 mutex_unlock(&mdsc->mutex); 4048 goto out; 4049 } 4050 4051 /* dup? */ 4052 if ((test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags) && !head->safe) || 4053 (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags) && head->safe)) { 4054 pr_warn_client(cl, "got a dup %s reply on %llu from mds%d\n", 4055 head->safe ? "safe" : "unsafe", tid, mds); 4056 mutex_unlock(&mdsc->mutex); 4057 goto out; 4058 } 4059 if (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags)) { 4060 pr_warn_client(cl, "got unsafe after safe on %llu from mds%d\n", 4061 tid, mds); 4062 mutex_unlock(&mdsc->mutex); 4063 goto out; 4064 } 4065 4066 result = le32_to_cpu(head->result); 4067 4068 if (head->safe) { 4069 set_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags); 4070 __unregister_request(mdsc, req); 4071 4072 /* last request during umount? */ 4073 if (mdsc->stopping && !__get_oldest_req(mdsc)) 4074 complete_all(&mdsc->safe_umount_waiters); 4075 4076 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) { 4077 /* 4078 * We already handled the unsafe response, now do the 4079 * cleanup. No need to examine the response; the MDS 4080 * doesn't include any result info in the safe 4081 * response. And even if it did, there is nothing 4082 * useful we could do with a revised return value. 4083 */ 4084 doutc(cl, "got safe reply %llu, mds%d\n", tid, mds); 4085 4086 mutex_unlock(&mdsc->mutex); 4087 goto out; 4088 } 4089 } else { 4090 set_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags); 4091 list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe); 4092 } 4093 4094 doutc(cl, "tid %lld result %d\n", tid, result); 4095 if (test_bit(CEPHFS_FEATURE_REPLY_ENCODING, &session->s_features)) 4096 err = parse_reply_info(session, msg, req, (u64)-1); 4097 else 4098 err = parse_reply_info(session, msg, req, 4099 session->s_con.peer_features); 4100 mutex_unlock(&mdsc->mutex); 4101 4102 /* Must find target inode outside of mutexes to avoid deadlocks */ 4103 rinfo = &req->r_reply_info; 4104 if ((err >= 0) && rinfo->head->is_target) { 4105 struct inode *in = xchg(&req->r_new_inode, NULL); 4106 struct ceph_vino tvino = { 4107 .ino = le64_to_cpu(rinfo->targeti.in->ino), 4108 .snap = le64_to_cpu(rinfo->targeti.in->snapid) 4109 }; 4110 4111 /* 4112 * If we ended up opening an existing inode, discard 4113 * r_new_inode 4114 */ 4115 if (req->r_op == CEPH_MDS_OP_CREATE && 4116 !req->r_reply_info.has_create_ino) { 4117 /* This should never happen on an async create */ 4118 WARN_ON_ONCE(req->r_deleg_ino); 4119 iput(in); 4120 in = NULL; 4121 } 4122 4123 in = ceph_get_inode(mdsc->fsc->sb, tvino, in); 4124 if (IS_ERR(in)) { 4125 err = PTR_ERR(in); 4126 mutex_lock(&session->s_mutex); 4127 goto out_err; 4128 } 4129 req->r_target_inode = in; 4130 ceph_inode_set_subvolume(in, rinfo->targeti.subvolume_id); 4131 } 4132 4133 mutex_lock(&session->s_mutex); 4134 if (err < 0) { 4135 pr_err_client(cl, "got corrupt reply mds%d(tid:%lld)\n", 4136 mds, tid); 4137 ceph_msg_dump(msg); 4138 goto out_err; 4139 } 4140 4141 /* snap trace */ 4142 realm = NULL; 4143 if (rinfo->snapblob_len) { 4144 down_write(&mdsc->snap_rwsem); 4145 err = ceph_update_snap_trace(mdsc, rinfo->snapblob, 4146 rinfo->snapblob + rinfo->snapblob_len, 4147 le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP, 4148 &realm); 4149 if (err) { 4150 up_write(&mdsc->snap_rwsem); 4151 close_sessions = true; 4152 if (err == -EIO) 4153 ceph_msg_dump(msg); 4154 goto out_err; 4155 } 4156 downgrade_write(&mdsc->snap_rwsem); 4157 } else { 4158 down_read(&mdsc->snap_rwsem); 4159 } 4160 4161 /* insert trace into our cache */ 4162 mutex_lock(&req->r_fill_mutex); 4163 4164 /* disable fs reclaim while we are using current->journal_info 4165 * for our own purposes, or else shrinkers of other 4166 * filesystems might dereference this pointer as a different 4167 * type 4168 */ 4169 nofs_flags = memalloc_nofs_save(); 4170 4171 current->journal_info = req; 4172 err = ceph_fill_trace(mdsc->fsc->sb, req); 4173 if (err == 0) { 4174 if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR || 4175 req->r_op == CEPH_MDS_OP_LSSNAP)) 4176 err = ceph_readdir_prepopulate(req, req->r_session); 4177 } 4178 current->journal_info = NULL; 4179 memalloc_nofs_restore(nofs_flags); 4180 mutex_unlock(&req->r_fill_mutex); 4181 4182 up_read(&mdsc->snap_rwsem); 4183 if (realm) 4184 ceph_put_snap_realm(mdsc, realm); 4185 4186 if (err == 0) { 4187 if (req->r_target_inode && 4188 test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) { 4189 struct ceph_inode_info *ci = 4190 ceph_inode(req->r_target_inode); 4191 spin_lock(&ci->i_unsafe_lock); 4192 list_add_tail(&req->r_unsafe_target_item, 4193 &ci->i_unsafe_iops); 4194 spin_unlock(&ci->i_unsafe_lock); 4195 } 4196 4197 ceph_unreserve_caps(mdsc, &req->r_caps_reservation); 4198 } 4199 out_err: 4200 mutex_lock(&mdsc->mutex); 4201 if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) { 4202 if (err) { 4203 req->r_err = err; 4204 } else { 4205 req->r_reply = ceph_msg_get(msg); 4206 set_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags); 4207 } 4208 } else { 4209 doutc(cl, "reply arrived after request %lld was aborted\n", tid); 4210 } 4211 mutex_unlock(&mdsc->mutex); 4212 4213 mutex_unlock(&session->s_mutex); 4214 4215 /* kick calling process */ 4216 complete_request(mdsc, req); 4217 4218 ceph_update_metadata_metrics(&mdsc->metric, req->r_start_latency, 4219 req->r_end_latency, err); 4220 out: 4221 ceph_mdsc_put_request(req); 4222 4223 /* Defer closing the sessions after s_mutex lock being released */ 4224 if (close_sessions) 4225 ceph_mdsc_close_sessions(mdsc); 4226 return; 4227 } 4228 4229 4230 4231 /* 4232 * handle mds notification that our request has been forwarded. 4233 */ 4234 static void handle_forward(struct ceph_mds_client *mdsc, 4235 struct ceph_mds_session *session, 4236 struct ceph_msg *msg) 4237 { 4238 struct ceph_client *cl = mdsc->fsc->client; 4239 struct ceph_mds_request *req; 4240 u64 tid = le64_to_cpu(msg->hdr.tid); 4241 u32 next_mds; 4242 u32 fwd_seq; 4243 int err = -EINVAL; 4244 void *p = msg->front.iov_base; 4245 void *end = p + msg->front.iov_len; 4246 bool aborted = false; 4247 4248 ceph_decode_need(&p, end, 2*sizeof(u32), bad); 4249 next_mds = ceph_decode_32(&p); 4250 fwd_seq = ceph_decode_32(&p); 4251 4252 mutex_lock(&mdsc->mutex); 4253 req = lookup_get_request(mdsc, tid); 4254 if (!req) { 4255 mutex_unlock(&mdsc->mutex); 4256 doutc(cl, "forward tid %llu to mds%d - req dne\n", tid, next_mds); 4257 return; /* dup reply? */ 4258 } 4259 4260 if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) { 4261 doutc(cl, "forward tid %llu aborted, unregistering\n", tid); 4262 __unregister_request(mdsc, req); 4263 } else if (fwd_seq <= req->r_num_fwd || (uint32_t)fwd_seq >= U32_MAX) { 4264 /* 4265 * Avoid infinite retrying after overflow. 4266 * 4267 * The MDS will increase the fwd count and in client side 4268 * if the num_fwd is less than the one saved in request 4269 * that means the MDS is an old version and overflowed of 4270 * 8 bits. 4271 */ 4272 mutex_lock(&req->r_fill_mutex); 4273 req->r_err = -EMULTIHOP; 4274 set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags); 4275 mutex_unlock(&req->r_fill_mutex); 4276 aborted = true; 4277 pr_warn_ratelimited_client(cl, "forward tid %llu seq overflow\n", 4278 tid); 4279 } else { 4280 /* resend. forward race not possible; mds would drop */ 4281 doutc(cl, "forward tid %llu to mds%d (we resend)\n", tid, next_mds); 4282 BUG_ON(req->r_err); 4283 BUG_ON(test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)); 4284 req->r_attempts = 0; 4285 req->r_num_fwd = fwd_seq; 4286 req->r_resend_mds = next_mds; 4287 put_request_session(req); 4288 __do_request(mdsc, req); 4289 } 4290 mutex_unlock(&mdsc->mutex); 4291 4292 /* kick calling process */ 4293 if (aborted) 4294 complete_request(mdsc, req); 4295 ceph_mdsc_put_request(req); 4296 return; 4297 4298 bad: 4299 pr_err_client(cl, "decode error err=%d\n", err); 4300 ceph_msg_dump(msg); 4301 } 4302 4303 static int __decode_session_metadata(void **p, void *end, 4304 bool *blocklisted) 4305 { 4306 /* map<string,string> */ 4307 u32 n; 4308 bool err_str; 4309 ceph_decode_32_safe(p, end, n, bad); 4310 while (n-- > 0) { 4311 u32 len; 4312 ceph_decode_32_safe(p, end, len, bad); 4313 ceph_decode_need(p, end, len, bad); 4314 err_str = !strncmp(*p, "error_string", len); 4315 *p += len; 4316 ceph_decode_32_safe(p, end, len, bad); 4317 ceph_decode_need(p, end, len, bad); 4318 /* 4319 * Match "blocklisted (blacklisted)" from newer MDSes, 4320 * or "blacklisted" from older MDSes. 4321 */ 4322 if (err_str && strnstr(*p, "blacklisted", len)) 4323 *blocklisted = true; 4324 *p += len; 4325 } 4326 return 0; 4327 bad: 4328 return -1; 4329 } 4330 4331 /* 4332 * handle a mds session control message 4333 */ 4334 static void handle_session(struct ceph_mds_session *session, 4335 struct ceph_msg *msg) 4336 { 4337 struct ceph_mds_client *mdsc = session->s_mdsc; 4338 struct ceph_client *cl = mdsc->fsc->client; 4339 int mds = session->s_mds; 4340 int msg_version = le16_to_cpu(msg->hdr.version); 4341 void *p = msg->front.iov_base; 4342 void *end = p + msg->front.iov_len; 4343 struct ceph_mds_session_head *h; 4344 struct ceph_mds_cap_auth *cap_auths = NULL; 4345 u32 op, cap_auths_num = 0; 4346 u64 seq, features = 0; 4347 int wake = 0; 4348 bool blocklisted = false; 4349 u32 i; 4350 4351 4352 /* decode */ 4353 ceph_decode_need(&p, end, sizeof(*h), bad); 4354 h = p; 4355 p += sizeof(*h); 4356 4357 op = le32_to_cpu(h->op); 4358 seq = le64_to_cpu(h->seq); 4359 4360 if (msg_version >= 3) { 4361 u32 len; 4362 /* version >= 2 and < 5, decode metadata, skip otherwise 4363 * as it's handled via flags. 4364 */ 4365 if (msg_version >= 5) 4366 ceph_decode_skip_map(&p, end, string, string, bad); 4367 else if (__decode_session_metadata(&p, end, &blocklisted) < 0) 4368 goto bad; 4369 4370 /* version >= 3, feature bits */ 4371 ceph_decode_32_safe(&p, end, len, bad); 4372 if (len) { 4373 ceph_decode_64_safe(&p, end, features, bad); 4374 p += len - sizeof(features); 4375 } 4376 } 4377 4378 if (msg_version >= 5) { 4379 u32 flags, len; 4380 4381 /* version >= 4 */ 4382 ceph_decode_skip_16(&p, end, bad); /* struct_v, struct_cv */ 4383 ceph_decode_32_safe(&p, end, len, bad); /* len */ 4384 ceph_decode_skip_n(&p, end, len, bad); /* metric_spec */ 4385 4386 /* version >= 5, flags */ 4387 ceph_decode_32_safe(&p, end, flags, bad); 4388 if (flags & CEPH_SESSION_BLOCKLISTED) { 4389 pr_warn_client(cl, "mds%d session blocklisted\n", 4390 session->s_mds); 4391 blocklisted = true; 4392 } 4393 } 4394 4395 if (msg_version >= 6) { 4396 ceph_decode_32_safe(&p, end, cap_auths_num, bad); 4397 doutc(cl, "cap_auths_num %d\n", cap_auths_num); 4398 4399 if (cap_auths_num && op != CEPH_SESSION_OPEN) { 4400 WARN_ON_ONCE(op != CEPH_SESSION_OPEN); 4401 goto skip_cap_auths; 4402 } 4403 4404 cap_auths = kzalloc_objs(struct ceph_mds_cap_auth, 4405 cap_auths_num); 4406 if (!cap_auths) { 4407 pr_err_client(cl, "No memory for cap_auths\n"); 4408 return; 4409 } 4410 4411 for (i = 0; i < cap_auths_num; i++) { 4412 u32 _len, j; 4413 4414 /* struct_v, struct_compat, and struct_len in MDSCapAuth */ 4415 ceph_decode_skip_n(&p, end, 2 + sizeof(u32), bad); 4416 4417 /* struct_v, struct_compat, and struct_len in MDSCapMatch */ 4418 ceph_decode_skip_n(&p, end, 2 + sizeof(u32), bad); 4419 ceph_decode_64_safe(&p, end, cap_auths[i].match.uid, bad); 4420 ceph_decode_32_safe(&p, end, _len, bad); 4421 if (_len) { 4422 cap_auths[i].match.gids = kcalloc(_len, sizeof(u32), 4423 GFP_KERNEL); 4424 if (!cap_auths[i].match.gids) { 4425 pr_err_client(cl, "No memory for gids\n"); 4426 goto fail; 4427 } 4428 4429 cap_auths[i].match.num_gids = _len; 4430 for (j = 0; j < _len; j++) 4431 ceph_decode_32_safe(&p, end, 4432 cap_auths[i].match.gids[j], 4433 bad); 4434 } 4435 4436 ceph_decode_32_safe(&p, end, _len, bad); 4437 if (_len) { 4438 cap_auths[i].match.path = kcalloc(_len + 1, sizeof(char), 4439 GFP_KERNEL); 4440 if (!cap_auths[i].match.path) { 4441 pr_err_client(cl, "No memory for path\n"); 4442 goto fail; 4443 } 4444 ceph_decode_copy(&p, cap_auths[i].match.path, _len); 4445 4446 /* Remove the tailing '/' */ 4447 while (_len && cap_auths[i].match.path[_len - 1] == '/') { 4448 cap_auths[i].match.path[_len - 1] = '\0'; 4449 _len -= 1; 4450 } 4451 } 4452 4453 ceph_decode_32_safe(&p, end, _len, bad); 4454 if (_len) { 4455 cap_auths[i].match.fs_name = kcalloc(_len + 1, sizeof(char), 4456 GFP_KERNEL); 4457 if (!cap_auths[i].match.fs_name) { 4458 pr_err_client(cl, "No memory for fs_name\n"); 4459 goto fail; 4460 } 4461 ceph_decode_copy(&p, cap_auths[i].match.fs_name, _len); 4462 } 4463 4464 ceph_decode_8_safe(&p, end, cap_auths[i].match.root_squash, bad); 4465 ceph_decode_8_safe(&p, end, cap_auths[i].readable, bad); 4466 ceph_decode_8_safe(&p, end, cap_auths[i].writeable, bad); 4467 doutc(cl, "uid %lld, num_gids %u, path %s, fs_name %s, root_squash %d, readable %d, writeable %d\n", 4468 cap_auths[i].match.uid, cap_auths[i].match.num_gids, 4469 cap_auths[i].match.path, cap_auths[i].match.fs_name, 4470 cap_auths[i].match.root_squash, 4471 cap_auths[i].readable, cap_auths[i].writeable); 4472 } 4473 } 4474 4475 skip_cap_auths: 4476 mutex_lock(&mdsc->mutex); 4477 if (op == CEPH_SESSION_OPEN) { 4478 if (mdsc->s_cap_auths) { 4479 for (i = 0; i < mdsc->s_cap_auths_num; i++) { 4480 kfree(mdsc->s_cap_auths[i].match.gids); 4481 kfree(mdsc->s_cap_auths[i].match.path); 4482 kfree(mdsc->s_cap_auths[i].match.fs_name); 4483 } 4484 kfree(mdsc->s_cap_auths); 4485 } 4486 mdsc->s_cap_auths_num = cap_auths_num; 4487 mdsc->s_cap_auths = cap_auths; 4488 4489 session->s_features = features; 4490 if (test_bit(CEPHFS_FEATURE_METRIC_COLLECT, 4491 &session->s_features)) 4492 ceph_metric_bind_session(mdsc, session); 4493 } 4494 if (op == CEPH_SESSION_CLOSE) { 4495 ceph_get_mds_session(session); 4496 __unregister_session(mdsc, session); 4497 } 4498 /* FIXME: this ttl calculation is generous */ 4499 session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose; 4500 mutex_unlock(&mdsc->mutex); 4501 4502 mutex_lock(&session->s_mutex); 4503 4504 doutc(cl, "mds%d %s %p state %s seq %llu\n", mds, 4505 ceph_session_op_name(op), session, 4506 ceph_session_state_name(session->s_state), seq); 4507 4508 if (session->s_state == CEPH_MDS_SESSION_HUNG) { 4509 session->s_state = CEPH_MDS_SESSION_OPEN; 4510 pr_info_client(cl, "mds%d came back\n", session->s_mds); 4511 } 4512 4513 switch (op) { 4514 case CEPH_SESSION_OPEN: 4515 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING) 4516 pr_info_client(cl, "mds%d reconnect success\n", 4517 session->s_mds); 4518 4519 if (test_bit(CEPHFS_FEATURE_SUBVOLUME_METRICS, 4520 &session->s_features)) 4521 ceph_subvolume_metrics_enable(&mdsc->subvol_metrics, true); 4522 else 4523 ceph_subvolume_metrics_enable(&mdsc->subvol_metrics, false); 4524 if (session->s_state == CEPH_MDS_SESSION_OPEN) { 4525 pr_notice_client(cl, "mds%d is already opened\n", 4526 session->s_mds); 4527 } else { 4528 session->s_state = CEPH_MDS_SESSION_OPEN; 4529 renewed_caps(mdsc, session, 0); 4530 if (test_bit(CEPHFS_FEATURE_METRIC_COLLECT, 4531 &session->s_features)) 4532 metric_schedule_delayed(&mdsc->metric); 4533 } 4534 4535 /* 4536 * The connection maybe broken and the session in client 4537 * side has been reinitialized, need to update the seq 4538 * anyway. 4539 */ 4540 if (!session->s_seq && seq) 4541 session->s_seq = seq; 4542 4543 wake = 1; 4544 if (mdsc->stopping) 4545 __close_session(mdsc, session); 4546 break; 4547 4548 case CEPH_SESSION_RENEWCAPS: 4549 if (session->s_renew_seq == seq) 4550 renewed_caps(mdsc, session, 1); 4551 break; 4552 4553 case CEPH_SESSION_CLOSE: 4554 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING) 4555 pr_info_client(cl, "mds%d reconnect denied\n", 4556 session->s_mds); 4557 session->s_state = CEPH_MDS_SESSION_CLOSED; 4558 cleanup_session_requests(mdsc, session); 4559 remove_session_caps(session); 4560 wake = 2; /* for good measure */ 4561 wake_up_all(&mdsc->session_close_wq); 4562 break; 4563 4564 case CEPH_SESSION_STALE: 4565 pr_info_client(cl, "mds%d caps went stale, renewing\n", 4566 session->s_mds); 4567 atomic_inc(&session->s_cap_gen); 4568 session->s_cap_ttl = jiffies - 1; 4569 send_renew_caps(mdsc, session); 4570 break; 4571 4572 case CEPH_SESSION_RECALL_STATE: 4573 ceph_trim_caps(mdsc, session, le32_to_cpu(h->max_caps)); 4574 break; 4575 4576 case CEPH_SESSION_FLUSHMSG: 4577 /* flush cap releases */ 4578 spin_lock(&session->s_cap_lock); 4579 if (session->s_num_cap_releases) 4580 ceph_flush_session_cap_releases(mdsc, session); 4581 spin_unlock(&session->s_cap_lock); 4582 4583 send_flushmsg_ack(mdsc, session, seq); 4584 break; 4585 4586 case CEPH_SESSION_FORCE_RO: 4587 doutc(cl, "force_session_readonly %p\n", session); 4588 spin_lock(&session->s_cap_lock); 4589 session->s_readonly = true; 4590 spin_unlock(&session->s_cap_lock); 4591 wake_up_session_caps(session, FORCE_RO); 4592 break; 4593 4594 case CEPH_SESSION_REJECT: 4595 WARN_ON(session->s_state != CEPH_MDS_SESSION_OPENING && 4596 session->s_state != CEPH_MDS_SESSION_RECONNECTING); 4597 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING) 4598 pr_info_client(cl, "mds%d reconnect rejected\n", 4599 session->s_mds); 4600 else 4601 pr_info_client(cl, "mds%d rejected session\n", 4602 session->s_mds); 4603 session->s_state = CEPH_MDS_SESSION_REJECTED; 4604 cleanup_session_requests(mdsc, session); 4605 remove_session_caps(session); 4606 if (blocklisted) 4607 mdsc->fsc->blocklisted = true; 4608 wake = 2; /* for good measure */ 4609 break; 4610 4611 default: 4612 pr_err_client(cl, "bad op %d mds%d\n", op, mds); 4613 WARN_ON(1); 4614 } 4615 4616 mutex_unlock(&session->s_mutex); 4617 if (wake) { 4618 mutex_lock(&mdsc->mutex); 4619 __wake_requests(mdsc, &session->s_waiting); 4620 if (wake == 2) 4621 kick_requests(mdsc, mds); 4622 mutex_unlock(&mdsc->mutex); 4623 } 4624 if (op == CEPH_SESSION_CLOSE) 4625 ceph_put_mds_session(session); 4626 return; 4627 4628 bad: 4629 pr_err_client(cl, "corrupt message mds%d len %d\n", mds, 4630 (int)msg->front.iov_len); 4631 ceph_msg_dump(msg); 4632 fail: 4633 for (i = 0; i < cap_auths_num; i++) { 4634 kfree(cap_auths[i].match.gids); 4635 kfree(cap_auths[i].match.path); 4636 kfree(cap_auths[i].match.fs_name); 4637 } 4638 kfree(cap_auths); 4639 return; 4640 } 4641 4642 void ceph_mdsc_release_dir_caps(struct ceph_mds_request *req) 4643 { 4644 struct ceph_client *cl = req->r_mdsc->fsc->client; 4645 int dcaps; 4646 4647 dcaps = xchg(&req->r_dir_caps, 0); 4648 if (dcaps) { 4649 doutc(cl, "releasing r_dir_caps=%s\n", ceph_cap_string(dcaps)); 4650 ceph_put_cap_refs(ceph_inode(req->r_parent), dcaps); 4651 } 4652 } 4653 4654 void ceph_mdsc_release_dir_caps_async(struct ceph_mds_request *req) 4655 { 4656 struct ceph_client *cl = req->r_mdsc->fsc->client; 4657 int dcaps; 4658 4659 dcaps = xchg(&req->r_dir_caps, 0); 4660 if (dcaps) { 4661 doutc(cl, "releasing r_dir_caps=%s\n", ceph_cap_string(dcaps)); 4662 ceph_put_cap_refs_async(ceph_inode(req->r_parent), dcaps); 4663 } 4664 } 4665 4666 /* 4667 * called under session->mutex. 4668 */ 4669 static void replay_unsafe_requests(struct ceph_mds_client *mdsc, 4670 struct ceph_mds_session *session) 4671 { 4672 struct ceph_mds_request *req, *nreq; 4673 struct rb_node *p; 4674 4675 doutc(mdsc->fsc->client, "mds%d\n", session->s_mds); 4676 4677 mutex_lock(&mdsc->mutex); 4678 list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item) 4679 __send_request(session, req, true); 4680 4681 /* 4682 * also re-send old requests when MDS enters reconnect stage. So that MDS 4683 * can process completed request in clientreplay stage. 4684 */ 4685 p = rb_first(&mdsc->request_tree); 4686 while (p) { 4687 req = rb_entry(p, struct ceph_mds_request, r_node); 4688 p = rb_next(p); 4689 if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) 4690 continue; 4691 if (req->r_attempts == 0) 4692 continue; /* only old requests */ 4693 if (!req->r_session) 4694 continue; 4695 if (req->r_session->s_mds != session->s_mds) 4696 continue; 4697 4698 ceph_mdsc_release_dir_caps_async(req); 4699 4700 __send_request(session, req, true); 4701 } 4702 mutex_unlock(&mdsc->mutex); 4703 } 4704 4705 static int send_reconnect_partial(struct ceph_reconnect_state *recon_state) 4706 { 4707 struct ceph_msg *reply; 4708 struct ceph_pagelist *_pagelist; 4709 struct page *page; 4710 __le32 *addr; 4711 int err = -ENOMEM; 4712 4713 if (!recon_state->allow_multi) 4714 return -ENOSPC; 4715 4716 /* can't handle message that contains both caps and realm */ 4717 BUG_ON(!recon_state->nr_caps == !recon_state->nr_realms); 4718 4719 /* pre-allocate new pagelist */ 4720 _pagelist = ceph_pagelist_alloc(GFP_NOFS); 4721 if (!_pagelist) 4722 return -ENOMEM; 4723 4724 reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false); 4725 if (!reply) 4726 goto fail_msg; 4727 4728 /* placeholder for nr_caps */ 4729 err = ceph_pagelist_encode_32(_pagelist, 0); 4730 if (err < 0) 4731 goto fail; 4732 4733 if (recon_state->nr_caps) { 4734 /* currently encoding caps */ 4735 err = ceph_pagelist_encode_32(recon_state->pagelist, 0); 4736 if (err) 4737 goto fail; 4738 } else { 4739 /* placeholder for nr_realms (currently encoding relams) */ 4740 err = ceph_pagelist_encode_32(_pagelist, 0); 4741 if (err < 0) 4742 goto fail; 4743 } 4744 4745 err = ceph_pagelist_encode_8(recon_state->pagelist, 1); 4746 if (err) 4747 goto fail; 4748 4749 page = list_first_entry(&recon_state->pagelist->head, struct page, lru); 4750 addr = kmap_atomic(page); 4751 if (recon_state->nr_caps) { 4752 /* currently encoding caps */ 4753 *addr = cpu_to_le32(recon_state->nr_caps); 4754 } else { 4755 /* currently encoding relams */ 4756 *(addr + 1) = cpu_to_le32(recon_state->nr_realms); 4757 } 4758 kunmap_atomic(addr); 4759 4760 reply->hdr.version = cpu_to_le16(5); 4761 reply->hdr.compat_version = cpu_to_le16(4); 4762 4763 reply->hdr.data_len = cpu_to_le32(recon_state->pagelist->length); 4764 ceph_msg_data_add_pagelist(reply, recon_state->pagelist); 4765 4766 ceph_con_send(&recon_state->session->s_con, reply); 4767 ceph_pagelist_release(recon_state->pagelist); 4768 4769 recon_state->pagelist = _pagelist; 4770 recon_state->nr_caps = 0; 4771 recon_state->nr_realms = 0; 4772 recon_state->msg_version = 5; 4773 return 0; 4774 fail: 4775 ceph_msg_put(reply); 4776 fail_msg: 4777 ceph_pagelist_release(_pagelist); 4778 return err; 4779 } 4780 4781 static struct dentry* d_find_primary(struct inode *inode) 4782 { 4783 struct dentry *alias, *dn = NULL; 4784 4785 if (hlist_empty(&inode->i_dentry)) 4786 return NULL; 4787 4788 spin_lock(&inode->i_lock); 4789 if (hlist_empty(&inode->i_dentry)) 4790 goto out_unlock; 4791 4792 if (S_ISDIR(inode->i_mode)) { 4793 alias = hlist_entry(inode->i_dentry.first, struct dentry, d_alias); 4794 if (!IS_ROOT(alias)) 4795 dn = dget(alias); 4796 goto out_unlock; 4797 } 4798 4799 for_each_alias(alias, inode) { 4800 spin_lock(&alias->d_lock); 4801 if (!d_unhashed(alias) && 4802 (ceph_dentry(alias)->flags & CEPH_DENTRY_PRIMARY_LINK)) { 4803 dn = dget_dlock(alias); 4804 } 4805 spin_unlock(&alias->d_lock); 4806 if (dn) 4807 break; 4808 } 4809 out_unlock: 4810 spin_unlock(&inode->i_lock); 4811 return dn; 4812 } 4813 4814 /* 4815 * Encode information about a cap for a reconnect with the MDS. 4816 */ 4817 static int reconnect_caps_cb(struct inode *inode, int mds, void *arg) 4818 { 4819 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb); 4820 struct ceph_client *cl = ceph_inode_to_client(inode); 4821 union { 4822 struct ceph_mds_cap_reconnect v2; 4823 struct ceph_mds_cap_reconnect_v1 v1; 4824 } rec; 4825 struct ceph_inode_info *ci = ceph_inode(inode); 4826 struct ceph_reconnect_state *recon_state = arg; 4827 struct ceph_pagelist *pagelist = recon_state->pagelist; 4828 struct dentry *dentry; 4829 struct ceph_cap *cap; 4830 struct ceph_path_info path_info = {0}; 4831 int err; 4832 u64 snap_follows; 4833 4834 dentry = d_find_primary(inode); 4835 if (dentry) { 4836 /* set pathbase to parent dir when msg_version >= 2 */ 4837 char *path = ceph_mdsc_build_path(mdsc, dentry, &path_info, 4838 recon_state->msg_version >= 2); 4839 dput(dentry); 4840 if (IS_ERR(path)) { 4841 err = PTR_ERR(path); 4842 goto out_err; 4843 } 4844 } 4845 4846 spin_lock(&ci->i_ceph_lock); 4847 cap = __get_cap_for_mds(ci, mds); 4848 if (!cap) { 4849 spin_unlock(&ci->i_ceph_lock); 4850 err = 0; 4851 goto out_err; 4852 } 4853 doutc(cl, " adding %p ino %llx.%llx cap %p %lld %s\n", inode, 4854 ceph_vinop(inode), cap, cap->cap_id, 4855 ceph_cap_string(cap->issued)); 4856 4857 cap->seq = 0; /* reset cap seq */ 4858 cap->issue_seq = 0; /* and issue_seq */ 4859 cap->mseq = 0; /* and migrate_seq */ 4860 cap->cap_gen = atomic_read(&cap->session->s_cap_gen); 4861 4862 /* 4863 * Note: CEPH_I_ERROR_FILELOCK is not set during reconnect. 4864 * Instead, locks are submitted for best-effort MDS reclaim 4865 * via the flock_len field below. If reclaim fails (e.g., 4866 * another client grabbed a conflicting lock), future lock 4867 * operations will fail and set the error flag at that point. 4868 */ 4869 4870 /* These are lost when the session goes away */ 4871 if (S_ISDIR(inode->i_mode)) { 4872 if (cap->issued & CEPH_CAP_DIR_CREATE) { 4873 ceph_put_string(rcu_dereference_raw(ci->i_cached_layout.pool_ns)); 4874 memset(&ci->i_cached_layout, 0, sizeof(ci->i_cached_layout)); 4875 } 4876 cap->issued &= ~CEPH_CAP_ANY_DIR_OPS; 4877 } 4878 4879 if (recon_state->msg_version >= 2) { 4880 rec.v2.cap_id = cpu_to_le64(cap->cap_id); 4881 rec.v2.wanted = cpu_to_le32(__ceph_caps_wanted(ci)); 4882 rec.v2.issued = cpu_to_le32(cap->issued); 4883 rec.v2.snaprealm = cpu_to_le64(ci->i_snap_realm->ino); 4884 rec.v2.pathbase = cpu_to_le64(path_info.vino.ino); 4885 rec.v2.flock_len = cpu_to_le32( 4886 test_bit(CEPH_I_ERROR_FILELOCK_BIT, 4887 &ci->i_ceph_flags) ? 0 : 1); 4888 } else { 4889 struct timespec64 ts; 4890 4891 rec.v1.cap_id = cpu_to_le64(cap->cap_id); 4892 rec.v1.wanted = cpu_to_le32(__ceph_caps_wanted(ci)); 4893 rec.v1.issued = cpu_to_le32(cap->issued); 4894 rec.v1.size = cpu_to_le64(i_size_read(inode)); 4895 ts = inode_get_mtime(inode); 4896 ceph_encode_timespec64(&rec.v1.mtime, &ts); 4897 ts = inode_get_atime(inode); 4898 ceph_encode_timespec64(&rec.v1.atime, &ts); 4899 rec.v1.snaprealm = cpu_to_le64(ci->i_snap_realm->ino); 4900 rec.v1.pathbase = cpu_to_le64(path_info.vino.ino); 4901 } 4902 4903 if (list_empty(&ci->i_cap_snaps)) { 4904 snap_follows = ci->i_head_snapc ? ci->i_head_snapc->seq : 0; 4905 } else { 4906 struct ceph_cap_snap *capsnap = 4907 list_first_entry(&ci->i_cap_snaps, 4908 struct ceph_cap_snap, ci_item); 4909 snap_follows = capsnap->follows; 4910 } 4911 spin_unlock(&ci->i_ceph_lock); 4912 4913 if (recon_state->msg_version >= 2) { 4914 int num_fcntl_locks, num_flock_locks; 4915 struct ceph_filelock *flocks = NULL; 4916 size_t struct_len, total_len = sizeof(u64); 4917 u8 struct_v = 0; 4918 4919 encode_again: 4920 if (rec.v2.flock_len) { 4921 ceph_count_locks(inode, &num_fcntl_locks, &num_flock_locks); 4922 } else { 4923 num_fcntl_locks = 0; 4924 num_flock_locks = 0; 4925 } 4926 if (num_fcntl_locks + num_flock_locks > 0) { 4927 flocks = kmalloc_objs(struct ceph_filelock, 4928 num_fcntl_locks + num_flock_locks, 4929 GFP_NOFS); 4930 if (!flocks) { 4931 err = -ENOMEM; 4932 goto out_err; 4933 } 4934 err = ceph_encode_locks_to_buffer(inode, flocks, 4935 num_fcntl_locks, 4936 num_flock_locks); 4937 if (err) { 4938 kfree(flocks); 4939 flocks = NULL; 4940 if (err == -ENOSPC) 4941 goto encode_again; 4942 goto out_err; 4943 } 4944 } else { 4945 kfree(flocks); 4946 flocks = NULL; 4947 } 4948 4949 if (recon_state->msg_version >= 3) { 4950 /* version, compat_version and struct_len */ 4951 total_len += 2 * sizeof(u8) + sizeof(u32); 4952 struct_v = 2; 4953 } 4954 /* 4955 * number of encoded locks is stable, so copy to pagelist 4956 */ 4957 struct_len = 2 * sizeof(u32) + 4958 (num_fcntl_locks + num_flock_locks) * 4959 sizeof(struct ceph_filelock); 4960 rec.v2.flock_len = cpu_to_le32(struct_len); 4961 4962 struct_len += sizeof(u32) + path_info.pathlen + sizeof(rec.v2); 4963 4964 if (struct_v >= 2) 4965 struct_len += sizeof(u64); /* snap_follows */ 4966 4967 total_len += struct_len; 4968 4969 if (pagelist->length + total_len > RECONNECT_MAX_SIZE) { 4970 err = send_reconnect_partial(recon_state); 4971 if (err) 4972 goto out_freeflocks; 4973 pagelist = recon_state->pagelist; 4974 } 4975 4976 err = ceph_pagelist_reserve(pagelist, total_len); 4977 if (err) 4978 goto out_freeflocks; 4979 4980 ceph_pagelist_encode_64(pagelist, ceph_ino(inode)); 4981 if (recon_state->msg_version >= 3) { 4982 ceph_pagelist_encode_8(pagelist, struct_v); 4983 ceph_pagelist_encode_8(pagelist, 1); 4984 ceph_pagelist_encode_32(pagelist, struct_len); 4985 } 4986 ceph_pagelist_encode_string(pagelist, (char *)path_info.path, path_info.pathlen); 4987 ceph_pagelist_append(pagelist, &rec, sizeof(rec.v2)); 4988 ceph_locks_to_pagelist(flocks, pagelist, 4989 num_fcntl_locks, num_flock_locks); 4990 if (struct_v >= 2) 4991 ceph_pagelist_encode_64(pagelist, snap_follows); 4992 out_freeflocks: 4993 kfree(flocks); 4994 } else { 4995 err = ceph_pagelist_reserve(pagelist, 4996 sizeof(u64) + sizeof(u32) + 4997 path_info.pathlen + sizeof(rec.v1)); 4998 if (err) 4999 goto out_err; 5000 5001 ceph_pagelist_encode_64(pagelist, ceph_ino(inode)); 5002 ceph_pagelist_encode_string(pagelist, (char *)path_info.path, path_info.pathlen); 5003 ceph_pagelist_append(pagelist, &rec, sizeof(rec.v1)); 5004 } 5005 5006 out_err: 5007 ceph_mdsc_free_path_info(&path_info); 5008 if (!err) 5009 recon_state->nr_caps++; 5010 return err; 5011 } 5012 5013 static int encode_snap_realms(struct ceph_mds_client *mdsc, 5014 struct ceph_reconnect_state *recon_state) 5015 { 5016 struct rb_node *p; 5017 struct ceph_pagelist *pagelist = recon_state->pagelist; 5018 struct ceph_client *cl = mdsc->fsc->client; 5019 int err = 0; 5020 5021 if (recon_state->msg_version >= 4) { 5022 err = ceph_pagelist_encode_32(pagelist, mdsc->num_snap_realms); 5023 if (err < 0) 5024 goto fail; 5025 } 5026 5027 /* 5028 * snaprealms. we provide mds with the ino, seq (version), and 5029 * parent for all of our realms. If the mds has any newer info, 5030 * it will tell us. 5031 */ 5032 for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) { 5033 struct ceph_snap_realm *realm = 5034 rb_entry(p, struct ceph_snap_realm, node); 5035 struct ceph_mds_snaprealm_reconnect sr_rec; 5036 5037 if (recon_state->msg_version >= 4) { 5038 size_t need = sizeof(u8) * 2 + sizeof(u32) + 5039 sizeof(sr_rec); 5040 5041 if (pagelist->length + need > RECONNECT_MAX_SIZE) { 5042 err = send_reconnect_partial(recon_state); 5043 if (err) 5044 goto fail; 5045 pagelist = recon_state->pagelist; 5046 } 5047 5048 err = ceph_pagelist_reserve(pagelist, need); 5049 if (err) 5050 goto fail; 5051 5052 ceph_pagelist_encode_8(pagelist, 1); 5053 ceph_pagelist_encode_8(pagelist, 1); 5054 ceph_pagelist_encode_32(pagelist, sizeof(sr_rec)); 5055 } 5056 5057 doutc(cl, " adding snap realm %llx seq %lld parent %llx\n", 5058 realm->ino, realm->seq, realm->parent_ino); 5059 sr_rec.ino = cpu_to_le64(realm->ino); 5060 sr_rec.seq = cpu_to_le64(realm->seq); 5061 sr_rec.parent = cpu_to_le64(realm->parent_ino); 5062 5063 err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec)); 5064 if (err) 5065 goto fail; 5066 5067 recon_state->nr_realms++; 5068 } 5069 fail: 5070 return err; 5071 } 5072 5073 5074 /* 5075 * If an MDS fails and recovers, clients need to reconnect in order to 5076 * reestablish shared state. This includes all caps issued through 5077 * this session _and_ the snap_realm hierarchy. Because it's not 5078 * clear which snap realms the mds cares about, we send everything we 5079 * know about.. that ensures we'll then get any new info the 5080 * recovering MDS might have. 5081 * 5082 * This is a relatively heavyweight operation, but it's rare. 5083 */ 5084 static int send_mds_reconnect(struct ceph_mds_client *mdsc, 5085 struct ceph_mds_session *session) 5086 { 5087 struct ceph_client *cl = mdsc->fsc->client; 5088 struct ceph_msg *reply; 5089 int mds = session->s_mds; 5090 int err = -ENOMEM; 5091 int old_state; 5092 struct ceph_reconnect_state recon_state = { 5093 .session = session, 5094 }; 5095 LIST_HEAD(dispose); 5096 5097 recon_state.pagelist = ceph_pagelist_alloc(GFP_NOFS); 5098 if (!recon_state.pagelist) 5099 goto fail_nopagelist; 5100 5101 reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false); 5102 if (!reply) 5103 goto fail_nomsg; 5104 5105 mutex_lock(&session->s_mutex); 5106 5107 /* Serialized by s_mutex against concurrent ceph_get_deleg_ino(). */ 5108 xa_destroy(&session->s_delegated_inos); 5109 if (session->s_state == CEPH_MDS_SESSION_CLOSED || 5110 session->s_state == CEPH_MDS_SESSION_REJECTED) { 5111 pr_info_client(cl, "mds%d skipping reconnect, session %s\n", 5112 mds, 5113 ceph_session_state_name(session->s_state)); 5114 mutex_unlock(&session->s_mutex); 5115 ceph_msg_put(reply); 5116 err = -ESTALE; 5117 goto fail_return; 5118 } 5119 5120 /* s_mutex -> mdsc->mutex matches cleanup_session_requests() order. */ 5121 mutex_lock(&mdsc->mutex); 5122 if (mds >= mdsc->max_sessions || mdsc->sessions[mds] != session) { 5123 mutex_unlock(&mdsc->mutex); 5124 pr_info_client(cl, 5125 "mds%d skipping reconnect, session unregistered\n", 5126 mds); 5127 mutex_unlock(&session->s_mutex); 5128 ceph_msg_put(reply); 5129 err = -ENOENT; 5130 goto fail_return; 5131 } 5132 mutex_unlock(&mdsc->mutex); 5133 5134 pr_info_client(cl, "mds%d reconnect start\n", mds); 5135 old_state = session->s_state; 5136 session->s_state = CEPH_MDS_SESSION_RECONNECTING; 5137 session->s_seq = 0; 5138 5139 doutc(cl, "session %p state %s\n", session, 5140 ceph_session_state_name(session->s_state)); 5141 5142 atomic_inc(&session->s_cap_gen); 5143 5144 spin_lock(&session->s_cap_lock); 5145 /* don't know if session is readonly */ 5146 session->s_readonly = 0; 5147 /* 5148 * notify __ceph_remove_cap() that we are composing cap reconnect. 5149 * If a cap get released before being added to the cap reconnect, 5150 * __ceph_remove_cap() should skip queuing cap release. 5151 */ 5152 session->s_cap_reconnect = 1; 5153 /* drop old cap expires; we're about to reestablish that state */ 5154 detach_cap_releases(session, &dispose); 5155 spin_unlock(&session->s_cap_lock); 5156 dispose_cap_releases(mdsc, &dispose); 5157 5158 /* trim unused caps to reduce MDS's cache rejoin time */ 5159 if (mdsc->fsc->sb->s_root) 5160 shrink_dcache_parent(mdsc->fsc->sb->s_root); 5161 5162 ceph_con_close(&session->s_con); 5163 ceph_con_open(&session->s_con, 5164 CEPH_ENTITY_TYPE_MDS, mds, 5165 ceph_mdsmap_get_addr(mdsc->mdsmap, mds)); 5166 5167 /* replay unsafe requests */ 5168 replay_unsafe_requests(mdsc, session); 5169 5170 ceph_early_kick_flushing_caps(mdsc, session); 5171 5172 down_read(&mdsc->snap_rwsem); 5173 5174 /* placeholder for nr_caps */ 5175 err = ceph_pagelist_encode_32(recon_state.pagelist, 0); 5176 if (err) 5177 goto fail_clear_cap_reconnect; 5178 5179 if (test_bit(CEPHFS_FEATURE_MULTI_RECONNECT, &session->s_features)) { 5180 recon_state.msg_version = 3; 5181 recon_state.allow_multi = true; 5182 } else if (session->s_con.peer_features & CEPH_FEATURE_MDSENC) { 5183 recon_state.msg_version = 3; 5184 } else { 5185 recon_state.msg_version = 2; 5186 } 5187 /* traverse this session's caps */ 5188 err = ceph_iterate_session_caps(session, reconnect_caps_cb, &recon_state); 5189 5190 spin_lock(&session->s_cap_lock); 5191 session->s_cap_reconnect = 0; 5192 spin_unlock(&session->s_cap_lock); 5193 5194 if (err < 0) 5195 goto fail; 5196 5197 /* check if all realms can be encoded into current message */ 5198 if (mdsc->num_snap_realms) { 5199 size_t total_len = 5200 recon_state.pagelist->length + 5201 mdsc->num_snap_realms * 5202 sizeof(struct ceph_mds_snaprealm_reconnect); 5203 if (recon_state.msg_version >= 4) { 5204 /* number of realms */ 5205 total_len += sizeof(u32); 5206 /* version, compat_version and struct_len */ 5207 total_len += mdsc->num_snap_realms * 5208 (2 * sizeof(u8) + sizeof(u32)); 5209 } 5210 if (total_len > RECONNECT_MAX_SIZE) { 5211 if (!recon_state.allow_multi) { 5212 err = -ENOSPC; 5213 goto fail; 5214 } 5215 if (recon_state.nr_caps) { 5216 err = send_reconnect_partial(&recon_state); 5217 if (err) 5218 goto fail; 5219 } 5220 recon_state.msg_version = 5; 5221 } 5222 } 5223 5224 err = encode_snap_realms(mdsc, &recon_state); 5225 if (err < 0) 5226 goto fail; 5227 5228 if (recon_state.msg_version >= 5) { 5229 err = ceph_pagelist_encode_8(recon_state.pagelist, 0); 5230 if (err < 0) 5231 goto fail; 5232 } 5233 5234 if (recon_state.nr_caps || recon_state.nr_realms) { 5235 struct page *page = 5236 list_first_entry(&recon_state.pagelist->head, 5237 struct page, lru); 5238 __le32 *addr = kmap_atomic(page); 5239 if (recon_state.nr_caps) { 5240 WARN_ON(recon_state.nr_realms != mdsc->num_snap_realms); 5241 *addr = cpu_to_le32(recon_state.nr_caps); 5242 } else if (recon_state.msg_version >= 4) { 5243 *(addr + 1) = cpu_to_le32(recon_state.nr_realms); 5244 } 5245 kunmap_atomic(addr); 5246 } 5247 5248 reply->hdr.version = cpu_to_le16(recon_state.msg_version); 5249 if (recon_state.msg_version >= 4) 5250 reply->hdr.compat_version = cpu_to_le16(4); 5251 5252 reply->hdr.data_len = cpu_to_le32(recon_state.pagelist->length); 5253 ceph_msg_data_add_pagelist(reply, recon_state.pagelist); 5254 5255 ceph_con_send(&session->s_con, reply); 5256 5257 mutex_unlock(&session->s_mutex); 5258 5259 mutex_lock(&mdsc->mutex); 5260 __wake_requests(mdsc, &session->s_waiting); 5261 mutex_unlock(&mdsc->mutex); 5262 5263 up_read(&mdsc->snap_rwsem); 5264 ceph_pagelist_release(recon_state.pagelist); 5265 return 0; 5266 5267 fail_clear_cap_reconnect: 5268 spin_lock(&session->s_cap_lock); 5269 session->s_cap_reconnect = 0; 5270 spin_unlock(&session->s_cap_lock); 5271 fail: 5272 ceph_msg_put(reply); 5273 up_read(&mdsc->snap_rwsem); 5274 /* 5275 * Restore prior session state so map-driven reconnect logic 5276 * (check_new_map) can retry. Without this, a transient build 5277 * failure strands the session in RECONNECTING indefinitely. 5278 */ 5279 session->s_state = old_state; 5280 mutex_unlock(&session->s_mutex); 5281 fail_nomsg: 5282 ceph_pagelist_release(recon_state.pagelist); 5283 fail_nopagelist: 5284 pr_err_client(cl, "error %d preparing reconnect for mds%d\n", 5285 err, mds); 5286 return err; 5287 5288 fail_return: 5289 /* 5290 * Early-exit path for expected concurrent-teardown races 5291 * (-ESTALE for closed/rejected sessions, -ENOENT for 5292 * unregistered sessions). Skip the pr_err_client diagnostic 5293 * since these are not genuine reconnect build failures. 5294 */ 5295 ceph_pagelist_release(recon_state.pagelist); 5296 return err; 5297 } 5298 5299 const char *ceph_reset_phase_name(enum ceph_client_reset_phase phase) 5300 { 5301 switch (phase) { 5302 case CEPH_CLIENT_RESET_IDLE: return "idle"; 5303 case CEPH_CLIENT_RESET_QUIESCING: return "quiescing"; 5304 case CEPH_CLIENT_RESET_DRAINING: return "draining"; 5305 case CEPH_CLIENT_RESET_TEARDOWN: return "teardown"; 5306 default: return "unknown"; 5307 } 5308 } 5309 5310 /** 5311 * ceph_mdsc_wait_for_reset - wait for an active reset to complete 5312 * @mdsc: MDS client 5313 * 5314 * Returns 0 if reset completed successfully or no reset was active. 5315 * Returns -EAGAIN if reset completed with an error, signalling the 5316 * caller to retry. The internal error (e.g. -ENOMEM) is not propagated 5317 * because callers like open() or flock() have no way to act on 5318 * work-function internals. The detailed error is available via debugfs 5319 * reset/status and tracepoints. 5320 * Returns -ETIMEDOUT if we timed out waiting. 5321 * Returns -ERESTARTSYS if interrupted by signal. 5322 */ 5323 int ceph_mdsc_wait_for_reset(struct ceph_mds_client *mdsc) 5324 { 5325 struct ceph_client_reset_state *st = &mdsc->reset_state; 5326 struct ceph_client *cl = mdsc->fsc->client; 5327 unsigned long deadline = jiffies + CEPH_CLIENT_RESET_WAIT_TIMEOUT_SEC * HZ; 5328 int blocked_count; 5329 long remaining; 5330 long wait_ret; 5331 int ret; 5332 5333 if (ceph_reset_is_idle(st)) 5334 return 0; 5335 5336 blocked_count = atomic_inc_return(&st->blocked_requests); 5337 doutc(cl, "request blocked during reset, %d total blocked\n", 5338 blocked_count); 5339 trace_ceph_client_reset_blocked(mdsc, blocked_count); 5340 5341 retry: 5342 remaining = max_t(long, deadline - jiffies, 1); 5343 wait_ret = wait_event_interruptible_timeout(st->blocked_wq, 5344 ceph_reset_is_idle(st), 5345 remaining); 5346 5347 if (wait_ret == 0) { 5348 atomic_dec(&st->blocked_requests); 5349 pr_warn_client(cl, "timed out waiting for reset to complete\n"); 5350 trace_ceph_client_reset_unblocked(mdsc, -ETIMEDOUT); 5351 return -ETIMEDOUT; 5352 } 5353 if (wait_ret < 0) { 5354 atomic_dec(&st->blocked_requests); 5355 trace_ceph_client_reset_unblocked(mdsc, (int)wait_ret); 5356 return (int)wait_ret; /* -ERESTARTSYS */ 5357 } 5358 5359 /* 5360 * Verify phase is still IDLE under the lock. If another reset 5361 * was scheduled between the wake-up and this check, loop back 5362 * and wait for it to finish rather than returning a stale result. 5363 */ 5364 spin_lock(&st->lock); 5365 if (st->phase != CEPH_CLIENT_RESET_IDLE) { 5366 spin_unlock(&st->lock); 5367 if (time_before(jiffies, deadline)) 5368 goto retry; 5369 atomic_dec(&st->blocked_requests); 5370 trace_ceph_client_reset_unblocked(mdsc, -ETIMEDOUT); 5371 return -ETIMEDOUT; 5372 } 5373 ret = st->last_errno; 5374 spin_unlock(&st->lock); 5375 5376 atomic_dec(&st->blocked_requests); 5377 trace_ceph_client_reset_unblocked(mdsc, ret); 5378 return ret ? -EAGAIN : 0; 5379 } 5380 5381 static void ceph_mdsc_reset_complete(struct ceph_mds_client *mdsc, int ret) 5382 { 5383 struct ceph_client_reset_state *st = &mdsc->reset_state; 5384 5385 spin_lock(&st->lock); 5386 /* 5387 * If destroy already marked us as shut down, it owns the 5388 * final bookkeeping and waiter wakeup. Just bail so we 5389 * don't overwrite its state. 5390 */ 5391 if (st->shutdown) { 5392 spin_unlock(&st->lock); 5393 return; 5394 } 5395 st->last_finish = jiffies; 5396 st->last_errno = ret; 5397 st->phase = CEPH_CLIENT_RESET_IDLE; 5398 if (ret) 5399 st->failure_count++; 5400 else 5401 st->success_count++; 5402 spin_unlock(&st->lock); 5403 5404 /* Wake up all requests that were blocked waiting for reset */ 5405 wake_up_all(&st->blocked_wq); 5406 5407 trace_ceph_client_reset_complete(mdsc, ret); 5408 } 5409 5410 static void ceph_mdsc_reset_workfn(struct work_struct *work) 5411 { 5412 struct ceph_mds_client *mdsc = 5413 container_of(work, struct ceph_mds_client, reset_work); 5414 struct ceph_client_reset_state *st = &mdsc->reset_state; 5415 struct ceph_client *cl = mdsc->fsc->client; 5416 struct ceph_mds_session **sessions = NULL; 5417 char reason[CEPH_CLIENT_RESET_REASON_LEN]; 5418 unsigned long drain_deadline; 5419 int max_sessions, i, n = 0, torn_down = 0; 5420 int ret = 0; 5421 5422 spin_lock(&st->lock); 5423 strscpy(reason, st->last_reason, sizeof(reason)); 5424 spin_unlock(&st->lock); 5425 5426 mutex_lock(&mdsc->mutex); 5427 max_sessions = mdsc->max_sessions; 5428 if (max_sessions <= 0) { 5429 mutex_unlock(&mdsc->mutex); 5430 goto out_complete; 5431 } 5432 5433 sessions = kcalloc(max_sessions, sizeof(*sessions), GFP_KERNEL); 5434 if (!sessions) { 5435 mutex_unlock(&mdsc->mutex); 5436 ret = -ENOMEM; 5437 pr_err_client(cl, 5438 "manual session reset failed to allocate session array\n"); 5439 ceph_mdsc_reset_complete(mdsc, ret); 5440 return; 5441 } 5442 5443 for (i = 0; i < max_sessions; i++) { 5444 struct ceph_mds_session *session = mdsc->sessions[i]; 5445 5446 if (!session) 5447 continue; 5448 5449 /* 5450 * Read session state without s_mutex to avoid nesting 5451 * mdsc->mutex -> s_mutex, which would invert the 5452 * s_mutex -> mdsc->mutex order used by 5453 * cleanup_session_requests(). s_state is an int 5454 * so loads are atomic; the teardown loop below 5455 * handles races with concurrent state transitions. 5456 */ 5457 switch (READ_ONCE(session->s_state)) { 5458 case CEPH_MDS_SESSION_OPEN: 5459 case CEPH_MDS_SESSION_HUNG: 5460 case CEPH_MDS_SESSION_OPENING: 5461 case CEPH_MDS_SESSION_RESTARTING: 5462 case CEPH_MDS_SESSION_RECONNECTING: 5463 case CEPH_MDS_SESSION_CLOSING: 5464 sessions[n++] = ceph_get_mds_session(session); 5465 break; 5466 default: 5467 pr_info_client(cl, 5468 "mds%d in state %s, skipping reset\n", 5469 session->s_mds, 5470 ceph_session_state_name(session->s_state)); 5471 break; 5472 } 5473 } 5474 mutex_unlock(&mdsc->mutex); 5475 5476 pr_info_client(cl, 5477 "manual session reset executing (sessions=%d, reason=\"%s\")\n", 5478 n, reason); 5479 5480 if (n == 0) { 5481 kfree(sessions); 5482 goto out_complete; 5483 } 5484 5485 spin_lock(&st->lock); 5486 if (st->shutdown) { 5487 spin_unlock(&st->lock); 5488 goto out_sessions; 5489 } 5490 st->phase = CEPH_CLIENT_RESET_DRAINING; 5491 spin_unlock(&st->lock); 5492 5493 /* 5494 * Best-effort drain: flush dirty state while sessions are still 5495 * alive. New requests are blocked while phase != IDLE. 5496 * The sessions are functional, so non-stuck state drains normally. 5497 * Stuck state (the cause of the stalemate the operator is trying 5498 * to break) will not drain -- that is expected, and we proceed to 5499 * forced teardown after the timeout. 5500 * 5501 * Four things are drained: 5502 * 1. MDS journal -- send_flush_mdlog asks each MDS to journal 5503 * pending unsafe operations (creates, renames, setattrs). 5504 * 2. Unsafe requests -- bounded wait for each unsafe write 5505 * request to reach safe status via r_safe_completion. 5506 * 3. Dirty caps -- ceph_flush_dirty_caps triggers cap flush on 5507 * all sessions. Non-stuck caps flush in milliseconds. 5508 * 4. Cap releases -- push pending cap release messages. 5509 * 5510 * The unsafe-request wait and cap-flush wait below provide 5511 * the bounded drain window during which all categories can 5512 * make progress. 5513 */ 5514 for (i = 0; i < n; i++) 5515 send_flush_mdlog(sessions[i]); 5516 5517 /* 5518 * Both drain legs (unsafe requests and cap flushes) share a 5519 * single deadline so the total drain time is bounded at 5520 * CEPH_CLIENT_RESET_DRAIN_SEC. 5521 */ 5522 drain_deadline = jiffies + CEPH_CLIENT_RESET_DRAIN_SEC * HZ; 5523 5524 /* 5525 * Wait for unsafe write requests (creates, renames, setattrs) 5526 * to reach safe status. Uses the same pattern as 5527 * flush_mdlog_and_wait_mdsc_unsafe_requests() but bounded by 5528 * the shared drain deadline. Requests that do not complete within 5529 * the window are force-dropped during teardown. 5530 */ 5531 { 5532 struct ceph_mds_request *req; 5533 struct rb_node *rn; 5534 u64 last_tid; 5535 5536 mutex_lock(&mdsc->mutex); 5537 last_tid = mdsc->last_tid; 5538 mutex_unlock(&mdsc->mutex); 5539 5540 mutex_lock(&mdsc->mutex); 5541 rn = rb_first(&mdsc->request_tree); 5542 while (rn) { 5543 req = rb_entry(rn, struct ceph_mds_request, r_node); 5544 if (req->r_tid > last_tid) 5545 break; 5546 if (req->r_op == CEPH_MDS_OP_SETFILELOCK || 5547 !(req->r_op & CEPH_MDS_OP_WRITE)) { 5548 rn = rb_next(rn); 5549 continue; 5550 } 5551 ceph_mdsc_get_request(req); 5552 mutex_unlock(&mdsc->mutex); 5553 5554 wait_for_completion_timeout(&req->r_safe_completion, 5555 max_t(long, drain_deadline - jiffies, 1)); 5556 5557 mutex_lock(&mdsc->mutex); 5558 ceph_mdsc_put_request(req); 5559 if (time_after(jiffies, drain_deadline)) 5560 break; 5561 rn = rb_first(&mdsc->request_tree); 5562 } 5563 mutex_unlock(&mdsc->mutex); 5564 5565 if (time_after_eq(jiffies, drain_deadline)) 5566 WRITE_ONCE(st->drain_timed_out, true); 5567 } 5568 5569 ceph_flush_dirty_caps(mdsc); 5570 ceph_flush_cap_releases(mdsc); 5571 5572 spin_lock(&mdsc->cap_dirty_lock); 5573 if (!list_empty(&mdsc->cap_flush_list)) { 5574 struct ceph_cap_flush *cf = 5575 list_last_entry(&mdsc->cap_flush_list, 5576 struct ceph_cap_flush, g_list); 5577 u64 want_flush = mdsc->last_cap_flush_tid; 5578 long drain_ret; 5579 5580 /* 5581 * Setting wake on the last entry is sufficient: flush 5582 * entries complete in order, so when this entry finishes 5583 * all earlier ones are already done. 5584 */ 5585 cf->wake = true; 5586 spin_unlock(&mdsc->cap_dirty_lock); 5587 pr_info_client(cl, 5588 "draining (want_flush=%llu, %d sessions)\n", 5589 want_flush, n); 5590 drain_ret = wait_event_timeout(mdsc->cap_flushing_wq, 5591 check_caps_flush(mdsc, 5592 want_flush), 5593 max_t(long, 5594 drain_deadline - jiffies, 5595 1)); 5596 if (drain_ret == 0) { 5597 pr_info_client(cl, 5598 "drain timed out, proceeding with forced teardown\n"); 5599 WRITE_ONCE(st->drain_timed_out, true); 5600 } else { 5601 pr_info_client(cl, "drain completed successfully\n"); 5602 } 5603 } else { 5604 spin_unlock(&mdsc->cap_dirty_lock); 5605 } 5606 5607 spin_lock(&st->lock); 5608 if (st->shutdown) { 5609 spin_unlock(&st->lock); 5610 goto out_sessions; 5611 } 5612 st->phase = CEPH_CLIENT_RESET_TEARDOWN; 5613 spin_unlock(&st->lock); 5614 5615 /* 5616 * Ask each MDS to close the session before we tear it down 5617 * locally. Without this the MDS sees only a connection drop and 5618 * waits for the client to reconnect (up to session_autoclose 5619 * seconds) before evicting the session and releasing locks. 5620 * 5621 * Reuse the normal close machinery so the session state/sequence 5622 * snapshot is serialized under s_mutex and a racing s_seq bump 5623 * retransmits REQUEST_CLOSE while the session remains CLOSING. 5624 * We send all close requests first, then yield briefly to let the 5625 * network stack transmit them before __unregister_session() 5626 * closes the connections. 5627 */ 5628 for (i = 0; i < n; i++) { 5629 int err; 5630 5631 mutex_lock(&sessions[i]->s_mutex); 5632 err = __close_session(mdsc, sessions[i]); 5633 mutex_unlock(&sessions[i]->s_mutex); 5634 if (err < 0) 5635 pr_warn_client(cl, 5636 "mds%d failed to queue close request before reset: %d\n", 5637 sessions[i]->s_mds, err); 5638 } 5639 /* 5640 * Best-effort grace period: yield briefly so the network stack 5641 * can transmit the queued REQUEST_CLOSE messages before we tear 5642 * down connections. Not a correctness requirement -- the MDS 5643 * will still evict via session_autoclose if it never receives 5644 * the close request. 5645 * 5646 * Event-based waiting is not viable here: there is no completion 5647 * event for "message left the NIC," and waiting for the MDS 5648 * SESSION_CLOSE response would re-create the stalemate that the 5649 * reset is meant to break. 5650 */ 5651 if (n > 0) 5652 msleep(CEPH_CLIENT_RESET_CLOSE_GRACE_MS); 5653 5654 /* 5655 * Tear down each session: close the connection, remove all 5656 * caps, clean up requests, then kick pending requests so they 5657 * re-open a fresh session on the next attempt. 5658 * 5659 * This is modeled on the check_new_map() forced-close path 5660 * for stopped MDS ranks - a proven pattern for hard session 5661 * teardown. We do NOT attempt send_mds_reconnect() because 5662 * the MDS only accepts reconnects during its own RECONNECT 5663 * phase (after MDS restart), not from an active client. 5664 * 5665 * Any state that did not drain (caps that didn't flush, unsafe 5666 * requests that the MDS didn't journal) is force-dropped here. 5667 * This is intentional: that state is stuck and is the reason 5668 * the operator triggered the reset. 5669 */ 5670 for (i = 0; i < n; i++) { 5671 int mds = sessions[i]->s_mds; 5672 5673 pr_info_client(cl, "mds%d resetting session\n", mds); 5674 5675 mutex_lock(&mdsc->mutex); 5676 if (mds >= mdsc->max_sessions || 5677 mdsc->sessions[mds] != sessions[i]) { 5678 pr_info_client(cl, 5679 "mds%d session already torn down, skipping\n", 5680 mds); 5681 mutex_unlock(&mdsc->mutex); 5682 ceph_put_mds_session(sessions[i]); 5683 sessions[i] = NULL; 5684 continue; 5685 } 5686 sessions[i]->s_state = CEPH_MDS_SESSION_CLOSED; 5687 __unregister_session(mdsc, sessions[i]); 5688 __wake_requests(mdsc, &sessions[i]->s_waiting); 5689 mutex_unlock(&mdsc->mutex); 5690 5691 mutex_lock(&sessions[i]->s_mutex); 5692 cleanup_session_requests(mdsc, sessions[i]); 5693 remove_session_caps(sessions[i]); 5694 mutex_unlock(&sessions[i]->s_mutex); 5695 5696 wake_up_all(&mdsc->session_close_wq); 5697 5698 ceph_put_mds_session(sessions[i]); 5699 5700 mutex_lock(&mdsc->mutex); 5701 kick_requests(mdsc, mds); 5702 mutex_unlock(&mdsc->mutex); 5703 5704 torn_down++; 5705 pr_info_client(cl, "mds%d session reset complete\n", mds); 5706 } 5707 5708 kfree(sessions); 5709 5710 spin_lock(&st->lock); 5711 st->sessions_reset = torn_down; 5712 spin_unlock(&st->lock); 5713 5714 out_complete: 5715 ceph_mdsc_reset_complete(mdsc, ret); 5716 return; 5717 5718 out_sessions: 5719 /* shutdown == true: ceph_mdsc_destroy() owns the final transition. */ 5720 for (i = 0; i < n; i++) 5721 ceph_put_mds_session(sessions[i]); 5722 kfree(sessions); 5723 } 5724 5725 int ceph_mdsc_schedule_reset(struct ceph_mds_client *mdsc, 5726 const char *reason) 5727 { 5728 struct ceph_client_reset_state *st = &mdsc->reset_state; 5729 struct ceph_fs_client *fsc = mdsc->fsc; 5730 const char *msg = (reason && reason[0]) ? reason : "manual"; 5731 int mount_state; 5732 5733 mount_state = READ_ONCE(fsc->mount_state); 5734 if (mount_state != CEPH_MOUNT_MOUNTED) { 5735 pr_warn_client(fsc->client, 5736 "reset rejected: mount_state=%d (not mounted)\n", 5737 mount_state); 5738 return -EINVAL; 5739 } 5740 5741 spin_lock(&st->lock); 5742 if (st->phase != CEPH_CLIENT_RESET_IDLE) { 5743 spin_unlock(&st->lock); 5744 return -EBUSY; 5745 } 5746 5747 st->phase = CEPH_CLIENT_RESET_QUIESCING; 5748 st->last_start = jiffies; 5749 st->last_errno = 0; 5750 st->drain_timed_out = false; 5751 st->sessions_reset = 0; 5752 st->trigger_count++; 5753 strscpy(st->last_reason, msg, sizeof(st->last_reason)); 5754 spin_unlock(&st->lock); 5755 5756 if (WARN_ON_ONCE(!queue_work(system_unbound_wq, &mdsc->reset_work))) { 5757 spin_lock(&st->lock); 5758 st->phase = CEPH_CLIENT_RESET_IDLE; 5759 st->last_errno = -EALREADY; 5760 st->last_finish = jiffies; 5761 st->failure_count++; 5762 spin_unlock(&st->lock); 5763 wake_up_all(&st->blocked_wq); 5764 return -EALREADY; 5765 } 5766 5767 pr_info_client(mdsc->fsc->client, 5768 "manual session reset scheduled (reason=\"%s\")\n", 5769 msg); 5770 trace_ceph_client_reset_schedule(mdsc, msg); 5771 return 0; 5772 } 5773 5774 5775 /* 5776 * compare old and new mdsmaps, kicking requests 5777 * and closing out old connections as necessary 5778 * 5779 * called under mdsc->mutex. 5780 */ 5781 static void check_new_map(struct ceph_mds_client *mdsc, 5782 struct ceph_mdsmap *newmap, 5783 struct ceph_mdsmap *oldmap) 5784 { 5785 int i, j, err; 5786 int oldstate, newstate; 5787 struct ceph_mds_session *s; 5788 unsigned long targets[DIV_ROUND_UP(CEPH_MAX_MDS, sizeof(unsigned long))] = {0}; 5789 struct ceph_client *cl = mdsc->fsc->client; 5790 5791 doutc(cl, "new %u old %u\n", newmap->m_epoch, oldmap->m_epoch); 5792 5793 if (newmap->m_info) { 5794 for (i = 0; i < newmap->possible_max_rank; i++) { 5795 for (j = 0; j < newmap->m_info[i].num_export_targets; j++) 5796 set_bit(newmap->m_info[i].export_targets[j], targets); 5797 } 5798 } 5799 5800 for (i = 0; i < oldmap->possible_max_rank && i < mdsc->max_sessions; i++) { 5801 if (!mdsc->sessions[i]) 5802 continue; 5803 s = mdsc->sessions[i]; 5804 oldstate = ceph_mdsmap_get_state(oldmap, i); 5805 newstate = ceph_mdsmap_get_state(newmap, i); 5806 5807 doutc(cl, "mds%d state %s%s -> %s%s (session %s)\n", 5808 i, ceph_mds_state_name(oldstate), 5809 ceph_mdsmap_is_laggy(oldmap, i) ? " (laggy)" : "", 5810 ceph_mds_state_name(newstate), 5811 ceph_mdsmap_is_laggy(newmap, i) ? " (laggy)" : "", 5812 ceph_session_state_name(s->s_state)); 5813 5814 if (i >= newmap->possible_max_rank) { 5815 /* force close session for stopped mds */ 5816 ceph_get_mds_session(s); 5817 __unregister_session(mdsc, s); 5818 __wake_requests(mdsc, &s->s_waiting); 5819 mutex_unlock(&mdsc->mutex); 5820 5821 mutex_lock(&s->s_mutex); 5822 cleanup_session_requests(mdsc, s); 5823 remove_session_caps(s); 5824 mutex_unlock(&s->s_mutex); 5825 5826 ceph_put_mds_session(s); 5827 5828 mutex_lock(&mdsc->mutex); 5829 kick_requests(mdsc, i); 5830 continue; 5831 } 5832 5833 if (memcmp(ceph_mdsmap_get_addr(oldmap, i), 5834 ceph_mdsmap_get_addr(newmap, i), 5835 sizeof(struct ceph_entity_addr))) { 5836 /* just close it */ 5837 mutex_unlock(&mdsc->mutex); 5838 mutex_lock(&s->s_mutex); 5839 mutex_lock(&mdsc->mutex); 5840 ceph_con_close(&s->s_con); 5841 mutex_unlock(&s->s_mutex); 5842 s->s_state = CEPH_MDS_SESSION_RESTARTING; 5843 } else if (oldstate == newstate) { 5844 continue; /* nothing new with this mds */ 5845 } 5846 5847 /* 5848 * send reconnect? 5849 */ 5850 if (s->s_state == CEPH_MDS_SESSION_RESTARTING && 5851 newstate >= CEPH_MDS_STATE_RECONNECT) { 5852 int rc; 5853 5854 mutex_unlock(&mdsc->mutex); 5855 clear_bit(i, targets); 5856 rc = send_mds_reconnect(mdsc, s); 5857 if (rc) 5858 pr_warn_client(cl, 5859 "mds%d reconnect failed: %d\n", 5860 i, rc); 5861 mutex_lock(&mdsc->mutex); 5862 } 5863 5864 /* 5865 * kick request on any mds that has gone active. 5866 */ 5867 if (oldstate < CEPH_MDS_STATE_ACTIVE && 5868 newstate >= CEPH_MDS_STATE_ACTIVE) { 5869 if (oldstate != CEPH_MDS_STATE_CREATING && 5870 oldstate != CEPH_MDS_STATE_STARTING) 5871 pr_info_client(cl, "mds%d recovery completed\n", 5872 s->s_mds); 5873 kick_requests(mdsc, i); 5874 mutex_unlock(&mdsc->mutex); 5875 mutex_lock(&s->s_mutex); 5876 mutex_lock(&mdsc->mutex); 5877 ceph_kick_flushing_caps(mdsc, s); 5878 mutex_unlock(&s->s_mutex); 5879 wake_up_session_caps(s, RECONNECT); 5880 } 5881 } 5882 5883 /* 5884 * Only open and reconnect sessions that don't exist yet. 5885 */ 5886 for (i = 0; i < newmap->possible_max_rank; i++) { 5887 /* 5888 * In case the import MDS is crashed just after 5889 * the EImportStart journal is flushed, so when 5890 * a standby MDS takes over it and is replaying 5891 * the EImportStart journal the new MDS daemon 5892 * will wait the client to reconnect it, but the 5893 * client may never register/open the session yet. 5894 * 5895 * Will try to reconnect that MDS daemon if the 5896 * rank number is in the export targets array and 5897 * is the up:reconnect state. 5898 */ 5899 newstate = ceph_mdsmap_get_state(newmap, i); 5900 if (!test_bit(i, targets) || newstate != CEPH_MDS_STATE_RECONNECT) 5901 continue; 5902 5903 /* 5904 * The session maybe registered and opened by some 5905 * requests which were choosing random MDSes during 5906 * the mdsc->mutex's unlock/lock gap below in rare 5907 * case. But the related MDS daemon will just queue 5908 * that requests and be still waiting for the client's 5909 * reconnection request in up:reconnect state. 5910 */ 5911 s = __ceph_lookup_mds_session(mdsc, i); 5912 if (likely(!s)) { 5913 s = __open_export_target_session(mdsc, i); 5914 if (IS_ERR(s)) { 5915 err = PTR_ERR(s); 5916 pr_err_client(cl, 5917 "failed to open export target session, err %d\n", 5918 err); 5919 continue; 5920 } 5921 } 5922 doutc(cl, "send reconnect to export target mds.%d\n", i); 5923 mutex_unlock(&mdsc->mutex); 5924 err = send_mds_reconnect(mdsc, s); 5925 if (err) 5926 pr_warn_client(cl, 5927 "mds%d export target reconnect failed: %d\n", 5928 i, err); 5929 ceph_put_mds_session(s); 5930 mutex_lock(&mdsc->mutex); 5931 } 5932 5933 for (i = 0; i < newmap->possible_max_rank && i < mdsc->max_sessions; i++) { 5934 s = mdsc->sessions[i]; 5935 if (!s) 5936 continue; 5937 if (!ceph_mdsmap_is_laggy(newmap, i)) 5938 continue; 5939 if (s->s_state == CEPH_MDS_SESSION_OPEN || 5940 s->s_state == CEPH_MDS_SESSION_HUNG || 5941 s->s_state == CEPH_MDS_SESSION_CLOSING) { 5942 doutc(cl, " connecting to export targets of laggy mds%d\n", i); 5943 __open_export_target_sessions(mdsc, s); 5944 } 5945 } 5946 } 5947 5948 5949 5950 /* 5951 * leases 5952 */ 5953 5954 /* 5955 * caller must hold session s_mutex, dentry->d_lock 5956 */ 5957 void __ceph_mdsc_drop_dentry_lease(struct dentry *dentry) 5958 { 5959 struct ceph_dentry_info *di = ceph_dentry(dentry); 5960 5961 ceph_put_mds_session(di->lease_session); 5962 di->lease_session = NULL; 5963 } 5964 5965 static void handle_lease(struct ceph_mds_client *mdsc, 5966 struct ceph_mds_session *session, 5967 struct ceph_msg *msg) 5968 { 5969 struct ceph_client *cl = mdsc->fsc->client; 5970 struct super_block *sb = mdsc->fsc->sb; 5971 struct inode *inode; 5972 struct dentry *parent, *dentry; 5973 struct ceph_dentry_info *di; 5974 int mds = session->s_mds; 5975 struct ceph_mds_lease *h = msg->front.iov_base; 5976 u32 seq; 5977 struct ceph_vino vino; 5978 struct qstr dname; 5979 int release = 0; 5980 5981 doutc(cl, "from mds%d\n", mds); 5982 5983 if (!ceph_inc_mds_stopping_blocker(mdsc, session)) 5984 return; 5985 5986 /* decode */ 5987 if (msg->front.iov_len < sizeof(*h) + sizeof(u32)) 5988 goto bad; 5989 vino.ino = le64_to_cpu(h->ino); 5990 vino.snap = CEPH_NOSNAP; 5991 seq = le32_to_cpu(h->seq); 5992 dname.len = get_unaligned_le32(h + 1); 5993 if (msg->front.iov_len < sizeof(*h) + sizeof(u32) + dname.len) 5994 goto bad; 5995 dname.name = (void *)(h + 1) + sizeof(u32); 5996 5997 /* lookup inode */ 5998 inode = ceph_find_inode(sb, vino); 5999 doutc(cl, "%s, ino %llx %p %.*s\n", ceph_lease_op_name(h->action), 6000 vino.ino, inode, dname.len, dname.name); 6001 6002 mutex_lock(&session->s_mutex); 6003 if (!inode) { 6004 doutc(cl, "no inode %llx\n", vino.ino); 6005 goto release; 6006 } 6007 6008 /* dentry */ 6009 parent = d_find_alias(inode); 6010 if (!parent) { 6011 doutc(cl, "no parent dentry on inode %p\n", inode); 6012 WARN_ON(1); 6013 goto release; /* hrm... */ 6014 } 6015 dname.hash = full_name_hash(parent, dname.name, dname.len); 6016 dentry = d_lookup(parent, &dname); 6017 dput(parent); 6018 if (!dentry) 6019 goto release; 6020 6021 spin_lock(&dentry->d_lock); 6022 di = ceph_dentry(dentry); 6023 switch (h->action) { 6024 case CEPH_MDS_LEASE_REVOKE: 6025 if (di->lease_session == session) { 6026 if (ceph_seq_cmp(di->lease_seq, seq) > 0) 6027 h->seq = cpu_to_le32(di->lease_seq); 6028 __ceph_mdsc_drop_dentry_lease(dentry); 6029 } 6030 release = 1; 6031 break; 6032 6033 case CEPH_MDS_LEASE_RENEW: 6034 if (di->lease_session == session && 6035 di->lease_gen == atomic_read(&session->s_cap_gen) && 6036 di->lease_renew_from && 6037 di->lease_renew_after == 0) { 6038 unsigned long duration = 6039 msecs_to_jiffies(le32_to_cpu(h->duration_ms)); 6040 6041 di->lease_seq = seq; 6042 di->time = di->lease_renew_from + duration; 6043 di->lease_renew_after = di->lease_renew_from + 6044 (duration >> 1); 6045 di->lease_renew_from = 0; 6046 } 6047 break; 6048 } 6049 spin_unlock(&dentry->d_lock); 6050 dput(dentry); 6051 6052 if (!release) 6053 goto out; 6054 6055 release: 6056 /* let's just reuse the same message */ 6057 h->action = CEPH_MDS_LEASE_REVOKE_ACK; 6058 ceph_msg_get(msg); 6059 ceph_con_send(&session->s_con, msg); 6060 6061 out: 6062 mutex_unlock(&session->s_mutex); 6063 iput(inode); 6064 6065 ceph_dec_mds_stopping_blocker(mdsc); 6066 return; 6067 6068 bad: 6069 ceph_dec_mds_stopping_blocker(mdsc); 6070 6071 pr_err_client(cl, "corrupt lease message\n"); 6072 ceph_msg_dump(msg); 6073 } 6074 6075 void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session, 6076 struct dentry *dentry, char action, 6077 u32 seq) 6078 { 6079 struct ceph_client *cl = session->s_mdsc->fsc->client; 6080 struct ceph_msg *msg; 6081 struct ceph_mds_lease *lease; 6082 struct inode *dir; 6083 int len = sizeof(*lease) + sizeof(u32) + NAME_MAX; 6084 6085 doutc(cl, "identry %p %s to mds%d\n", dentry, ceph_lease_op_name(action), 6086 session->s_mds); 6087 6088 msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS, false); 6089 if (!msg) 6090 return; 6091 lease = msg->front.iov_base; 6092 lease->action = action; 6093 lease->seq = cpu_to_le32(seq); 6094 6095 spin_lock(&dentry->d_lock); 6096 dir = d_inode(dentry->d_parent); 6097 lease->ino = cpu_to_le64(ceph_ino(dir)); 6098 lease->first = lease->last = cpu_to_le64(ceph_snap(dir)); 6099 6100 put_unaligned_le32(dentry->d_name.len, lease + 1); 6101 memcpy((void *)(lease + 1) + 4, 6102 dentry->d_name.name, dentry->d_name.len); 6103 spin_unlock(&dentry->d_lock); 6104 6105 ceph_con_send(&session->s_con, msg); 6106 } 6107 6108 /* 6109 * lock unlock the session, to wait ongoing session activities 6110 */ 6111 static void lock_unlock_session(struct ceph_mds_session *s) 6112 { 6113 mutex_lock(&s->s_mutex); 6114 mutex_unlock(&s->s_mutex); 6115 } 6116 6117 static void maybe_recover_session(struct ceph_mds_client *mdsc) 6118 { 6119 struct ceph_client *cl = mdsc->fsc->client; 6120 struct ceph_fs_client *fsc = mdsc->fsc; 6121 6122 if (!ceph_test_mount_opt(fsc, CLEANRECOVER)) 6123 return; 6124 6125 if (READ_ONCE(fsc->mount_state) != CEPH_MOUNT_MOUNTED) 6126 return; 6127 6128 if (!READ_ONCE(fsc->blocklisted)) 6129 return; 6130 6131 pr_info_client(cl, "auto reconnect after blocklisted\n"); 6132 ceph_force_reconnect(fsc->sb); 6133 } 6134 6135 bool check_session_state(struct ceph_mds_session *s) 6136 { 6137 struct ceph_client *cl = s->s_mdsc->fsc->client; 6138 6139 switch (s->s_state) { 6140 case CEPH_MDS_SESSION_OPEN: 6141 if (s->s_ttl && time_after(jiffies, s->s_ttl)) { 6142 s->s_state = CEPH_MDS_SESSION_HUNG; 6143 pr_info_client(cl, "mds%d hung\n", s->s_mds); 6144 } 6145 break; 6146 case CEPH_MDS_SESSION_CLOSING: 6147 case CEPH_MDS_SESSION_NEW: 6148 case CEPH_MDS_SESSION_RESTARTING: 6149 case CEPH_MDS_SESSION_CLOSED: 6150 case CEPH_MDS_SESSION_REJECTED: 6151 return false; 6152 } 6153 6154 return true; 6155 } 6156 6157 /* 6158 * If the sequence is incremented while we're waiting on a REQUEST_CLOSE reply, 6159 * then we need to retransmit that request. 6160 */ 6161 void inc_session_sequence(struct ceph_mds_session *s) 6162 { 6163 struct ceph_client *cl = s->s_mdsc->fsc->client; 6164 6165 lockdep_assert_held(&s->s_mutex); 6166 6167 s->s_seq++; 6168 6169 if (s->s_state == CEPH_MDS_SESSION_CLOSING) { 6170 int ret; 6171 6172 doutc(cl, "resending session close request for mds%d\n", s->s_mds); 6173 ret = request_close_session(s); 6174 if (ret < 0) 6175 pr_err_client(cl, "unable to close session to mds%d: %d\n", 6176 s->s_mds, ret); 6177 } 6178 } 6179 6180 /* 6181 * delayed work -- periodically trim expired leases, renew caps with mds. If 6182 * the @delay parameter is set to 0 or if it's more than 5 secs, the default 6183 * workqueue delay value of 5 secs will be used. 6184 */ 6185 static void schedule_delayed(struct ceph_mds_client *mdsc, unsigned long delay) 6186 { 6187 unsigned long max_delay = HZ * 5; 6188 6189 /* 5 secs default delay */ 6190 if (!delay || (delay > max_delay)) 6191 delay = max_delay; 6192 schedule_delayed_work(&mdsc->delayed_work, 6193 round_jiffies_relative(delay)); 6194 } 6195 6196 static void delayed_work(struct work_struct *work) 6197 { 6198 struct ceph_mds_client *mdsc = 6199 container_of(work, struct ceph_mds_client, delayed_work.work); 6200 unsigned long delay; 6201 int renew_interval; 6202 int renew_caps; 6203 int i; 6204 6205 doutc(mdsc->fsc->client, "mdsc delayed_work\n"); 6206 6207 if (mdsc->stopping >= CEPH_MDSC_STOPPING_FLUSHED) 6208 return; 6209 6210 mutex_lock(&mdsc->mutex); 6211 renew_interval = mdsc->mdsmap->m_session_timeout >> 2; 6212 renew_caps = time_after_eq(jiffies, HZ*renew_interval + 6213 mdsc->last_renew_caps); 6214 if (renew_caps) 6215 mdsc->last_renew_caps = jiffies; 6216 6217 for (i = 0; i < mdsc->max_sessions; i++) { 6218 struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i); 6219 if (!s) 6220 continue; 6221 6222 if (!check_session_state(s)) { 6223 ceph_put_mds_session(s); 6224 continue; 6225 } 6226 mutex_unlock(&mdsc->mutex); 6227 6228 ceph_flush_session_cap_releases(mdsc, s); 6229 6230 mutex_lock(&s->s_mutex); 6231 if (renew_caps) 6232 send_renew_caps(mdsc, s); 6233 else 6234 ceph_con_keepalive(&s->s_con); 6235 if (s->s_state == CEPH_MDS_SESSION_OPEN || 6236 s->s_state == CEPH_MDS_SESSION_HUNG) 6237 ceph_send_cap_releases(mdsc, s); 6238 mutex_unlock(&s->s_mutex); 6239 ceph_put_mds_session(s); 6240 6241 mutex_lock(&mdsc->mutex); 6242 } 6243 mutex_unlock(&mdsc->mutex); 6244 6245 delay = ceph_check_delayed_caps(mdsc); 6246 6247 ceph_queue_cap_reclaim_work(mdsc); 6248 6249 ceph_trim_snapid_map(mdsc); 6250 6251 maybe_recover_session(mdsc); 6252 6253 schedule_delayed(mdsc, delay); 6254 } 6255 6256 int ceph_mdsc_init(struct ceph_fs_client *fsc) 6257 6258 { 6259 struct ceph_mds_client *mdsc; 6260 int err; 6261 6262 mdsc = kzalloc_obj(struct ceph_mds_client, GFP_NOFS); 6263 if (!mdsc) 6264 return -ENOMEM; 6265 mdsc->fsc = fsc; 6266 mutex_init(&mdsc->mutex); 6267 mdsc->mdsmap = kzalloc_obj(*mdsc->mdsmap, GFP_NOFS); 6268 if (!mdsc->mdsmap) { 6269 err = -ENOMEM; 6270 goto err_mdsc; 6271 } 6272 6273 init_completion(&mdsc->safe_umount_waiters); 6274 spin_lock_init(&mdsc->stopping_lock); 6275 atomic_set(&mdsc->stopping_blockers, 0); 6276 init_completion(&mdsc->stopping_waiter); 6277 atomic64_set(&mdsc->dirty_folios, 0); 6278 init_waitqueue_head(&mdsc->flush_end_wq); 6279 init_waitqueue_head(&mdsc->session_close_wq); 6280 INIT_LIST_HEAD(&mdsc->waiting_for_map); 6281 mdsc->quotarealms_inodes = RB_ROOT; 6282 mutex_init(&mdsc->quotarealms_inodes_mutex); 6283 init_rwsem(&mdsc->snap_rwsem); 6284 mdsc->snap_realms = RB_ROOT; 6285 INIT_LIST_HEAD(&mdsc->snap_empty); 6286 spin_lock_init(&mdsc->snap_empty_lock); 6287 mdsc->request_tree = RB_ROOT; 6288 INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work); 6289 mdsc->last_renew_caps = jiffies; 6290 INIT_LIST_HEAD(&mdsc->cap_delay_list); 6291 #ifdef CONFIG_DEBUG_FS 6292 INIT_LIST_HEAD(&mdsc->cap_wait_list); 6293 #endif 6294 spin_lock_init(&mdsc->cap_delay_lock); 6295 INIT_LIST_HEAD(&mdsc->cap_unlink_delay_list); 6296 INIT_LIST_HEAD(&mdsc->snap_flush_list); 6297 spin_lock_init(&mdsc->snap_flush_lock); 6298 mdsc->last_cap_flush_tid = 1; 6299 INIT_LIST_HEAD(&mdsc->cap_flush_list); 6300 INIT_LIST_HEAD(&mdsc->cap_dirty_migrating); 6301 spin_lock_init(&mdsc->cap_dirty_lock); 6302 init_waitqueue_head(&mdsc->cap_flushing_wq); 6303 INIT_WORK(&mdsc->cap_reclaim_work, ceph_cap_reclaim_work); 6304 INIT_WORK(&mdsc->cap_unlink_work, ceph_cap_unlink_work); 6305 err = ceph_metric_init(&mdsc->metric); 6306 if (err) 6307 goto err_mdsmap; 6308 ceph_subvolume_metrics_init(&mdsc->subvol_metrics); 6309 mutex_init(&mdsc->subvol_metrics_last_mutex); 6310 mdsc->subvol_metrics_last = NULL; 6311 mdsc->subvol_metrics_last_nr = 0; 6312 mdsc->subvol_metrics_sent = 0; 6313 mdsc->subvol_metrics_nonzero_sends = 0; 6314 6315 spin_lock_init(&mdsc->dentry_list_lock); 6316 INIT_LIST_HEAD(&mdsc->dentry_leases); 6317 INIT_LIST_HEAD(&mdsc->dentry_dir_leases); 6318 6319 spin_lock_init(&mdsc->reset_state.lock); 6320 init_waitqueue_head(&mdsc->reset_state.blocked_wq); 6321 atomic_set(&mdsc->reset_state.blocked_requests, 0); 6322 INIT_WORK(&mdsc->reset_work, ceph_mdsc_reset_workfn); 6323 6324 ceph_caps_init(mdsc); 6325 ceph_adjust_caps_max_min(mdsc, fsc->mount_options); 6326 6327 spin_lock_init(&mdsc->snapid_map_lock); 6328 mdsc->snapid_map_tree = RB_ROOT; 6329 INIT_LIST_HEAD(&mdsc->snapid_map_lru); 6330 6331 init_rwsem(&mdsc->pool_perm_rwsem); 6332 mdsc->pool_perm_tree = RB_ROOT; 6333 6334 strscpy(mdsc->nodename, utsname()->nodename, 6335 sizeof(mdsc->nodename)); 6336 6337 fsc->mdsc = mdsc; 6338 return 0; 6339 6340 err_mdsmap: 6341 kfree(mdsc->mdsmap); 6342 err_mdsc: 6343 kfree(mdsc); 6344 return err; 6345 } 6346 6347 /* 6348 * Wait for safe replies on open mds requests. If we time out, drop 6349 * all requests from the tree to avoid dangling dentry refs. 6350 */ 6351 static void wait_requests(struct ceph_mds_client *mdsc) 6352 { 6353 struct ceph_client *cl = mdsc->fsc->client; 6354 struct ceph_options *opts = mdsc->fsc->client->options; 6355 struct ceph_mds_request *req; 6356 6357 mutex_lock(&mdsc->mutex); 6358 if (__get_oldest_req(mdsc)) { 6359 mutex_unlock(&mdsc->mutex); 6360 6361 doutc(cl, "waiting for requests\n"); 6362 wait_for_completion_timeout(&mdsc->safe_umount_waiters, 6363 ceph_timeout_jiffies(opts->mount_timeout)); 6364 6365 /* tear down remaining requests */ 6366 mutex_lock(&mdsc->mutex); 6367 while ((req = __get_oldest_req(mdsc))) { 6368 doutc(cl, "timed out on tid %llu\n", req->r_tid); 6369 list_del_init(&req->r_wait); 6370 __unregister_request(mdsc, req); 6371 } 6372 } 6373 mutex_unlock(&mdsc->mutex); 6374 doutc(cl, "done\n"); 6375 } 6376 6377 void send_flush_mdlog(struct ceph_mds_session *s) 6378 { 6379 struct ceph_client *cl = s->s_mdsc->fsc->client; 6380 struct ceph_msg *msg; 6381 6382 /* 6383 * Pre-luminous MDS crashes when it sees an unknown session request 6384 */ 6385 if (!CEPH_HAVE_FEATURE(s->s_con.peer_features, SERVER_LUMINOUS)) 6386 return; 6387 6388 mutex_lock(&s->s_mutex); 6389 doutc(cl, "request mdlog flush to mds%d (%s)s seq %lld\n", 6390 s->s_mds, ceph_session_state_name(s->s_state), s->s_seq); 6391 msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_FLUSH_MDLOG, 6392 s->s_seq); 6393 if (!msg) { 6394 pr_err_client(cl, "failed to request mdlog flush to mds%d (%s) seq %lld\n", 6395 s->s_mds, ceph_session_state_name(s->s_state), s->s_seq); 6396 } else { 6397 ceph_con_send(&s->s_con, msg); 6398 } 6399 mutex_unlock(&s->s_mutex); 6400 } 6401 6402 static int ceph_mds_auth_match(struct ceph_mds_client *mdsc, 6403 struct ceph_mds_cap_auth *auth, 6404 const struct cred *cred, 6405 char *tpath) 6406 { 6407 u32 caller_uid = from_kuid(&init_user_ns, cred->fsuid); 6408 u32 caller_gid = from_kgid(&init_user_ns, cred->fsgid); 6409 struct ceph_client *cl = mdsc->fsc->client; 6410 const char *fs_name = mdsc->mdsmap->m_fs_name; 6411 const char *spath = mdsc->fsc->mount_options->server_path; 6412 bool gid_matched = false; 6413 u32 gid, tlen, len; 6414 int i, j; 6415 6416 doutc(cl, "fsname check fs_name=%s match.fs_name=%s\n", 6417 fs_name, auth->match.fs_name ? auth->match.fs_name : ""); 6418 6419 if (!ceph_namespace_match(auth->match.fs_name, fs_name)) { 6420 /* fsname mismatch, try next one */ 6421 return 0; 6422 } 6423 6424 doutc(cl, "match.uid %lld\n", auth->match.uid); 6425 if (auth->match.uid != MDS_AUTH_UID_ANY) { 6426 if (auth->match.uid != caller_uid) 6427 return 0; 6428 if (auth->match.num_gids) { 6429 for (i = 0; i < auth->match.num_gids; i++) { 6430 if (caller_gid == auth->match.gids[i]) 6431 gid_matched = true; 6432 } 6433 if (!gid_matched && cred->group_info->ngroups) { 6434 for (i = 0; i < cred->group_info->ngroups; i++) { 6435 gid = from_kgid(&init_user_ns, 6436 cred->group_info->gid[i]); 6437 for (j = 0; j < auth->match.num_gids; j++) { 6438 if (gid == auth->match.gids[j]) { 6439 gid_matched = true; 6440 break; 6441 } 6442 } 6443 if (gid_matched) 6444 break; 6445 } 6446 } 6447 if (!gid_matched) 6448 return 0; 6449 } 6450 } 6451 6452 /* path match */ 6453 if (auth->match.path) { 6454 if (!tpath) 6455 return 0; 6456 6457 tlen = strlen(tpath); 6458 len = strlen(auth->match.path); 6459 if (len) { 6460 char *_tpath = tpath; 6461 bool free_tpath = false; 6462 int m, n; 6463 6464 doutc(cl, "server path %s, tpath %s, match.path %s\n", 6465 spath, tpath, auth->match.path); 6466 if (spath && (m = strlen(spath)) != 1) { 6467 /* mount path + '/' + tpath + an extra space */ 6468 n = m + 1 + tlen + 1; 6469 _tpath = kmalloc(n, GFP_NOFS); 6470 if (!_tpath) 6471 return -ENOMEM; 6472 /* remove the leading '/' */ 6473 snprintf(_tpath, n, "%s/%s", spath + 1, tpath); 6474 free_tpath = true; 6475 tlen = strlen(_tpath); 6476 } 6477 6478 /* 6479 * Please note the tailing '/' for match.path has already 6480 * been removed when parsing. 6481 * 6482 * Remove the tailing '/' for the target path. 6483 */ 6484 while (tlen && _tpath[tlen - 1] == '/') { 6485 _tpath[tlen - 1] = '\0'; 6486 tlen -= 1; 6487 } 6488 doutc(cl, "_tpath %s\n", _tpath); 6489 6490 /* 6491 * In case first == _tpath && tlen == len: 6492 * match.path=/foo --> /foo _path=/foo --> match 6493 * match.path=/foo/ --> /foo _path=/foo --> match 6494 * 6495 * In case first == _tmatch.path && tlen > len: 6496 * match.path=/foo/ --> /foo _path=/foo/ --> match 6497 * match.path=/foo --> /foo _path=/foo/ --> match 6498 * match.path=/foo/ --> /foo _path=/foo/d --> match 6499 * match.path=/foo --> /foo _path=/food --> mismatch 6500 * 6501 * All the other cases --> mismatch 6502 */ 6503 bool path_matched = true; 6504 char *first = strstr(_tpath, auth->match.path); 6505 if (first != _tpath || 6506 (tlen > len && _tpath[len] != '/')) { 6507 path_matched = false; 6508 } 6509 6510 if (free_tpath) 6511 kfree(_tpath); 6512 6513 if (!path_matched) 6514 return 0; 6515 } 6516 } 6517 6518 doutc(cl, "matched\n"); 6519 return 1; 6520 } 6521 6522 int ceph_mds_check_access(struct ceph_mds_client *mdsc, char *tpath, int mask) 6523 { 6524 const struct cred *cred = get_current_cred(); 6525 u32 caller_uid = from_kuid(&init_user_ns, cred->fsuid); 6526 u32 caller_gid = from_kgid(&init_user_ns, cred->fsgid); 6527 struct ceph_mds_cap_auth *rw_perms_s = NULL; 6528 struct ceph_client *cl = mdsc->fsc->client; 6529 bool root_squash_perms = true; 6530 int i, err; 6531 6532 doutc(cl, "tpath '%s', mask %d, caller_uid %d, caller_gid %d\n", 6533 tpath, mask, caller_uid, caller_gid); 6534 6535 for (i = 0; i < mdsc->s_cap_auths_num; i++) { 6536 struct ceph_mds_cap_auth *s = &mdsc->s_cap_auths[i]; 6537 6538 err = ceph_mds_auth_match(mdsc, s, cred, tpath); 6539 if (err < 0) { 6540 put_cred(cred); 6541 return err; 6542 } else if (err > 0) { 6543 /* always follow the last auth caps' permission */ 6544 root_squash_perms = true; 6545 rw_perms_s = NULL; 6546 if ((mask & MAY_WRITE) && s->writeable && 6547 s->match.root_squash && (!caller_uid || !caller_gid)) 6548 root_squash_perms = false; 6549 6550 if (((mask & MAY_WRITE) && !s->writeable) || 6551 ((mask & MAY_READ) && !s->readable)) 6552 rw_perms_s = s; 6553 } 6554 } 6555 6556 put_cred(cred); 6557 6558 doutc(cl, "root_squash_perms %d, rw_perms_s %p\n", root_squash_perms, 6559 rw_perms_s); 6560 if (root_squash_perms && rw_perms_s == NULL) { 6561 doutc(cl, "access allowed\n"); 6562 return 0; 6563 } 6564 6565 if (!root_squash_perms) { 6566 doutc(cl, "root_squash is enabled and user(%d %d) isn't allowed to write", 6567 caller_uid, caller_gid); 6568 } 6569 if (rw_perms_s) { 6570 doutc(cl, "mds auth caps readable/writeable %d/%d while request r/w %d/%d", 6571 rw_perms_s->readable, rw_perms_s->writeable, 6572 !!(mask & MAY_READ), !!(mask & MAY_WRITE)); 6573 } 6574 doutc(cl, "access denied\n"); 6575 return -EACCES; 6576 } 6577 6578 /* 6579 * called before mount is ro, and before dentries are torn down. 6580 * (hmm, does this still race with new lookups?) 6581 */ 6582 void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc) 6583 { 6584 doutc(mdsc->fsc->client, "begin\n"); 6585 mdsc->stopping = CEPH_MDSC_STOPPING_BEGIN; 6586 6587 ceph_mdsc_iterate_sessions(mdsc, send_flush_mdlog, true); 6588 ceph_mdsc_iterate_sessions(mdsc, lock_unlock_session, false); 6589 ceph_flush_dirty_caps(mdsc); 6590 wait_requests(mdsc); 6591 6592 /* 6593 * wait for reply handlers to drop their request refs and 6594 * their inode/dcache refs 6595 */ 6596 ceph_msgr_flush(); 6597 6598 ceph_cleanup_quotarealms_inodes(mdsc); 6599 doutc(mdsc->fsc->client, "done\n"); 6600 } 6601 6602 /* 6603 * flush the mdlog and wait for all write mds requests to flush. 6604 */ 6605 static void flush_mdlog_and_wait_mdsc_unsafe_requests(struct ceph_mds_client *mdsc, 6606 u64 want_tid) 6607 { 6608 struct ceph_client *cl = mdsc->fsc->client; 6609 struct ceph_mds_request *req = NULL, *nextreq; 6610 struct ceph_mds_session *last_session = NULL; 6611 struct rb_node *n; 6612 6613 mutex_lock(&mdsc->mutex); 6614 doutc(cl, "want %lld\n", want_tid); 6615 restart: 6616 req = __get_oldest_req(mdsc); 6617 while (req && req->r_tid <= want_tid) { 6618 /* find next request */ 6619 n = rb_next(&req->r_node); 6620 if (n) 6621 nextreq = rb_entry(n, struct ceph_mds_request, r_node); 6622 else 6623 nextreq = NULL; 6624 if (req->r_op != CEPH_MDS_OP_SETFILELOCK && 6625 (req->r_op & CEPH_MDS_OP_WRITE)) { 6626 struct ceph_mds_session *s = req->r_session; 6627 6628 if (!s) { 6629 req = nextreq; 6630 continue; 6631 } 6632 6633 /* write op */ 6634 ceph_mdsc_get_request(req); 6635 if (nextreq) 6636 ceph_mdsc_get_request(nextreq); 6637 s = ceph_get_mds_session(s); 6638 mutex_unlock(&mdsc->mutex); 6639 6640 /* send flush mdlog request to MDS */ 6641 if (last_session != s) { 6642 send_flush_mdlog(s); 6643 ceph_put_mds_session(last_session); 6644 last_session = s; 6645 } else { 6646 ceph_put_mds_session(s); 6647 } 6648 doutc(cl, "wait on %llu (want %llu)\n", 6649 req->r_tid, want_tid); 6650 wait_for_completion(&req->r_safe_completion); 6651 6652 mutex_lock(&mdsc->mutex); 6653 ceph_mdsc_put_request(req); 6654 if (!nextreq) 6655 break; /* next dne before, so we're done! */ 6656 if (RB_EMPTY_NODE(&nextreq->r_node)) { 6657 /* next request was removed from tree */ 6658 ceph_mdsc_put_request(nextreq); 6659 goto restart; 6660 } 6661 ceph_mdsc_put_request(nextreq); /* won't go away */ 6662 } 6663 req = nextreq; 6664 } 6665 mutex_unlock(&mdsc->mutex); 6666 ceph_put_mds_session(last_session); 6667 doutc(cl, "done\n"); 6668 } 6669 6670 void ceph_mdsc_sync(struct ceph_mds_client *mdsc) 6671 { 6672 struct ceph_client *cl = mdsc->fsc->client; 6673 u64 want_tid, want_flush; 6674 6675 if (READ_ONCE(mdsc->fsc->mount_state) >= CEPH_MOUNT_SHUTDOWN) 6676 return; 6677 6678 doutc(cl, "sync\n"); 6679 mutex_lock(&mdsc->mutex); 6680 want_tid = mdsc->last_tid; 6681 mutex_unlock(&mdsc->mutex); 6682 6683 ceph_flush_dirty_caps(mdsc); 6684 ceph_flush_cap_releases(mdsc); 6685 spin_lock(&mdsc->cap_dirty_lock); 6686 want_flush = mdsc->last_cap_flush_tid; 6687 if (!list_empty(&mdsc->cap_flush_list)) { 6688 struct ceph_cap_flush *cf = 6689 list_last_entry(&mdsc->cap_flush_list, 6690 struct ceph_cap_flush, g_list); 6691 cf->wake = true; 6692 } 6693 spin_unlock(&mdsc->cap_dirty_lock); 6694 6695 doutc(cl, "sync want tid %lld flush_seq %lld\n", want_tid, want_flush); 6696 6697 flush_mdlog_and_wait_mdsc_unsafe_requests(mdsc, want_tid); 6698 wait_caps_flush(mdsc, want_flush); 6699 } 6700 6701 /* 6702 * true if all sessions are closed, or we force unmount 6703 */ 6704 static bool done_closing_sessions(struct ceph_mds_client *mdsc, int skipped) 6705 { 6706 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) 6707 return true; 6708 return atomic_read(&mdsc->num_sessions) <= skipped; 6709 } 6710 6711 /* 6712 * called after sb is ro or when metadata corrupted. 6713 */ 6714 void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc) 6715 { 6716 struct ceph_options *opts = mdsc->fsc->client->options; 6717 struct ceph_client *cl = mdsc->fsc->client; 6718 struct ceph_mds_session *session; 6719 int i; 6720 int skipped = 0; 6721 6722 doutc(cl, "begin\n"); 6723 6724 /* close sessions */ 6725 mutex_lock(&mdsc->mutex); 6726 for (i = 0; i < mdsc->max_sessions; i++) { 6727 session = __ceph_lookup_mds_session(mdsc, i); 6728 if (!session) 6729 continue; 6730 mutex_unlock(&mdsc->mutex); 6731 mutex_lock(&session->s_mutex); 6732 if (__close_session(mdsc, session) <= 0) 6733 skipped++; 6734 mutex_unlock(&session->s_mutex); 6735 ceph_put_mds_session(session); 6736 mutex_lock(&mdsc->mutex); 6737 } 6738 mutex_unlock(&mdsc->mutex); 6739 6740 doutc(cl, "waiting for sessions to close\n"); 6741 wait_event_timeout(mdsc->session_close_wq, 6742 done_closing_sessions(mdsc, skipped), 6743 ceph_timeout_jiffies(opts->mount_timeout)); 6744 6745 /* tear down remaining sessions */ 6746 mutex_lock(&mdsc->mutex); 6747 for (i = 0; i < mdsc->max_sessions; i++) { 6748 if (mdsc->sessions[i]) { 6749 session = ceph_get_mds_session(mdsc->sessions[i]); 6750 __unregister_session(mdsc, session); 6751 mutex_unlock(&mdsc->mutex); 6752 mutex_lock(&session->s_mutex); 6753 remove_session_caps(session); 6754 mutex_unlock(&session->s_mutex); 6755 ceph_put_mds_session(session); 6756 mutex_lock(&mdsc->mutex); 6757 } 6758 } 6759 WARN_ON(!list_empty(&mdsc->cap_delay_list)); 6760 mutex_unlock(&mdsc->mutex); 6761 6762 ceph_cleanup_snapid_map(mdsc); 6763 ceph_cleanup_global_and_empty_realms(mdsc); 6764 6765 cancel_work_sync(&mdsc->cap_reclaim_work); 6766 cancel_work_sync(&mdsc->cap_unlink_work); 6767 cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */ 6768 6769 doutc(cl, "done\n"); 6770 } 6771 6772 void ceph_mdsc_force_umount(struct ceph_mds_client *mdsc) 6773 { 6774 struct ceph_mds_session *session; 6775 int mds; 6776 6777 doutc(mdsc->fsc->client, "force umount\n"); 6778 6779 mutex_lock(&mdsc->mutex); 6780 for (mds = 0; mds < mdsc->max_sessions; mds++) { 6781 session = __ceph_lookup_mds_session(mdsc, mds); 6782 if (!session) 6783 continue; 6784 6785 if (session->s_state == CEPH_MDS_SESSION_REJECTED) 6786 __unregister_session(mdsc, session); 6787 __wake_requests(mdsc, &session->s_waiting); 6788 mutex_unlock(&mdsc->mutex); 6789 6790 mutex_lock(&session->s_mutex); 6791 __close_session(mdsc, session); 6792 if (session->s_state == CEPH_MDS_SESSION_CLOSING) { 6793 cleanup_session_requests(mdsc, session); 6794 remove_session_caps(session); 6795 } 6796 mutex_unlock(&session->s_mutex); 6797 ceph_put_mds_session(session); 6798 6799 mutex_lock(&mdsc->mutex); 6800 kick_requests(mdsc, mds); 6801 } 6802 __wake_requests(mdsc, &mdsc->waiting_for_map); 6803 mutex_unlock(&mdsc->mutex); 6804 } 6805 6806 static void ceph_mdsc_stop(struct ceph_mds_client *mdsc) 6807 { 6808 doutc(mdsc->fsc->client, "stop\n"); 6809 /* 6810 * Make sure the delayed work stopped before releasing 6811 * the resources. 6812 * 6813 * Because the cancel_delayed_work_sync() will only 6814 * guarantee that the work finishes executing. But the 6815 * delayed work will re-arm itself again after that. 6816 */ 6817 flush_delayed_work(&mdsc->delayed_work); 6818 6819 if (mdsc->mdsmap) 6820 ceph_mdsmap_destroy(mdsc->mdsmap); 6821 kfree(mdsc->sessions); 6822 ceph_caps_finalize(mdsc); 6823 6824 if (mdsc->s_cap_auths) { 6825 int i; 6826 6827 for (i = 0; i < mdsc->s_cap_auths_num; i++) { 6828 kfree(mdsc->s_cap_auths[i].match.gids); 6829 kfree(mdsc->s_cap_auths[i].match.path); 6830 kfree(mdsc->s_cap_auths[i].match.fs_name); 6831 } 6832 kfree(mdsc->s_cap_auths); 6833 } 6834 6835 ceph_pool_perm_destroy(mdsc); 6836 } 6837 6838 void ceph_mdsc_destroy(struct ceph_fs_client *fsc) 6839 { 6840 struct ceph_mds_client *mdsc = fsc->mdsc; 6841 doutc(fsc->client, "%p\n", mdsc); 6842 6843 if (!mdsc) 6844 return; 6845 6846 /* flush out any connection work with references to us */ 6847 ceph_msgr_flush(); 6848 6849 /* 6850 * Mark reset as failed and wake any blocked waiters before 6851 * cancelling, so unmount doesn't stall on blocked_wq timeout 6852 * if cancel_work_sync() prevents the work from running. 6853 */ 6854 spin_lock(&mdsc->reset_state.lock); 6855 mdsc->reset_state.shutdown = true; 6856 if (mdsc->reset_state.phase != CEPH_CLIENT_RESET_IDLE) { 6857 mdsc->reset_state.phase = CEPH_CLIENT_RESET_IDLE; 6858 mdsc->reset_state.last_errno = -ESHUTDOWN; 6859 mdsc->reset_state.last_finish = jiffies; 6860 mdsc->reset_state.failure_count++; 6861 } 6862 spin_unlock(&mdsc->reset_state.lock); 6863 wake_up_all(&mdsc->reset_state.blocked_wq); 6864 6865 cancel_work_sync(&mdsc->reset_work); 6866 ceph_mdsc_stop(mdsc); 6867 6868 ceph_metric_destroy(&mdsc->metric); 6869 ceph_subvolume_metrics_destroy(&mdsc->subvol_metrics); 6870 kfree(mdsc->subvol_metrics_last); 6871 6872 fsc->mdsc = NULL; 6873 kfree(mdsc); 6874 doutc(fsc->client, "%p done\n", mdsc); 6875 } 6876 6877 void ceph_mdsc_handle_fsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg) 6878 { 6879 struct ceph_fs_client *fsc = mdsc->fsc; 6880 struct ceph_client *cl = fsc->client; 6881 const char *mds_namespace = fsc->mount_options->mds_namespace; 6882 void *p = msg->front.iov_base; 6883 void *end = p + msg->front.iov_len; 6884 u32 epoch; 6885 u32 num_fs; 6886 u32 mount_fscid = (u32)-1; 6887 int err = -EINVAL; 6888 6889 ceph_decode_need(&p, end, sizeof(u32), bad); 6890 epoch = ceph_decode_32(&p); 6891 6892 doutc(cl, "epoch %u\n", epoch); 6893 6894 /* struct_v, struct_cv, map_len, epoch, legacy_client_fscid */ 6895 ceph_decode_skip_n(&p, end, 2 + sizeof(u32) * 3, bad); 6896 6897 ceph_decode_32_safe(&p, end, num_fs, bad); 6898 while (num_fs-- > 0) { 6899 void *info_p, *info_end; 6900 u32 info_len; 6901 u32 fscid, namelen; 6902 6903 ceph_decode_need(&p, end, 2 + sizeof(u32), bad); 6904 p += 2; // info_v, info_cv 6905 info_len = ceph_decode_32(&p); 6906 ceph_decode_need(&p, end, info_len, bad); 6907 info_p = p; 6908 info_end = p + info_len; 6909 p = info_end; 6910 6911 ceph_decode_need(&info_p, info_end, sizeof(u32) * 2, bad); 6912 fscid = ceph_decode_32(&info_p); 6913 namelen = ceph_decode_32(&info_p); 6914 ceph_decode_need(&info_p, info_end, namelen, bad); 6915 6916 if (mds_namespace && 6917 strlen(mds_namespace) == namelen && 6918 !strncmp(mds_namespace, (char *)info_p, namelen)) { 6919 mount_fscid = fscid; 6920 break; 6921 } 6922 } 6923 6924 ceph_monc_got_map(&fsc->client->monc, CEPH_SUB_FSMAP, epoch); 6925 if (mount_fscid != (u32)-1) { 6926 fsc->client->monc.fs_cluster_id = mount_fscid; 6927 ceph_monc_want_map(&fsc->client->monc, CEPH_SUB_MDSMAP, 6928 0, true); 6929 ceph_monc_renew_subs(&fsc->client->monc); 6930 } else { 6931 err = -ENOENT; 6932 goto err_out; 6933 } 6934 return; 6935 6936 bad: 6937 pr_err_client(cl, "error decoding fsmap %d. Shutting down mount.\n", 6938 err); 6939 ceph_umount_begin(mdsc->fsc->sb); 6940 ceph_msg_dump(msg); 6941 err_out: 6942 mutex_lock(&mdsc->mutex); 6943 mdsc->mdsmap_err = err; 6944 __wake_requests(mdsc, &mdsc->waiting_for_map); 6945 mutex_unlock(&mdsc->mutex); 6946 } 6947 6948 /* 6949 * handle mds map update. 6950 */ 6951 void ceph_mdsc_handle_mdsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg) 6952 { 6953 struct ceph_client *cl = mdsc->fsc->client; 6954 u32 epoch; 6955 u32 maplen; 6956 void *p = msg->front.iov_base; 6957 void *end = p + msg->front.iov_len; 6958 struct ceph_mdsmap *newmap, *oldmap; 6959 struct ceph_fsid fsid; 6960 int err = -EINVAL; 6961 6962 ceph_decode_need(&p, end, sizeof(fsid)+2*sizeof(u32), bad); 6963 ceph_decode_copy(&p, &fsid, sizeof(fsid)); 6964 if (ceph_check_fsid(mdsc->fsc->client, &fsid) < 0) 6965 return; 6966 epoch = ceph_decode_32(&p); 6967 maplen = ceph_decode_32(&p); 6968 doutc(cl, "epoch %u len %d\n", epoch, (int)maplen); 6969 6970 /* do we need it? */ 6971 mutex_lock(&mdsc->mutex); 6972 if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) { 6973 doutc(cl, "epoch %u <= our %u\n", epoch, mdsc->mdsmap->m_epoch); 6974 mutex_unlock(&mdsc->mutex); 6975 return; 6976 } 6977 6978 newmap = ceph_mdsmap_decode(mdsc, &p, end, ceph_msgr2(mdsc->fsc->client)); 6979 if (IS_ERR(newmap)) { 6980 err = PTR_ERR(newmap); 6981 goto bad_unlock; 6982 } 6983 6984 /* swap into place */ 6985 if (mdsc->mdsmap) { 6986 oldmap = mdsc->mdsmap; 6987 mdsc->mdsmap = newmap; 6988 check_new_map(mdsc, newmap, oldmap); 6989 ceph_mdsmap_destroy(oldmap); 6990 } else { 6991 mdsc->mdsmap = newmap; /* first mds map */ 6992 } 6993 mdsc->fsc->max_file_size = min((loff_t)mdsc->mdsmap->m_max_file_size, 6994 MAX_LFS_FILESIZE); 6995 6996 __wake_requests(mdsc, &mdsc->waiting_for_map); 6997 ceph_monc_got_map(&mdsc->fsc->client->monc, CEPH_SUB_MDSMAP, 6998 mdsc->mdsmap->m_epoch); 6999 7000 mutex_unlock(&mdsc->mutex); 7001 schedule_delayed(mdsc, 0); 7002 return; 7003 7004 bad_unlock: 7005 mutex_unlock(&mdsc->mutex); 7006 bad: 7007 pr_err_client(cl, "error decoding mdsmap %d. Shutting down mount.\n", 7008 err); 7009 ceph_umount_begin(mdsc->fsc->sb); 7010 ceph_msg_dump(msg); 7011 return; 7012 } 7013 7014 static struct ceph_connection *mds_get_con(struct ceph_connection *con) 7015 { 7016 struct ceph_mds_session *s = con->private; 7017 7018 if (ceph_get_mds_session(s)) 7019 return con; 7020 return NULL; 7021 } 7022 7023 static void mds_put_con(struct ceph_connection *con) 7024 { 7025 struct ceph_mds_session *s = con->private; 7026 7027 ceph_put_mds_session(s); 7028 } 7029 7030 /* 7031 * if the client is unresponsive for long enough, the mds will kill 7032 * the session entirely. 7033 */ 7034 static void mds_peer_reset(struct ceph_connection *con) 7035 { 7036 struct ceph_mds_session *s = con->private; 7037 struct ceph_mds_client *mdsc = s->s_mdsc; 7038 int session_state; 7039 7040 pr_warn_client(mdsc->fsc->client, "mds%d closed our session\n", 7041 s->s_mds); 7042 7043 if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO || 7044 ceph_mdsmap_get_state(mdsc->mdsmap, s->s_mds) < CEPH_MDS_STATE_RECONNECT) 7045 return; 7046 7047 /* 7048 * Only reconnect if MDS is in its RECONNECT phase. An MDS past 7049 * RECONNECT (REJOIN, CLIENTREPLAY, ACTIVE) will reject reconnect 7050 * attempts, so those states fall through to session teardown below. 7051 */ 7052 if (ceph_mdsmap_get_state(mdsc->mdsmap, s->s_mds) == CEPH_MDS_STATE_RECONNECT) { 7053 int rc = send_mds_reconnect(mdsc, s); 7054 7055 if (rc) 7056 pr_warn_client(mdsc->fsc->client, 7057 "mds%d reconnect failed: %d\n", 7058 s->s_mds, rc); 7059 return; 7060 } 7061 7062 /* 7063 * MDS is active (past RECONNECT). It will not accept a 7064 * CLIENT_RECONNECT from us, so tear the session down locally 7065 * and let new requests re-open a fresh session. 7066 * 7067 * Snapshot session state with READ_ONCE, then revalidate under 7068 * mdsc->mutex before acting. The subsequent mdsc->mutex 7069 * section rechecks s_state to catch concurrent transitions, so 7070 * the lockless snapshot here is safe. s->s_mutex is taken 7071 * separately for cleanup after unregistration, which avoids 7072 * introducing a new s->s_mutex + mdsc->mutex nesting. 7073 */ 7074 session_state = READ_ONCE(s->s_state); 7075 7076 switch (session_state) { 7077 case CEPH_MDS_SESSION_RESTARTING: 7078 case CEPH_MDS_SESSION_RECONNECTING: 7079 case CEPH_MDS_SESSION_CLOSING: 7080 case CEPH_MDS_SESSION_OPEN: 7081 case CEPH_MDS_SESSION_HUNG: 7082 case CEPH_MDS_SESSION_OPENING: 7083 mutex_lock(&mdsc->mutex); 7084 if (s->s_mds >= mdsc->max_sessions || 7085 mdsc->sessions[s->s_mds] != s || 7086 s->s_state != session_state) { 7087 pr_info_client(mdsc->fsc->client, 7088 "mds%d state changed to %s during peer reset\n", 7089 s->s_mds, 7090 ceph_session_state_name(s->s_state)); 7091 mutex_unlock(&mdsc->mutex); 7092 return; 7093 } 7094 7095 ceph_get_mds_session(s); 7096 s->s_state = CEPH_MDS_SESSION_CLOSED; 7097 __unregister_session(mdsc, s); 7098 __wake_requests(mdsc, &s->s_waiting); 7099 mutex_unlock(&mdsc->mutex); 7100 7101 mutex_lock(&s->s_mutex); 7102 cleanup_session_requests(mdsc, s); 7103 remove_session_caps(s); 7104 mutex_unlock(&s->s_mutex); 7105 7106 wake_up_all(&mdsc->session_close_wq); 7107 7108 mutex_lock(&mdsc->mutex); 7109 kick_requests(mdsc, s->s_mds); 7110 mutex_unlock(&mdsc->mutex); 7111 7112 ceph_put_mds_session(s); 7113 break; 7114 case CEPH_MDS_SESSION_CLOSED: 7115 case CEPH_MDS_SESSION_REJECTED: 7116 break; 7117 default: 7118 pr_warn_client(mdsc->fsc->client, 7119 "mds%d peer reset in unexpected state %s\n", 7120 s->s_mds, 7121 ceph_session_state_name(session_state)); 7122 break; 7123 } 7124 } 7125 7126 static void mds_dispatch(struct ceph_connection *con, struct ceph_msg *msg) 7127 { 7128 struct ceph_mds_session *s = con->private; 7129 struct ceph_mds_client *mdsc = s->s_mdsc; 7130 struct ceph_client *cl = mdsc->fsc->client; 7131 int type = le16_to_cpu(msg->hdr.type); 7132 7133 mutex_lock(&mdsc->mutex); 7134 if (__verify_registered_session(mdsc, s) < 0) { 7135 doutc(cl, "dropping tid %llu from unregistered session %d\n", 7136 le64_to_cpu(msg->hdr.tid), s->s_mds); 7137 mutex_unlock(&mdsc->mutex); 7138 goto out; 7139 } 7140 mutex_unlock(&mdsc->mutex); 7141 7142 switch (type) { 7143 case CEPH_MSG_MDS_MAP: 7144 ceph_mdsc_handle_mdsmap(mdsc, msg); 7145 break; 7146 case CEPH_MSG_FS_MAP_USER: 7147 ceph_mdsc_handle_fsmap(mdsc, msg); 7148 break; 7149 case CEPH_MSG_CLIENT_SESSION: 7150 handle_session(s, msg); 7151 break; 7152 case CEPH_MSG_CLIENT_REPLY: 7153 handle_reply(s, msg); 7154 break; 7155 case CEPH_MSG_CLIENT_REQUEST_FORWARD: 7156 handle_forward(mdsc, s, msg); 7157 break; 7158 case CEPH_MSG_CLIENT_CAPS: 7159 ceph_handle_caps(s, msg); 7160 break; 7161 case CEPH_MSG_CLIENT_SNAP: 7162 ceph_handle_snap(mdsc, s, msg); 7163 break; 7164 case CEPH_MSG_CLIENT_LEASE: 7165 handle_lease(mdsc, s, msg); 7166 break; 7167 case CEPH_MSG_CLIENT_QUOTA: 7168 ceph_handle_quota(mdsc, s, msg); 7169 break; 7170 7171 default: 7172 pr_err_client(cl, "received unknown message type %d %s\n", 7173 type, ceph_msg_type_name(type)); 7174 } 7175 out: 7176 ceph_msg_put(msg); 7177 } 7178 7179 /* 7180 * authentication 7181 */ 7182 7183 /* 7184 * Note: returned pointer is the address of a structure that's 7185 * managed separately. Caller must *not* attempt to free it. 7186 */ 7187 static struct ceph_auth_handshake * 7188 mds_get_authorizer(struct ceph_connection *con, int *proto, int force_new) 7189 { 7190 struct ceph_mds_session *s = con->private; 7191 struct ceph_mds_client *mdsc = s->s_mdsc; 7192 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth; 7193 struct ceph_auth_handshake *auth = &s->s_auth; 7194 int ret; 7195 7196 ret = __ceph_auth_get_authorizer(ac, auth, CEPH_ENTITY_TYPE_MDS, 7197 force_new, proto, NULL, NULL); 7198 if (ret) 7199 return ERR_PTR(ret); 7200 7201 return auth; 7202 } 7203 7204 static int mds_add_authorizer_challenge(struct ceph_connection *con, 7205 void *challenge_buf, int challenge_buf_len) 7206 { 7207 struct ceph_mds_session *s = con->private; 7208 struct ceph_mds_client *mdsc = s->s_mdsc; 7209 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth; 7210 7211 return ceph_auth_add_authorizer_challenge(ac, s->s_auth.authorizer, 7212 challenge_buf, challenge_buf_len); 7213 } 7214 7215 static int mds_verify_authorizer_reply(struct ceph_connection *con) 7216 { 7217 struct ceph_mds_session *s = con->private; 7218 struct ceph_mds_client *mdsc = s->s_mdsc; 7219 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth; 7220 struct ceph_auth_handshake *auth = &s->s_auth; 7221 7222 return ceph_auth_verify_authorizer_reply(ac, auth->authorizer, 7223 auth->authorizer_reply_buf, auth->authorizer_reply_buf_len, 7224 NULL, NULL, NULL, NULL); 7225 } 7226 7227 static int mds_invalidate_authorizer(struct ceph_connection *con) 7228 { 7229 struct ceph_mds_session *s = con->private; 7230 struct ceph_mds_client *mdsc = s->s_mdsc; 7231 struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth; 7232 7233 ceph_auth_invalidate_authorizer(ac, CEPH_ENTITY_TYPE_MDS); 7234 7235 return ceph_monc_validate_auth(&mdsc->fsc->client->monc); 7236 } 7237 7238 static int mds_get_auth_request(struct ceph_connection *con, 7239 void *buf, int *buf_len, 7240 void **authorizer, int *authorizer_len) 7241 { 7242 struct ceph_mds_session *s = con->private; 7243 struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth; 7244 struct ceph_auth_handshake *auth = &s->s_auth; 7245 int ret; 7246 7247 ret = ceph_auth_get_authorizer(ac, auth, CEPH_ENTITY_TYPE_MDS, 7248 buf, buf_len); 7249 if (ret) 7250 return ret; 7251 7252 *authorizer = auth->authorizer_buf; 7253 *authorizer_len = auth->authorizer_buf_len; 7254 return 0; 7255 } 7256 7257 static int mds_handle_auth_reply_more(struct ceph_connection *con, 7258 void *reply, int reply_len, 7259 void *buf, int *buf_len, 7260 void **authorizer, int *authorizer_len) 7261 { 7262 struct ceph_mds_session *s = con->private; 7263 struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth; 7264 struct ceph_auth_handshake *auth = &s->s_auth; 7265 int ret; 7266 7267 ret = ceph_auth_handle_svc_reply_more(ac, auth, reply, reply_len, 7268 buf, buf_len); 7269 if (ret) 7270 return ret; 7271 7272 *authorizer = auth->authorizer_buf; 7273 *authorizer_len = auth->authorizer_buf_len; 7274 return 0; 7275 } 7276 7277 static int mds_handle_auth_done(struct ceph_connection *con, 7278 u64 global_id, void *reply, int reply_len, 7279 u8 *session_key, int *session_key_len, 7280 u8 *con_secret, int *con_secret_len) 7281 { 7282 struct ceph_mds_session *s = con->private; 7283 struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth; 7284 struct ceph_auth_handshake *auth = &s->s_auth; 7285 7286 return ceph_auth_handle_svc_reply_done(ac, auth, reply, reply_len, 7287 session_key, session_key_len, 7288 con_secret, con_secret_len); 7289 } 7290 7291 static int mds_handle_auth_bad_method(struct ceph_connection *con, 7292 int used_proto, int result, 7293 const int *allowed_protos, int proto_cnt, 7294 const int *allowed_modes, int mode_cnt) 7295 { 7296 struct ceph_mds_session *s = con->private; 7297 struct ceph_mon_client *monc = &s->s_mdsc->fsc->client->monc; 7298 int ret; 7299 7300 if (ceph_auth_handle_bad_authorizer(monc->auth, CEPH_ENTITY_TYPE_MDS, 7301 used_proto, result, 7302 allowed_protos, proto_cnt, 7303 allowed_modes, mode_cnt)) { 7304 ret = ceph_monc_validate_auth(monc); 7305 if (ret) 7306 return ret; 7307 } 7308 7309 return -EACCES; 7310 } 7311 7312 static struct ceph_msg *mds_alloc_msg(struct ceph_connection *con, 7313 struct ceph_msg_header *hdr, int *skip) 7314 { 7315 struct ceph_msg *msg; 7316 int type = (int) le16_to_cpu(hdr->type); 7317 int front_len = (int) le32_to_cpu(hdr->front_len); 7318 7319 if (con->in_msg) 7320 return con->in_msg; 7321 7322 *skip = 0; 7323 msg = ceph_msg_new(type, front_len, GFP_NOFS, false); 7324 if (!msg) { 7325 pr_err("unable to allocate msg type %d len %d\n", 7326 type, front_len); 7327 return NULL; 7328 } 7329 7330 return msg; 7331 } 7332 7333 static int mds_sign_message(struct ceph_msg *msg) 7334 { 7335 struct ceph_mds_session *s = msg->con->private; 7336 struct ceph_auth_handshake *auth = &s->s_auth; 7337 7338 return ceph_auth_sign_message(auth, msg); 7339 } 7340 7341 static int mds_check_message_signature(struct ceph_msg *msg) 7342 { 7343 struct ceph_mds_session *s = msg->con->private; 7344 struct ceph_auth_handshake *auth = &s->s_auth; 7345 7346 return ceph_auth_check_message_signature(auth, msg); 7347 } 7348 7349 static const struct ceph_connection_operations mds_con_ops = { 7350 .get = mds_get_con, 7351 .put = mds_put_con, 7352 .alloc_msg = mds_alloc_msg, 7353 .dispatch = mds_dispatch, 7354 .peer_reset = mds_peer_reset, 7355 .get_authorizer = mds_get_authorizer, 7356 .add_authorizer_challenge = mds_add_authorizer_challenge, 7357 .verify_authorizer_reply = mds_verify_authorizer_reply, 7358 .invalidate_authorizer = mds_invalidate_authorizer, 7359 .sign_message = mds_sign_message, 7360 .check_message_signature = mds_check_message_signature, 7361 .get_auth_request = mds_get_auth_request, 7362 .handle_auth_reply_more = mds_handle_auth_reply_more, 7363 .handle_auth_done = mds_handle_auth_done, 7364 .handle_auth_bad_method = mds_handle_auth_bad_method, 7365 }; 7366 7367 /* eof */ 7368