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