1 // SPDX-License-Identifier: GPL-2.0 2 3 #include <linux/ceph/ceph_debug.h> 4 5 #include <linux/module.h> 6 #include <linux/slab.h> 7 8 #include <linux/ceph/libceph.h> 9 #include <linux/ceph/osdmap.h> 10 #include <linux/ceph/decode.h> 11 #include <linux/crush/hash.h> 12 #include <linux/crush/mapper.h> 13 14 static __printf(2, 3) 15 void osdmap_info(const struct ceph_osdmap *map, const char *fmt, ...) 16 { 17 struct va_format vaf; 18 va_list args; 19 20 va_start(args, fmt); 21 vaf.fmt = fmt; 22 vaf.va = &args; 23 24 printk(KERN_INFO "%s (%pU e%u): %pV", KBUILD_MODNAME, &map->fsid, 25 map->epoch, &vaf); 26 27 va_end(args); 28 } 29 30 char *ceph_osdmap_state_str(char *str, int len, u32 state) 31 { 32 if (!len) 33 return str; 34 35 if ((state & CEPH_OSD_EXISTS) && (state & CEPH_OSD_UP)) 36 snprintf(str, len, "exists, up"); 37 else if (state & CEPH_OSD_EXISTS) 38 snprintf(str, len, "exists"); 39 else if (state & CEPH_OSD_UP) 40 snprintf(str, len, "up"); 41 else 42 snprintf(str, len, "doesn't exist"); 43 44 return str; 45 } 46 47 /* maps */ 48 49 static int calc_bits_of(unsigned int t) 50 { 51 int b = 0; 52 while (t) { 53 t = t >> 1; 54 b++; 55 } 56 return b; 57 } 58 59 /* 60 * the foo_mask is the smallest value 2^n-1 that is >= foo. 61 */ 62 static void calc_pg_masks(struct ceph_pg_pool_info *pi) 63 { 64 pi->pg_num_mask = (1 << calc_bits_of(pi->pg_num-1)) - 1; 65 pi->pgp_num_mask = (1 << calc_bits_of(pi->pgp_num-1)) - 1; 66 } 67 68 /* 69 * decode crush map 70 */ 71 static int crush_decode_uniform_bucket(void **p, void *end, 72 struct crush_bucket_uniform *b) 73 { 74 dout("crush_decode_uniform_bucket %p to %p\n", *p, end); 75 ceph_decode_32_safe(p, end, b->item_weight, bad); 76 return 0; 77 bad: 78 return -EINVAL; 79 } 80 81 static int crush_decode_list_bucket(void **p, void *end, 82 struct crush_bucket_list *b) 83 { 84 int j; 85 dout("crush_decode_list_bucket %p to %p\n", *p, end); 86 b->item_weights = kcalloc(b->h.size, sizeof(u32), GFP_NOFS); 87 if (b->item_weights == NULL) 88 return -ENOMEM; 89 b->sum_weights = kcalloc(b->h.size, sizeof(u32), GFP_NOFS); 90 if (b->sum_weights == NULL) 91 return -ENOMEM; 92 ceph_decode_need(p, end, 2 * b->h.size * sizeof(u32), bad); 93 for (j = 0; j < b->h.size; j++) { 94 b->item_weights[j] = ceph_decode_32(p); 95 b->sum_weights[j] = ceph_decode_32(p); 96 } 97 return 0; 98 bad: 99 return -EINVAL; 100 } 101 102 static int crush_decode_tree_bucket(void **p, void *end, 103 struct crush_bucket_tree *b) 104 { 105 int j; 106 dout("crush_decode_tree_bucket %p to %p\n", *p, end); 107 ceph_decode_8_safe(p, end, b->num_nodes, bad); 108 b->node_weights = kcalloc(b->num_nodes, sizeof(u32), GFP_NOFS); 109 if (b->node_weights == NULL) 110 return -ENOMEM; 111 ceph_decode_need(p, end, b->num_nodes * sizeof(u32), bad); 112 for (j = 0; j < b->num_nodes; j++) 113 b->node_weights[j] = ceph_decode_32(p); 114 return 0; 115 bad: 116 return -EINVAL; 117 } 118 119 static int crush_decode_straw_bucket(void **p, void *end, 120 struct crush_bucket_straw *b) 121 { 122 int j; 123 dout("crush_decode_straw_bucket %p to %p\n", *p, end); 124 b->item_weights = kcalloc(b->h.size, sizeof(u32), GFP_NOFS); 125 if (b->item_weights == NULL) 126 return -ENOMEM; 127 b->straws = kcalloc(b->h.size, sizeof(u32), GFP_NOFS); 128 if (b->straws == NULL) 129 return -ENOMEM; 130 ceph_decode_need(p, end, 2 * b->h.size * sizeof(u32), bad); 131 for (j = 0; j < b->h.size; j++) { 132 b->item_weights[j] = ceph_decode_32(p); 133 b->straws[j] = ceph_decode_32(p); 134 } 135 return 0; 136 bad: 137 return -EINVAL; 138 } 139 140 static int crush_decode_straw2_bucket(void **p, void *end, 141 struct crush_bucket_straw2 *b) 142 { 143 int j; 144 dout("crush_decode_straw2_bucket %p to %p\n", *p, end); 145 b->item_weights = kcalloc(b->h.size, sizeof(u32), GFP_NOFS); 146 if (b->item_weights == NULL) 147 return -ENOMEM; 148 ceph_decode_need(p, end, b->h.size * sizeof(u32), bad); 149 for (j = 0; j < b->h.size; j++) 150 b->item_weights[j] = ceph_decode_32(p); 151 return 0; 152 bad: 153 return -EINVAL; 154 } 155 156 struct crush_name_node { 157 struct rb_node cn_node; 158 int cn_id; 159 char cn_name[]; 160 }; 161 162 static struct crush_name_node *alloc_crush_name(size_t name_len) 163 { 164 struct crush_name_node *cn; 165 166 cn = kmalloc(sizeof(*cn) + name_len + 1, GFP_NOIO); 167 if (!cn) 168 return NULL; 169 170 RB_CLEAR_NODE(&cn->cn_node); 171 return cn; 172 } 173 174 static void free_crush_name(struct crush_name_node *cn) 175 { 176 WARN_ON(!RB_EMPTY_NODE(&cn->cn_node)); 177 178 kfree(cn); 179 } 180 181 DEFINE_RB_FUNCS(crush_name, struct crush_name_node, cn_id, cn_node) 182 183 static int decode_crush_names(void **p, void *end, struct rb_root *root) 184 { 185 u32 n; 186 187 ceph_decode_32_safe(p, end, n, e_inval); 188 while (n--) { 189 struct crush_name_node *cn; 190 int id; 191 u32 name_len; 192 193 ceph_decode_32_safe(p, end, id, e_inval); 194 ceph_decode_32_safe(p, end, name_len, e_inval); 195 ceph_decode_need(p, end, name_len, e_inval); 196 197 cn = alloc_crush_name(name_len); 198 if (!cn) 199 return -ENOMEM; 200 201 cn->cn_id = id; 202 memcpy(cn->cn_name, *p, name_len); 203 cn->cn_name[name_len] = '\0'; 204 *p += name_len; 205 206 if (!__insert_crush_name(root, cn)) { 207 free_crush_name(cn); 208 return -EEXIST; 209 } 210 } 211 212 return 0; 213 214 e_inval: 215 return -EINVAL; 216 } 217 218 void clear_crush_names(struct rb_root *root) 219 { 220 while (!RB_EMPTY_ROOT(root)) { 221 struct crush_name_node *cn = 222 rb_entry(rb_first(root), struct crush_name_node, cn_node); 223 224 erase_crush_name(root, cn); 225 free_crush_name(cn); 226 } 227 } 228 229 static struct crush_choose_arg_map *alloc_choose_arg_map(void) 230 { 231 struct crush_choose_arg_map *arg_map; 232 233 arg_map = kzalloc_obj(*arg_map, GFP_NOIO); 234 if (!arg_map) 235 return NULL; 236 237 RB_CLEAR_NODE(&arg_map->node); 238 return arg_map; 239 } 240 241 static void free_choose_arg_map(struct crush_choose_arg_map *arg_map) 242 { 243 int i, j; 244 245 if (!arg_map) 246 return; 247 248 WARN_ON(!RB_EMPTY_NODE(&arg_map->node)); 249 250 if (arg_map->args) { 251 for (i = 0; i < arg_map->size; i++) { 252 struct crush_choose_arg *arg = &arg_map->args[i]; 253 if (arg->weight_set) { 254 for (j = 0; j < arg->weight_set_size; j++) 255 kfree(arg->weight_set[j].weights); 256 kfree(arg->weight_set); 257 } 258 kfree(arg->ids); 259 } 260 kfree(arg_map->args); 261 } 262 kfree(arg_map); 263 } 264 265 DEFINE_RB_FUNCS(choose_arg_map, struct crush_choose_arg_map, choose_args_index, 266 node); 267 268 void clear_choose_args(struct crush_map *c) 269 { 270 while (!RB_EMPTY_ROOT(&c->choose_args)) { 271 struct crush_choose_arg_map *arg_map = 272 rb_entry(rb_first(&c->choose_args), 273 struct crush_choose_arg_map, node); 274 275 erase_choose_arg_map(&c->choose_args, arg_map); 276 free_choose_arg_map(arg_map); 277 } 278 } 279 280 static u32 *decode_array_32_alloc(void **p, void *end, u32 *plen) 281 { 282 u32 *a = NULL; 283 u32 len; 284 int ret; 285 286 ceph_decode_32_safe(p, end, len, e_inval); 287 if (len) { 288 u32 i; 289 290 a = kmalloc_array(len, sizeof(u32), GFP_NOIO); 291 if (!a) { 292 ret = -ENOMEM; 293 goto fail; 294 } 295 296 ceph_decode_need(p, end, len * sizeof(u32), e_inval); 297 for (i = 0; i < len; i++) 298 a[i] = ceph_decode_32(p); 299 } 300 301 *plen = len; 302 return a; 303 304 e_inval: 305 ret = -EINVAL; 306 fail: 307 kfree(a); 308 return ERR_PTR(ret); 309 } 310 311 /* 312 * Assumes @arg is zero-initialized. 313 */ 314 static int decode_choose_arg(void **p, void *end, struct crush_choose_arg *arg) 315 { 316 int ret; 317 318 ceph_decode_32_safe(p, end, arg->weight_set_size, e_inval); 319 if (arg->weight_set_size) { 320 u32 i; 321 322 arg->weight_set = kmalloc_objs(*arg->weight_set, 323 arg->weight_set_size, GFP_NOIO); 324 if (!arg->weight_set) 325 return -ENOMEM; 326 327 for (i = 0; i < arg->weight_set_size; i++) { 328 struct crush_weight_set *w = &arg->weight_set[i]; 329 330 w->weights = decode_array_32_alloc(p, end, &w->size); 331 if (IS_ERR(w->weights)) { 332 ret = PTR_ERR(w->weights); 333 w->weights = NULL; 334 return ret; 335 } 336 } 337 } 338 339 arg->ids = decode_array_32_alloc(p, end, &arg->ids_size); 340 if (IS_ERR(arg->ids)) { 341 ret = PTR_ERR(arg->ids); 342 arg->ids = NULL; 343 return ret; 344 } 345 346 return 0; 347 348 e_inval: 349 return -EINVAL; 350 } 351 352 static int decode_choose_args(void **p, void *end, struct crush_map *c) 353 { 354 struct crush_choose_arg_map *arg_map = NULL; 355 u32 num_choose_arg_maps, num_buckets; 356 int ret; 357 358 ceph_decode_32_safe(p, end, num_choose_arg_maps, e_inval); 359 while (num_choose_arg_maps--) { 360 arg_map = alloc_choose_arg_map(); 361 if (!arg_map) { 362 ret = -ENOMEM; 363 goto fail; 364 } 365 366 ceph_decode_64_safe(p, end, arg_map->choose_args_index, 367 e_inval); 368 arg_map->size = c->max_buckets; 369 arg_map->args = kzalloc_objs(*arg_map->args, arg_map->size, 370 GFP_NOIO); 371 if (!arg_map->args) { 372 ret = -ENOMEM; 373 goto fail; 374 } 375 376 ceph_decode_32_safe(p, end, num_buckets, e_inval); 377 while (num_buckets--) { 378 struct crush_choose_arg *arg; 379 u32 bucket_index; 380 381 ceph_decode_32_safe(p, end, bucket_index, e_inval); 382 if (bucket_index >= arg_map->size) 383 goto e_inval; 384 385 arg = &arg_map->args[bucket_index]; 386 ret = decode_choose_arg(p, end, arg); 387 if (ret) 388 goto fail; 389 390 if (arg->ids_size && 391 (!c->buckets[bucket_index] || 392 arg->ids_size != c->buckets[bucket_index]->size)) 393 goto e_inval; 394 } 395 396 if (!__insert_choose_arg_map(&c->choose_args, arg_map)) { 397 ret = -EEXIST; 398 goto fail; 399 } 400 } 401 402 return 0; 403 404 e_inval: 405 ret = -EINVAL; 406 fail: 407 free_choose_arg_map(arg_map); 408 return ret; 409 } 410 411 static void crush_finalize(struct crush_map *c) 412 { 413 __s32 b; 414 415 /* Space for the array of pointers to per-bucket workspace */ 416 c->working_size = sizeof(struct crush_work) + 417 c->max_buckets * sizeof(struct crush_work_bucket *); 418 419 for (b = 0; b < c->max_buckets; b++) { 420 if (!c->buckets[b]) 421 continue; 422 423 switch (c->buckets[b]->alg) { 424 default: 425 /* 426 * The base case, permutation variables and 427 * the pointer to the permutation array. 428 */ 429 c->working_size += sizeof(struct crush_work_bucket); 430 break; 431 } 432 /* Every bucket has a permutation array. */ 433 c->working_size += c->buckets[b]->size * sizeof(__u32); 434 } 435 } 436 437 static struct crush_map *crush_decode(void *pbyval, void *end) 438 { 439 struct crush_map *c; 440 int err; 441 int i, j; 442 void **p = &pbyval; 443 void *start = pbyval; 444 u32 magic; 445 446 dout("crush_decode %p to %p len %d\n", *p, end, (int)(end - *p)); 447 448 c = kzalloc_obj(*c, GFP_NOFS); 449 if (c == NULL) 450 return ERR_PTR(-ENOMEM); 451 452 c->type_names = RB_ROOT; 453 c->names = RB_ROOT; 454 c->choose_args = RB_ROOT; 455 456 /* set tunables to default values */ 457 c->choose_local_tries = 2; 458 c->choose_local_fallback_tries = 5; 459 c->choose_total_tries = 19; 460 c->chooseleaf_descend_once = 0; 461 462 ceph_decode_need(p, end, 4*sizeof(u32), bad); 463 magic = ceph_decode_32(p); 464 if (magic != CRUSH_MAGIC) { 465 pr_err("crush_decode magic %x != current %x\n", 466 (unsigned int)magic, (unsigned int)CRUSH_MAGIC); 467 goto bad; 468 } 469 c->max_buckets = ceph_decode_32(p); 470 c->max_rules = ceph_decode_32(p); 471 c->max_devices = ceph_decode_32(p); 472 473 c->buckets = kzalloc_objs(*c->buckets, c->max_buckets, GFP_NOFS); 474 if (c->buckets == NULL) 475 goto badmem; 476 c->rules = kzalloc_objs(*c->rules, c->max_rules, GFP_NOFS); 477 if (c->rules == NULL) 478 goto badmem; 479 480 /* buckets */ 481 for (i = 0; i < c->max_buckets; i++) { 482 int size = 0; 483 u32 alg; 484 struct crush_bucket *b; 485 486 ceph_decode_32_safe(p, end, alg, bad); 487 if (alg == 0) { 488 c->buckets[i] = NULL; 489 continue; 490 } 491 dout("crush_decode bucket %d off %x %p to %p\n", 492 i, (int)(*p-start), *p, end); 493 494 switch (alg) { 495 case CRUSH_BUCKET_UNIFORM: 496 size = sizeof(struct crush_bucket_uniform); 497 break; 498 case CRUSH_BUCKET_LIST: 499 size = sizeof(struct crush_bucket_list); 500 break; 501 case CRUSH_BUCKET_TREE: 502 size = sizeof(struct crush_bucket_tree); 503 break; 504 case CRUSH_BUCKET_STRAW: 505 size = sizeof(struct crush_bucket_straw); 506 break; 507 case CRUSH_BUCKET_STRAW2: 508 size = sizeof(struct crush_bucket_straw2); 509 break; 510 default: 511 goto bad; 512 } 513 BUG_ON(size == 0); 514 b = c->buckets[i] = kzalloc(size, GFP_NOFS); 515 if (b == NULL) 516 goto badmem; 517 518 ceph_decode_need(p, end, 4*sizeof(u32), bad); 519 b->id = ceph_decode_32(p); 520 if (b->id != -1 - i) 521 goto bad; 522 b->type = ceph_decode_16(p); 523 if (b->type == 0) 524 goto bad; 525 b->alg = ceph_decode_8(p); 526 if (b->alg != alg) { 527 b->alg = 0; 528 goto bad; 529 } 530 b->hash = ceph_decode_8(p); 531 b->weight = ceph_decode_32(p); 532 b->size = ceph_decode_32(p); 533 534 dout("crush_decode bucket size %d off %x %p to %p\n", 535 b->size, (int)(*p-start), *p, end); 536 537 b->items = kzalloc_objs(__s32, b->size, GFP_NOFS); 538 if (b->items == NULL) 539 goto badmem; 540 541 ceph_decode_need(p, end, b->size*sizeof(u32), bad); 542 for (j = 0; j < b->size; j++) 543 b->items[j] = ceph_decode_32(p); 544 545 switch (b->alg) { 546 case CRUSH_BUCKET_UNIFORM: 547 err = crush_decode_uniform_bucket(p, end, 548 (struct crush_bucket_uniform *)b); 549 if (err < 0) 550 goto fail; 551 break; 552 case CRUSH_BUCKET_LIST: 553 err = crush_decode_list_bucket(p, end, 554 (struct crush_bucket_list *)b); 555 if (err < 0) 556 goto fail; 557 break; 558 case CRUSH_BUCKET_TREE: 559 err = crush_decode_tree_bucket(p, end, 560 (struct crush_bucket_tree *)b); 561 if (err < 0) 562 goto fail; 563 break; 564 case CRUSH_BUCKET_STRAW: 565 err = crush_decode_straw_bucket(p, end, 566 (struct crush_bucket_straw *)b); 567 if (err < 0) 568 goto fail; 569 break; 570 case CRUSH_BUCKET_STRAW2: 571 err = crush_decode_straw2_bucket(p, end, 572 (struct crush_bucket_straw2 *)b); 573 if (err < 0) 574 goto fail; 575 break; 576 } 577 } 578 579 /* rules */ 580 dout("rule vec is %p\n", c->rules); 581 for (i = 0; i < c->max_rules; i++) { 582 u32 yes; 583 struct crush_rule *r; 584 585 ceph_decode_32_safe(p, end, yes, bad); 586 if (!yes) { 587 dout("crush_decode NO rule %d off %x %p to %p\n", 588 i, (int)(*p-start), *p, end); 589 c->rules[i] = NULL; 590 continue; 591 } 592 593 dout("crush_decode rule %d off %x %p to %p\n", 594 i, (int)(*p-start), *p, end); 595 596 /* len */ 597 ceph_decode_32_safe(p, end, yes, bad); 598 #if BITS_PER_LONG == 32 599 if (yes > (ULONG_MAX - sizeof(*r)) 600 / sizeof(struct crush_rule_step)) 601 goto bad; 602 #endif 603 r = kmalloc_flex(*r, steps, yes, GFP_NOFS); 604 if (r == NULL) 605 goto badmem; 606 dout(" rule %d is at %p\n", i, r); 607 c->rules[i] = r; 608 r->len = yes; 609 ceph_decode_copy_safe(p, end, &r->mask, 4, bad); /* 4 u8's */ 610 ceph_decode_need(p, end, r->len*3*sizeof(u32), bad); 611 for (j = 0; j < r->len; j++) { 612 r->steps[j].op = ceph_decode_32(p); 613 r->steps[j].arg1 = ceph_decode_32(p); 614 r->steps[j].arg2 = ceph_decode_32(p); 615 } 616 } 617 618 err = decode_crush_names(p, end, &c->type_names); 619 if (err) 620 goto fail; 621 622 err = decode_crush_names(p, end, &c->names); 623 if (err) 624 goto fail; 625 626 ceph_decode_skip_map(p, end, 32, string, bad); /* rule_name_map */ 627 628 /* tunables */ 629 ceph_decode_need(p, end, 3*sizeof(u32), done); 630 c->choose_local_tries = ceph_decode_32(p); 631 c->choose_local_fallback_tries = ceph_decode_32(p); 632 c->choose_total_tries = ceph_decode_32(p); 633 dout("crush decode tunable choose_local_tries = %d\n", 634 c->choose_local_tries); 635 dout("crush decode tunable choose_local_fallback_tries = %d\n", 636 c->choose_local_fallback_tries); 637 dout("crush decode tunable choose_total_tries = %d\n", 638 c->choose_total_tries); 639 640 ceph_decode_need(p, end, sizeof(u32), done); 641 c->chooseleaf_descend_once = ceph_decode_32(p); 642 dout("crush decode tunable chooseleaf_descend_once = %d\n", 643 c->chooseleaf_descend_once); 644 645 ceph_decode_need(p, end, sizeof(u8), done); 646 c->chooseleaf_vary_r = ceph_decode_8(p); 647 dout("crush decode tunable chooseleaf_vary_r = %d\n", 648 c->chooseleaf_vary_r); 649 650 /* skip straw_calc_version, allowed_bucket_algs */ 651 ceph_decode_need(p, end, sizeof(u8) + sizeof(u32), done); 652 *p += sizeof(u8) + sizeof(u32); 653 654 ceph_decode_need(p, end, sizeof(u8), done); 655 c->chooseleaf_stable = ceph_decode_8(p); 656 dout("crush decode tunable chooseleaf_stable = %d\n", 657 c->chooseleaf_stable); 658 659 if (*p != end) { 660 /* class_map */ 661 ceph_decode_skip_map(p, end, 32, 32, bad); 662 /* class_name */ 663 ceph_decode_skip_map(p, end, 32, string, bad); 664 /* class_bucket */ 665 ceph_decode_skip_map_of_map(p, end, 32, 32, 32, bad); 666 } 667 668 if (*p != end) { 669 err = decode_choose_args(p, end, c); 670 if (err) 671 goto fail; 672 } 673 674 done: 675 crush_finalize(c); 676 dout("crush_decode success\n"); 677 return c; 678 679 badmem: 680 err = -ENOMEM; 681 fail: 682 dout("crush_decode fail %d\n", err); 683 crush_destroy(c); 684 return ERR_PTR(err); 685 686 bad: 687 err = -EINVAL; 688 goto fail; 689 } 690 691 int ceph_pg_compare(const struct ceph_pg *lhs, const struct ceph_pg *rhs) 692 { 693 if (lhs->pool < rhs->pool) 694 return -1; 695 if (lhs->pool > rhs->pool) 696 return 1; 697 if (lhs->seed < rhs->seed) 698 return -1; 699 if (lhs->seed > rhs->seed) 700 return 1; 701 702 return 0; 703 } 704 705 int ceph_spg_compare(const struct ceph_spg *lhs, const struct ceph_spg *rhs) 706 { 707 int ret; 708 709 ret = ceph_pg_compare(&lhs->pgid, &rhs->pgid); 710 if (ret) 711 return ret; 712 713 if (lhs->shard < rhs->shard) 714 return -1; 715 if (lhs->shard > rhs->shard) 716 return 1; 717 718 return 0; 719 } 720 721 static struct ceph_pg_mapping *alloc_pg_mapping(size_t payload_len) 722 { 723 struct ceph_pg_mapping *pg; 724 725 pg = kmalloc(sizeof(*pg) + payload_len, GFP_NOIO); 726 if (!pg) 727 return NULL; 728 729 RB_CLEAR_NODE(&pg->node); 730 return pg; 731 } 732 733 static void free_pg_mapping(struct ceph_pg_mapping *pg) 734 { 735 WARN_ON(!RB_EMPTY_NODE(&pg->node)); 736 737 kfree(pg); 738 } 739 740 /* 741 * rbtree of pg_mapping for handling pg_temp (explicit mapping of pgid 742 * to a set of osds) and primary_temp (explicit primary setting) 743 */ 744 DEFINE_RB_FUNCS2(pg_mapping, struct ceph_pg_mapping, pgid, ceph_pg_compare, 745 RB_BYPTR, const struct ceph_pg *, node) 746 747 /* 748 * rbtree of pg pool info 749 */ 750 DEFINE_RB_FUNCS(pg_pool, struct ceph_pg_pool_info, id, node) 751 752 struct ceph_pg_pool_info *ceph_pg_pool_by_id(struct ceph_osdmap *map, u64 id) 753 { 754 return lookup_pg_pool(&map->pg_pools, id); 755 } 756 757 const char *ceph_pg_pool_name_by_id(struct ceph_osdmap *map, u64 id) 758 { 759 struct ceph_pg_pool_info *pi; 760 761 if (id == CEPH_NOPOOL) 762 return NULL; 763 764 if (WARN_ON_ONCE(id > (u64) INT_MAX)) 765 return NULL; 766 767 pi = lookup_pg_pool(&map->pg_pools, id); 768 return pi ? pi->name : NULL; 769 } 770 EXPORT_SYMBOL(ceph_pg_pool_name_by_id); 771 772 int ceph_pg_poolid_by_name(struct ceph_osdmap *map, const char *name) 773 { 774 struct rb_node *rbp; 775 776 for (rbp = rb_first(&map->pg_pools); rbp; rbp = rb_next(rbp)) { 777 struct ceph_pg_pool_info *pi = 778 rb_entry(rbp, struct ceph_pg_pool_info, node); 779 if (pi->name && strcmp(pi->name, name) == 0) 780 return pi->id; 781 } 782 return -ENOENT; 783 } 784 EXPORT_SYMBOL(ceph_pg_poolid_by_name); 785 786 u64 ceph_pg_pool_flags(struct ceph_osdmap *map, u64 id) 787 { 788 struct ceph_pg_pool_info *pi; 789 790 pi = lookup_pg_pool(&map->pg_pools, id); 791 return pi ? pi->flags : 0; 792 } 793 EXPORT_SYMBOL(ceph_pg_pool_flags); 794 795 static void __remove_pg_pool(struct rb_root *root, struct ceph_pg_pool_info *pi) 796 { 797 erase_pg_pool(root, pi); 798 kfree(pi->name); 799 kfree(pi); 800 } 801 802 static int decode_pool(void **p, void *end, struct ceph_pg_pool_info *pi) 803 { 804 u8 ev, cv; 805 unsigned len, num; 806 void *pool_end; 807 808 ceph_decode_need(p, end, 2 + 4, bad); 809 ev = ceph_decode_8(p); /* encoding version */ 810 cv = ceph_decode_8(p); /* compat version */ 811 if (ev < 5) { 812 pr_warn("got v %d < 5 cv %d of ceph_pg_pool\n", ev, cv); 813 return -EINVAL; 814 } 815 if (cv > 9) { 816 pr_warn("got v %d cv %d > 9 of ceph_pg_pool\n", ev, cv); 817 return -EINVAL; 818 } 819 len = ceph_decode_32(p); 820 ceph_decode_need(p, end, len, bad); 821 pool_end = *p + len; 822 823 ceph_decode_need(p, end, 4 + 4 + 4, bad); 824 pi->type = ceph_decode_8(p); 825 pi->size = ceph_decode_8(p); 826 pi->crush_ruleset = ceph_decode_8(p); 827 pi->object_hash = ceph_decode_8(p); 828 pi->pg_num = ceph_decode_32(p); 829 pi->pgp_num = ceph_decode_32(p); 830 831 /* lpg*, last_change, snap_seq, snap_epoch */ 832 ceph_decode_skip_n(p, end, 8 + 4 + 8 + 4, bad); 833 834 /* skip snaps */ 835 ceph_decode_32_safe(p, end, num, bad); 836 while (num--) { 837 /* snapid key, pool snap (with versions) */ 838 ceph_decode_skip_n(p, end, 8 + 2, bad); 839 ceph_decode_skip_string(p, end, bad); 840 } 841 842 /* removed_snaps */ 843 ceph_decode_skip_map(p, end, 64, 64, bad); 844 845 ceph_decode_need(p, end, 8 + 8 + 4, bad); 846 *p += 8; /* skip auid */ 847 pi->flags = ceph_decode_64(p); 848 *p += 4; /* skip crash_replay_interval */ 849 850 if (ev >= 7) 851 ceph_decode_8_safe(p, end, pi->min_size, bad); 852 else 853 pi->min_size = pi->size - pi->size / 2; 854 855 if (ev >= 8) 856 /* quota_max_* */ 857 ceph_decode_skip_n(p, end, 8 + 8, bad); 858 859 if (ev >= 9) { 860 /* tiers */ 861 ceph_decode_skip_set(p, end, 64, bad); 862 863 ceph_decode_need(p, end, 8 + 1 + 8 + 8, bad); 864 *p += 8; /* skip tier_of */ 865 *p += 1; /* skip cache_mode */ 866 pi->read_tier = ceph_decode_64(p); 867 pi->write_tier = ceph_decode_64(p); 868 } else { 869 pi->read_tier = -1; 870 pi->write_tier = -1; 871 } 872 873 if (ev >= 10) 874 /* properties */ 875 ceph_decode_skip_map(p, end, string, string, bad); 876 877 if (ev >= 11) { 878 /* hit_set_params (with versions) */ 879 ceph_decode_skip_n(p, end, 2, bad); 880 ceph_decode_skip_string(p, end, bad); 881 882 /* hit_set_period, hit_set_count */ 883 ceph_decode_skip_n(p, end, 4 + 4, bad); 884 } 885 886 if (ev >= 12) 887 /* stripe_width */ 888 ceph_decode_skip_32(p, end, bad); 889 890 if (ev >= 13) 891 /* target_max_*, cache_target_*, cache_min_* */ 892 ceph_decode_skip_n(p, end, 16 + 8 + 8, bad); 893 894 if (ev >= 14) 895 /* erasure_code_profile */ 896 ceph_decode_skip_string(p, end, bad); 897 898 /* 899 * last_force_op_resend_preluminous, will be overridden if the 900 * map was encoded with RESEND_ON_SPLIT 901 */ 902 if (ev >= 15) 903 ceph_decode_32_safe(p, end, pi->last_force_request_resend, bad); 904 else 905 pi->last_force_request_resend = 0; 906 907 if (ev >= 16) 908 /* min_read_recency_for_promote */ 909 ceph_decode_skip_32(p, end, bad); 910 911 if (ev >= 17) 912 /* expected_num_objects */ 913 ceph_decode_skip_64(p, end, bad); 914 915 if (ev >= 19) 916 /* cache_target_dirty_high_ratio_micro */ 917 ceph_decode_skip_32(p, end, bad); 918 919 if (ev >= 20) 920 /* min_write_recency_for_promote */ 921 ceph_decode_skip_32(p, end, bad); 922 923 if (ev >= 21) 924 /* use_gmt_hitset */ 925 ceph_decode_skip_8(p, end, bad); 926 927 if (ev >= 22) 928 /* fast_read */ 929 ceph_decode_skip_8(p, end, bad); 930 931 if (ev >= 23) 932 /* hit_set_grade_decay_rate, hit_set_search_last_n */ 933 ceph_decode_skip_n(p, end, 4 + 4, bad); 934 935 if (ev >= 24) { 936 /* opts (with versions) */ 937 ceph_decode_skip_n(p, end, 2, bad); 938 ceph_decode_skip_string(p, end, bad); 939 } 940 941 if (ev >= 25) 942 ceph_decode_32_safe(p, end, pi->last_force_request_resend, bad); 943 944 /* ignore the rest */ 945 946 *p = pool_end; 947 calc_pg_masks(pi); 948 return 0; 949 950 bad: 951 return -EINVAL; 952 } 953 954 static int decode_pool_names(void **p, void *end, struct ceph_osdmap *map) 955 { 956 struct ceph_pg_pool_info *pi; 957 u32 num, len; 958 u64 pool; 959 960 ceph_decode_32_safe(p, end, num, bad); 961 dout(" %d pool names\n", num); 962 while (num--) { 963 ceph_decode_64_safe(p, end, pool, bad); 964 ceph_decode_32_safe(p, end, len, bad); 965 dout(" pool %llu len %d\n", pool, len); 966 ceph_decode_need(p, end, len, bad); 967 pi = lookup_pg_pool(&map->pg_pools, pool); 968 if (pi) { 969 char *name = kstrndup(*p, len, GFP_NOFS); 970 971 if (!name) 972 return -ENOMEM; 973 kfree(pi->name); 974 pi->name = name; 975 dout(" name is %s\n", pi->name); 976 } 977 *p += len; 978 } 979 return 0; 980 981 bad: 982 return -EINVAL; 983 } 984 985 /* 986 * CRUSH workspaces 987 * 988 * workspace_manager framework borrowed from fs/btrfs/compression.c. 989 * Two simplifications: there is only one type of workspace and there 990 * is always at least one workspace. 991 */ 992 static struct crush_work *alloc_workspace(const struct crush_map *c) 993 { 994 struct crush_work *work; 995 size_t work_size; 996 997 WARN_ON(!c->working_size); 998 work_size = crush_work_size(c, CEPH_PG_MAX_SIZE); 999 dout("%s work_size %zu bytes\n", __func__, work_size); 1000 1001 work = kvmalloc(work_size, GFP_NOIO); 1002 if (!work) 1003 return NULL; 1004 1005 INIT_LIST_HEAD(&work->item); 1006 crush_init_workspace(c, work); 1007 return work; 1008 } 1009 1010 static void free_workspace(struct crush_work *work) 1011 { 1012 WARN_ON(!list_empty(&work->item)); 1013 kvfree(work); 1014 } 1015 1016 static void init_workspace_manager(struct workspace_manager *wsm) 1017 { 1018 INIT_LIST_HEAD(&wsm->idle_ws); 1019 spin_lock_init(&wsm->ws_lock); 1020 atomic_set(&wsm->total_ws, 0); 1021 wsm->free_ws = 0; 1022 init_waitqueue_head(&wsm->ws_wait); 1023 } 1024 1025 static void add_initial_workspace(struct workspace_manager *wsm, 1026 struct crush_work *work) 1027 { 1028 WARN_ON(!list_empty(&wsm->idle_ws)); 1029 1030 list_add(&work->item, &wsm->idle_ws); 1031 atomic_set(&wsm->total_ws, 1); 1032 wsm->free_ws = 1; 1033 } 1034 1035 static void cleanup_workspace_manager(struct workspace_manager *wsm) 1036 { 1037 struct crush_work *work; 1038 1039 while (!list_empty(&wsm->idle_ws)) { 1040 work = list_first_entry(&wsm->idle_ws, struct crush_work, 1041 item); 1042 list_del_init(&work->item); 1043 free_workspace(work); 1044 } 1045 atomic_set(&wsm->total_ws, 0); 1046 wsm->free_ws = 0; 1047 } 1048 1049 /* 1050 * Finds an available workspace or allocates a new one. If it's not 1051 * possible to allocate a new one, waits until there is one. 1052 */ 1053 static struct crush_work *get_workspace(struct workspace_manager *wsm, 1054 const struct crush_map *c) 1055 { 1056 struct crush_work *work; 1057 int cpus = num_online_cpus(); 1058 1059 again: 1060 spin_lock(&wsm->ws_lock); 1061 if (!list_empty(&wsm->idle_ws)) { 1062 work = list_first_entry(&wsm->idle_ws, struct crush_work, 1063 item); 1064 list_del_init(&work->item); 1065 wsm->free_ws--; 1066 spin_unlock(&wsm->ws_lock); 1067 return work; 1068 1069 } 1070 if (atomic_read(&wsm->total_ws) > cpus) { 1071 DEFINE_WAIT(wait); 1072 1073 spin_unlock(&wsm->ws_lock); 1074 prepare_to_wait(&wsm->ws_wait, &wait, TASK_UNINTERRUPTIBLE); 1075 if (atomic_read(&wsm->total_ws) > cpus && !wsm->free_ws) 1076 schedule(); 1077 finish_wait(&wsm->ws_wait, &wait); 1078 goto again; 1079 } 1080 atomic_inc(&wsm->total_ws); 1081 spin_unlock(&wsm->ws_lock); 1082 1083 work = alloc_workspace(c); 1084 if (!work) { 1085 atomic_dec(&wsm->total_ws); 1086 wake_up(&wsm->ws_wait); 1087 1088 /* 1089 * Do not return the error but go back to waiting. We 1090 * have the initial workspace and the CRUSH computation 1091 * time is bounded so we will get it eventually. 1092 */ 1093 WARN_ON(atomic_read(&wsm->total_ws) < 1); 1094 goto again; 1095 } 1096 return work; 1097 } 1098 1099 /* 1100 * Puts a workspace back on the list or frees it if we have enough 1101 * idle ones sitting around. 1102 */ 1103 static void put_workspace(struct workspace_manager *wsm, 1104 struct crush_work *work) 1105 { 1106 spin_lock(&wsm->ws_lock); 1107 if (wsm->free_ws <= num_online_cpus()) { 1108 list_add(&work->item, &wsm->idle_ws); 1109 wsm->free_ws++; 1110 spin_unlock(&wsm->ws_lock); 1111 goto wake; 1112 } 1113 spin_unlock(&wsm->ws_lock); 1114 1115 free_workspace(work); 1116 atomic_dec(&wsm->total_ws); 1117 wake: 1118 if (wq_has_sleeper(&wsm->ws_wait)) 1119 wake_up(&wsm->ws_wait); 1120 } 1121 1122 /* 1123 * osd map 1124 */ 1125 struct ceph_osdmap *ceph_osdmap_alloc(void) 1126 { 1127 struct ceph_osdmap *map; 1128 1129 map = kzalloc_obj(*map, GFP_NOIO); 1130 if (!map) 1131 return NULL; 1132 1133 map->pg_pools = RB_ROOT; 1134 map->pool_max = -1; 1135 map->pg_temp = RB_ROOT; 1136 map->primary_temp = RB_ROOT; 1137 map->pg_upmap = RB_ROOT; 1138 map->pg_upmap_items = RB_ROOT; 1139 1140 init_workspace_manager(&map->crush_wsm); 1141 1142 return map; 1143 } 1144 1145 void ceph_osdmap_destroy(struct ceph_osdmap *map) 1146 { 1147 dout("osdmap_destroy %p\n", map); 1148 1149 if (map->crush) 1150 crush_destroy(map->crush); 1151 cleanup_workspace_manager(&map->crush_wsm); 1152 1153 while (!RB_EMPTY_ROOT(&map->pg_temp)) { 1154 struct ceph_pg_mapping *pg = 1155 rb_entry(rb_first(&map->pg_temp), 1156 struct ceph_pg_mapping, node); 1157 erase_pg_mapping(&map->pg_temp, pg); 1158 free_pg_mapping(pg); 1159 } 1160 while (!RB_EMPTY_ROOT(&map->primary_temp)) { 1161 struct ceph_pg_mapping *pg = 1162 rb_entry(rb_first(&map->primary_temp), 1163 struct ceph_pg_mapping, node); 1164 erase_pg_mapping(&map->primary_temp, pg); 1165 free_pg_mapping(pg); 1166 } 1167 while (!RB_EMPTY_ROOT(&map->pg_upmap)) { 1168 struct ceph_pg_mapping *pg = 1169 rb_entry(rb_first(&map->pg_upmap), 1170 struct ceph_pg_mapping, node); 1171 rb_erase(&pg->node, &map->pg_upmap); 1172 kfree(pg); 1173 } 1174 while (!RB_EMPTY_ROOT(&map->pg_upmap_items)) { 1175 struct ceph_pg_mapping *pg = 1176 rb_entry(rb_first(&map->pg_upmap_items), 1177 struct ceph_pg_mapping, node); 1178 rb_erase(&pg->node, &map->pg_upmap_items); 1179 kfree(pg); 1180 } 1181 while (!RB_EMPTY_ROOT(&map->pg_pools)) { 1182 struct ceph_pg_pool_info *pi = 1183 rb_entry(rb_first(&map->pg_pools), 1184 struct ceph_pg_pool_info, node); 1185 __remove_pg_pool(&map->pg_pools, pi); 1186 } 1187 kvfree(map->osd_state); 1188 kvfree(map->osd_weight); 1189 kvfree(map->osd_addr); 1190 kvfree(map->osd_primary_affinity); 1191 kfree(map); 1192 } 1193 1194 /* 1195 * Adjust max_osd value, (re)allocate arrays. 1196 * 1197 * The new elements are properly initialized. 1198 */ 1199 static int osdmap_set_max_osd(struct ceph_osdmap *map, u32 max) 1200 { 1201 u32 *state; 1202 u32 *weight; 1203 struct ceph_entity_addr *addr; 1204 u32 to_copy; 1205 int i; 1206 1207 dout("%s old %u new %u\n", __func__, map->max_osd, max); 1208 if (max == map->max_osd) 1209 return 0; 1210 1211 state = kvmalloc(array_size(max, sizeof(*state)), GFP_NOFS); 1212 weight = kvmalloc(array_size(max, sizeof(*weight)), GFP_NOFS); 1213 addr = kvmalloc(array_size(max, sizeof(*addr)), GFP_NOFS); 1214 if (!state || !weight || !addr) { 1215 kvfree(state); 1216 kvfree(weight); 1217 kvfree(addr); 1218 return -ENOMEM; 1219 } 1220 1221 to_copy = min(map->max_osd, max); 1222 if (map->osd_state) { 1223 memcpy(state, map->osd_state, to_copy * sizeof(*state)); 1224 memcpy(weight, map->osd_weight, to_copy * sizeof(*weight)); 1225 memcpy(addr, map->osd_addr, to_copy * sizeof(*addr)); 1226 kvfree(map->osd_state); 1227 kvfree(map->osd_weight); 1228 kvfree(map->osd_addr); 1229 } 1230 1231 map->osd_state = state; 1232 map->osd_weight = weight; 1233 map->osd_addr = addr; 1234 for (i = map->max_osd; i < max; i++) { 1235 map->osd_state[i] = 0; 1236 map->osd_weight[i] = CEPH_OSD_OUT; 1237 memset(map->osd_addr + i, 0, sizeof(*map->osd_addr)); 1238 } 1239 1240 if (map->osd_primary_affinity) { 1241 u32 *affinity; 1242 1243 affinity = kvmalloc(array_size(max, sizeof(*affinity)), 1244 GFP_NOFS); 1245 if (!affinity) 1246 return -ENOMEM; 1247 1248 memcpy(affinity, map->osd_primary_affinity, 1249 to_copy * sizeof(*affinity)); 1250 kvfree(map->osd_primary_affinity); 1251 1252 map->osd_primary_affinity = affinity; 1253 for (i = map->max_osd; i < max; i++) 1254 map->osd_primary_affinity[i] = 1255 CEPH_OSD_DEFAULT_PRIMARY_AFFINITY; 1256 } 1257 1258 map->max_osd = max; 1259 1260 return 0; 1261 } 1262 1263 static int osdmap_set_crush(struct ceph_osdmap *map, struct crush_map *crush) 1264 { 1265 struct crush_work *work; 1266 1267 if (IS_ERR(crush)) 1268 return PTR_ERR(crush); 1269 1270 work = alloc_workspace(crush); 1271 if (!work) { 1272 crush_destroy(crush); 1273 return -ENOMEM; 1274 } 1275 1276 if (map->crush) 1277 crush_destroy(map->crush); 1278 cleanup_workspace_manager(&map->crush_wsm); 1279 map->crush = crush; 1280 add_initial_workspace(&map->crush_wsm, work); 1281 return 0; 1282 } 1283 1284 #define OSDMAP_WRAPPER_COMPAT_VER 7 1285 #define OSDMAP_CLIENT_DATA_COMPAT_VER 1 1286 1287 /* 1288 * Return 0 or error. On success, *v is set to 0 for old (v6) osdmaps, 1289 * to struct_v of the client_data section for new (v7 and above) 1290 * osdmaps. 1291 */ 1292 static int get_osdmap_client_data_v(void **p, void *end, 1293 const char *prefix, u8 *v) 1294 { 1295 u8 struct_v; 1296 1297 ceph_decode_8_safe(p, end, struct_v, e_inval); 1298 if (struct_v >= 7) { 1299 u8 struct_compat; 1300 1301 ceph_decode_8_safe(p, end, struct_compat, e_inval); 1302 if (struct_compat > OSDMAP_WRAPPER_COMPAT_VER) { 1303 pr_warn("got v %d cv %d > %d of %s ceph_osdmap\n", 1304 struct_v, struct_compat, 1305 OSDMAP_WRAPPER_COMPAT_VER, prefix); 1306 return -EINVAL; 1307 } 1308 *p += 4; /* ignore wrapper struct_len */ 1309 1310 ceph_decode_8_safe(p, end, struct_v, e_inval); 1311 ceph_decode_8_safe(p, end, struct_compat, e_inval); 1312 if (struct_compat > OSDMAP_CLIENT_DATA_COMPAT_VER) { 1313 pr_warn("got v %d cv %d > %d of %s ceph_osdmap client data\n", 1314 struct_v, struct_compat, 1315 OSDMAP_CLIENT_DATA_COMPAT_VER, prefix); 1316 return -EINVAL; 1317 } 1318 *p += 4; /* ignore client data struct_len */ 1319 } else { 1320 u16 version; 1321 1322 *p -= 1; 1323 ceph_decode_16_safe(p, end, version, e_inval); 1324 if (version < 6) { 1325 pr_warn("got v %d < 6 of %s ceph_osdmap\n", 1326 version, prefix); 1327 return -EINVAL; 1328 } 1329 1330 /* old osdmap encoding */ 1331 struct_v = 0; 1332 } 1333 1334 *v = struct_v; 1335 return 0; 1336 1337 e_inval: 1338 return -EINVAL; 1339 } 1340 1341 static int __decode_pools(void **p, void *end, struct ceph_osdmap *map, 1342 bool incremental) 1343 { 1344 u32 n; 1345 1346 ceph_decode_32_safe(p, end, n, e_inval); 1347 while (n--) { 1348 struct ceph_pg_pool_info *pi; 1349 u64 pool; 1350 int ret; 1351 1352 ceph_decode_64_safe(p, end, pool, e_inval); 1353 1354 pi = lookup_pg_pool(&map->pg_pools, pool); 1355 if (!incremental || !pi) { 1356 pi = kzalloc_obj(*pi, GFP_NOFS); 1357 if (!pi) 1358 return -ENOMEM; 1359 1360 RB_CLEAR_NODE(&pi->node); 1361 pi->id = pool; 1362 1363 if (!__insert_pg_pool(&map->pg_pools, pi)) { 1364 kfree(pi); 1365 return -EEXIST; 1366 } 1367 } 1368 1369 ret = decode_pool(p, end, pi); 1370 if (ret) 1371 return ret; 1372 } 1373 1374 return 0; 1375 1376 e_inval: 1377 return -EINVAL; 1378 } 1379 1380 static int decode_pools(void **p, void *end, struct ceph_osdmap *map) 1381 { 1382 return __decode_pools(p, end, map, false); 1383 } 1384 1385 static int decode_new_pools(void **p, void *end, struct ceph_osdmap *map) 1386 { 1387 return __decode_pools(p, end, map, true); 1388 } 1389 1390 typedef struct ceph_pg_mapping *(*decode_mapping_fn_t)(void **, void *, bool); 1391 1392 static int decode_pg_mapping(void **p, void *end, struct rb_root *mapping_root, 1393 decode_mapping_fn_t fn, bool incremental) 1394 { 1395 u32 n; 1396 1397 WARN_ON(!incremental && !fn); 1398 1399 ceph_decode_32_safe(p, end, n, e_inval); 1400 while (n--) { 1401 struct ceph_pg_mapping *pg; 1402 struct ceph_pg pgid; 1403 int ret; 1404 1405 ret = ceph_decode_pgid(p, end, &pgid); 1406 if (ret) 1407 return ret; 1408 1409 pg = lookup_pg_mapping(mapping_root, &pgid); 1410 if (pg) { 1411 WARN_ON(!incremental); 1412 erase_pg_mapping(mapping_root, pg); 1413 free_pg_mapping(pg); 1414 } 1415 1416 if (fn) { 1417 pg = fn(p, end, incremental); 1418 if (IS_ERR(pg)) 1419 return PTR_ERR(pg); 1420 1421 if (pg) { 1422 pg->pgid = pgid; /* struct */ 1423 insert_pg_mapping(mapping_root, pg); 1424 } 1425 } 1426 } 1427 1428 return 0; 1429 1430 e_inval: 1431 return -EINVAL; 1432 } 1433 1434 static struct ceph_pg_mapping *__decode_pg_temp(void **p, void *end, 1435 bool incremental) 1436 { 1437 struct ceph_pg_mapping *pg; 1438 u32 len, i; 1439 1440 ceph_decode_32_safe(p, end, len, e_inval); 1441 if (len == 0 && incremental) 1442 return NULL; /* new_pg_temp: [] to remove */ 1443 if (len > CEPH_PG_MAX_SIZE) 1444 return ERR_PTR(-EINVAL); 1445 1446 ceph_decode_need(p, end, len * sizeof(u32), e_inval); 1447 pg = alloc_pg_mapping(len * sizeof(u32)); 1448 if (!pg) 1449 return ERR_PTR(-ENOMEM); 1450 1451 pg->pg_temp.len = len; 1452 for (i = 0; i < len; i++) 1453 pg->pg_temp.osds[i] = ceph_decode_32(p); 1454 1455 return pg; 1456 1457 e_inval: 1458 return ERR_PTR(-EINVAL); 1459 } 1460 1461 static int decode_pg_temp(void **p, void *end, struct ceph_osdmap *map) 1462 { 1463 return decode_pg_mapping(p, end, &map->pg_temp, __decode_pg_temp, 1464 false); 1465 } 1466 1467 static int decode_new_pg_temp(void **p, void *end, struct ceph_osdmap *map) 1468 { 1469 return decode_pg_mapping(p, end, &map->pg_temp, __decode_pg_temp, 1470 true); 1471 } 1472 1473 static struct ceph_pg_mapping *__decode_primary_temp(void **p, void *end, 1474 bool incremental) 1475 { 1476 struct ceph_pg_mapping *pg; 1477 u32 osd; 1478 1479 ceph_decode_32_safe(p, end, osd, e_inval); 1480 if (osd == (u32)-1 && incremental) 1481 return NULL; /* new_primary_temp: -1 to remove */ 1482 1483 pg = alloc_pg_mapping(0); 1484 if (!pg) 1485 return ERR_PTR(-ENOMEM); 1486 1487 pg->primary_temp.osd = osd; 1488 return pg; 1489 1490 e_inval: 1491 return ERR_PTR(-EINVAL); 1492 } 1493 1494 static int decode_primary_temp(void **p, void *end, struct ceph_osdmap *map) 1495 { 1496 return decode_pg_mapping(p, end, &map->primary_temp, 1497 __decode_primary_temp, false); 1498 } 1499 1500 static int decode_new_primary_temp(void **p, void *end, 1501 struct ceph_osdmap *map) 1502 { 1503 return decode_pg_mapping(p, end, &map->primary_temp, 1504 __decode_primary_temp, true); 1505 } 1506 1507 u32 ceph_get_primary_affinity(struct ceph_osdmap *map, int osd) 1508 { 1509 if (!map->osd_primary_affinity) 1510 return CEPH_OSD_DEFAULT_PRIMARY_AFFINITY; 1511 1512 return map->osd_primary_affinity[osd]; 1513 } 1514 1515 static int set_primary_affinity(struct ceph_osdmap *map, int osd, u32 aff) 1516 { 1517 if (!map->osd_primary_affinity) { 1518 int i; 1519 1520 map->osd_primary_affinity = kvmalloc( 1521 array_size(map->max_osd, sizeof(*map->osd_primary_affinity)), 1522 GFP_NOFS); 1523 if (!map->osd_primary_affinity) 1524 return -ENOMEM; 1525 1526 for (i = 0; i < map->max_osd; i++) 1527 map->osd_primary_affinity[i] = 1528 CEPH_OSD_DEFAULT_PRIMARY_AFFINITY; 1529 } 1530 1531 map->osd_primary_affinity[osd] = aff; 1532 1533 return 0; 1534 } 1535 1536 static int decode_primary_affinity(void **p, void *end, 1537 struct ceph_osdmap *map) 1538 { 1539 u32 len, i; 1540 1541 ceph_decode_32_safe(p, end, len, e_inval); 1542 if (len == 0) { 1543 kvfree(map->osd_primary_affinity); 1544 map->osd_primary_affinity = NULL; 1545 return 0; 1546 } 1547 if (len != map->max_osd) 1548 goto e_inval; 1549 1550 ceph_decode_need(p, end, map->max_osd*sizeof(u32), e_inval); 1551 1552 for (i = 0; i < map->max_osd; i++) { 1553 int ret; 1554 1555 ret = set_primary_affinity(map, i, ceph_decode_32(p)); 1556 if (ret) 1557 return ret; 1558 } 1559 1560 return 0; 1561 1562 e_inval: 1563 return -EINVAL; 1564 } 1565 1566 static int decode_new_primary_affinity(void **p, void *end, 1567 struct ceph_osdmap *map) 1568 { 1569 u32 n; 1570 1571 ceph_decode_32_safe(p, end, n, e_inval); 1572 while (n--) { 1573 u32 osd, aff; 1574 int ret; 1575 1576 ceph_decode_32_safe(p, end, osd, e_inval); 1577 ceph_decode_32_safe(p, end, aff, e_inval); 1578 if (osd >= map->max_osd) 1579 goto e_inval; 1580 1581 ret = set_primary_affinity(map, osd, aff); 1582 if (ret) 1583 return ret; 1584 1585 osdmap_info(map, "osd%d primary-affinity 0x%x\n", osd, aff); 1586 } 1587 1588 return 0; 1589 1590 e_inval: 1591 return -EINVAL; 1592 } 1593 1594 static struct ceph_pg_mapping *__decode_pg_upmap(void **p, void *end, 1595 bool __unused) 1596 { 1597 return __decode_pg_temp(p, end, false); 1598 } 1599 1600 static int decode_pg_upmap(void **p, void *end, struct ceph_osdmap *map) 1601 { 1602 return decode_pg_mapping(p, end, &map->pg_upmap, __decode_pg_upmap, 1603 false); 1604 } 1605 1606 static int decode_new_pg_upmap(void **p, void *end, struct ceph_osdmap *map) 1607 { 1608 return decode_pg_mapping(p, end, &map->pg_upmap, __decode_pg_upmap, 1609 true); 1610 } 1611 1612 static int decode_old_pg_upmap(void **p, void *end, struct ceph_osdmap *map) 1613 { 1614 return decode_pg_mapping(p, end, &map->pg_upmap, NULL, true); 1615 } 1616 1617 static struct ceph_pg_mapping *__decode_pg_upmap_items(void **p, void *end, 1618 bool __unused) 1619 { 1620 struct ceph_pg_mapping *pg; 1621 u32 len, i; 1622 1623 ceph_decode_32_safe(p, end, len, e_inval); 1624 if (len > CEPH_PG_MAX_SIZE) 1625 return ERR_PTR(-EINVAL); 1626 1627 ceph_decode_need(p, end, 2 * len * sizeof(u32), e_inval); 1628 pg = alloc_pg_mapping(2 * len * sizeof(u32)); 1629 if (!pg) 1630 return ERR_PTR(-ENOMEM); 1631 1632 pg->pg_upmap_items.len = len; 1633 for (i = 0; i < len; i++) { 1634 pg->pg_upmap_items.from_to[i][0] = ceph_decode_32(p); 1635 pg->pg_upmap_items.from_to[i][1] = ceph_decode_32(p); 1636 } 1637 1638 return pg; 1639 1640 e_inval: 1641 return ERR_PTR(-EINVAL); 1642 } 1643 1644 static int decode_pg_upmap_items(void **p, void *end, struct ceph_osdmap *map) 1645 { 1646 return decode_pg_mapping(p, end, &map->pg_upmap_items, 1647 __decode_pg_upmap_items, false); 1648 } 1649 1650 static int decode_new_pg_upmap_items(void **p, void *end, 1651 struct ceph_osdmap *map) 1652 { 1653 return decode_pg_mapping(p, end, &map->pg_upmap_items, 1654 __decode_pg_upmap_items, true); 1655 } 1656 1657 static int decode_old_pg_upmap_items(void **p, void *end, 1658 struct ceph_osdmap *map) 1659 { 1660 return decode_pg_mapping(p, end, &map->pg_upmap_items, NULL, true); 1661 } 1662 1663 /* 1664 * decode a full map. 1665 */ 1666 static int osdmap_decode(void **p, void *end, bool msgr2, 1667 struct ceph_osdmap *map) 1668 { 1669 u8 struct_v; 1670 u32 epoch = 0; 1671 void *start = *p; 1672 u32 max; 1673 u32 len, i; 1674 int err; 1675 1676 dout("%s %p to %p len %d\n", __func__, *p, end, (int)(end - *p)); 1677 1678 err = get_osdmap_client_data_v(p, end, "full", &struct_v); 1679 if (err) 1680 goto bad; 1681 1682 /* fsid, epoch, created, modified */ 1683 ceph_decode_need(p, end, sizeof(map->fsid) + sizeof(u32) + 1684 sizeof(map->created) + sizeof(map->modified), e_inval); 1685 ceph_decode_copy(p, &map->fsid, sizeof(map->fsid)); 1686 epoch = map->epoch = ceph_decode_32(p); 1687 ceph_decode_copy(p, &map->created, sizeof(map->created)); 1688 ceph_decode_copy(p, &map->modified, sizeof(map->modified)); 1689 1690 /* pools */ 1691 err = decode_pools(p, end, map); 1692 if (err) 1693 goto bad; 1694 1695 /* pool_name */ 1696 err = decode_pool_names(p, end, map); 1697 if (err) 1698 goto bad; 1699 1700 ceph_decode_32_safe(p, end, map->pool_max, e_inval); 1701 1702 ceph_decode_32_safe(p, end, map->flags, e_inval); 1703 1704 /* max_osd */ 1705 ceph_decode_32_safe(p, end, max, e_inval); 1706 1707 /* (re)alloc osd arrays */ 1708 err = osdmap_set_max_osd(map, max); 1709 if (err) 1710 goto bad; 1711 1712 /* osd_state, osd_weight, osd_addrs->client_addr */ 1713 ceph_decode_need(p, end, 3*sizeof(u32) + 1714 map->max_osd*(struct_v >= 5 ? sizeof(u32) : 1715 sizeof(u8)) + 1716 map->max_osd*sizeof(*map->osd_weight), e_inval); 1717 if (ceph_decode_32(p) != map->max_osd) 1718 goto e_inval; 1719 1720 if (struct_v >= 5) { 1721 for (i = 0; i < map->max_osd; i++) 1722 map->osd_state[i] = ceph_decode_32(p); 1723 } else { 1724 for (i = 0; i < map->max_osd; i++) 1725 map->osd_state[i] = ceph_decode_8(p); 1726 } 1727 1728 if (ceph_decode_32(p) != map->max_osd) 1729 goto e_inval; 1730 1731 for (i = 0; i < map->max_osd; i++) 1732 map->osd_weight[i] = ceph_decode_32(p); 1733 1734 if (ceph_decode_32(p) != map->max_osd) 1735 goto e_inval; 1736 1737 for (i = 0; i < map->max_osd; i++) { 1738 struct ceph_entity_addr *addr = &map->osd_addr[i]; 1739 1740 if (struct_v >= 8) 1741 err = ceph_decode_entity_addrvec(p, end, msgr2, addr); 1742 else 1743 err = ceph_decode_entity_addr(p, end, addr); 1744 if (err) 1745 goto bad; 1746 1747 dout("%s osd%d addr %s\n", __func__, i, ceph_pr_addr(addr)); 1748 } 1749 1750 /* pg_temp */ 1751 err = decode_pg_temp(p, end, map); 1752 if (err) 1753 goto bad; 1754 1755 /* primary_temp */ 1756 if (struct_v >= 1) { 1757 err = decode_primary_temp(p, end, map); 1758 if (err) 1759 goto bad; 1760 } 1761 1762 /* primary_affinity */ 1763 if (struct_v >= 2) { 1764 err = decode_primary_affinity(p, end, map); 1765 if (err) 1766 goto bad; 1767 } else { 1768 WARN_ON(map->osd_primary_affinity); 1769 } 1770 1771 /* crush */ 1772 ceph_decode_32_safe(p, end, len, e_inval); 1773 err = osdmap_set_crush(map, crush_decode(*p, min(*p + len, end))); 1774 if (err) 1775 goto bad; 1776 1777 *p += len; 1778 if (struct_v >= 3) { 1779 /* erasure_code_profiles */ 1780 ceph_decode_skip_map_of_map(p, end, string, string, string, 1781 e_inval); 1782 } 1783 1784 if (struct_v >= 4) { 1785 err = decode_pg_upmap(p, end, map); 1786 if (err) 1787 goto bad; 1788 1789 err = decode_pg_upmap_items(p, end, map); 1790 if (err) 1791 goto bad; 1792 } else { 1793 WARN_ON(!RB_EMPTY_ROOT(&map->pg_upmap)); 1794 WARN_ON(!RB_EMPTY_ROOT(&map->pg_upmap_items)); 1795 } 1796 1797 /* ignore the rest */ 1798 *p = end; 1799 1800 dout("full osdmap epoch %d max_osd %d\n", map->epoch, map->max_osd); 1801 return 0; 1802 1803 e_inval: 1804 err = -EINVAL; 1805 bad: 1806 pr_err("corrupt full osdmap (%d) epoch %d off %d (%p of %p-%p)\n", 1807 err, epoch, (int)(*p - start), *p, start, end); 1808 print_hex_dump(KERN_DEBUG, "osdmap: ", 1809 DUMP_PREFIX_OFFSET, 16, 1, 1810 start, end - start, true); 1811 return err; 1812 } 1813 1814 /* 1815 * Allocate and decode a full map. 1816 */ 1817 struct ceph_osdmap *ceph_osdmap_decode(void **p, void *end, bool msgr2) 1818 { 1819 struct ceph_osdmap *map; 1820 int ret; 1821 1822 map = ceph_osdmap_alloc(); 1823 if (!map) 1824 return ERR_PTR(-ENOMEM); 1825 1826 ret = osdmap_decode(p, end, msgr2, map); 1827 if (ret) { 1828 ceph_osdmap_destroy(map); 1829 return ERR_PTR(ret); 1830 } 1831 1832 return map; 1833 } 1834 1835 /* 1836 * Encoding order is (new_up_client, new_state, new_weight). Need to 1837 * apply in the (new_weight, new_state, new_up_client) order, because 1838 * an incremental map may look like e.g. 1839 * 1840 * new_up_client: { osd=6, addr=... } # set osd_state and addr 1841 * new_state: { osd=6, xorstate=EXISTS } # clear osd_state 1842 */ 1843 static int decode_new_up_state_weight(void **p, void *end, u8 struct_v, 1844 bool msgr2, struct ceph_osdmap *map) 1845 { 1846 void *new_up_client; 1847 void *new_state; 1848 void *new_weight_end; 1849 const u32 new_state_item_size = 1850 sizeof(u32) + (struct_v >= 5 ? sizeof(u32) : sizeof(u8)); 1851 u32 len; 1852 int ret; 1853 int i; 1854 1855 new_up_client = *p; 1856 ceph_decode_32_safe(p, end, len, e_inval); 1857 for (i = 0; i < len; ++i) { 1858 struct ceph_entity_addr addr; 1859 1860 ceph_decode_skip_32(p, end, e_inval); 1861 if (struct_v >= 7) 1862 ret = ceph_decode_entity_addrvec(p, end, msgr2, &addr); 1863 else 1864 ret = ceph_decode_entity_addr(p, end, &addr); 1865 if (ret) 1866 return ret; 1867 } 1868 1869 new_state = *p; 1870 ceph_decode_32_safe(p, end, len, e_inval); 1871 if (check_mul_overflow(len, new_state_item_size, &len)) 1872 goto e_inval; 1873 ceph_decode_need(p, end, len, e_inval); 1874 *p += len; 1875 1876 /* new_weight */ 1877 ceph_decode_32_safe(p, end, len, e_inval); 1878 while (len--) { 1879 s32 osd; 1880 u32 w; 1881 1882 ceph_decode_need(p, end, 2*sizeof(u32), e_inval); 1883 osd = ceph_decode_32(p); 1884 w = ceph_decode_32(p); 1885 if (osd >= map->max_osd) 1886 goto e_inval; 1887 1888 osdmap_info(map, "osd%d weight 0x%x %s\n", osd, w, 1889 w == CEPH_OSD_IN ? "(in)" : 1890 (w == CEPH_OSD_OUT ? "(out)" : "")); 1891 map->osd_weight[osd] = w; 1892 1893 /* 1894 * If we are marking in, set the EXISTS, and clear the 1895 * AUTOOUT and NEW bits. 1896 */ 1897 if (w) { 1898 map->osd_state[osd] |= CEPH_OSD_EXISTS; 1899 map->osd_state[osd] &= ~(CEPH_OSD_AUTOOUT | 1900 CEPH_OSD_NEW); 1901 } 1902 } 1903 new_weight_end = *p; 1904 1905 /* new_state (up/down) */ 1906 *p = new_state; 1907 len = ceph_decode_32(p); 1908 while (len--) { 1909 s32 osd; 1910 u32 xorstate; 1911 1912 osd = ceph_decode_32(p); 1913 if (osd >= map->max_osd) 1914 goto e_inval; 1915 1916 if (struct_v >= 5) 1917 xorstate = ceph_decode_32(p); 1918 else 1919 xorstate = ceph_decode_8(p); 1920 if (xorstate == 0) 1921 xorstate = CEPH_OSD_UP; 1922 if ((map->osd_state[osd] & CEPH_OSD_UP) && 1923 (xorstate & CEPH_OSD_UP)) 1924 osdmap_info(map, "osd%d down\n", osd); 1925 if ((map->osd_state[osd] & CEPH_OSD_EXISTS) && 1926 (xorstate & CEPH_OSD_EXISTS)) { 1927 osdmap_info(map, "osd%d does not exist\n", osd); 1928 ret = set_primary_affinity(map, osd, 1929 CEPH_OSD_DEFAULT_PRIMARY_AFFINITY); 1930 if (ret) 1931 return ret; 1932 memset(map->osd_addr + osd, 0, sizeof(*map->osd_addr)); 1933 map->osd_state[osd] = 0; 1934 } else { 1935 map->osd_state[osd] ^= xorstate; 1936 } 1937 } 1938 1939 /* new_up_client */ 1940 *p = new_up_client; 1941 len = ceph_decode_32(p); 1942 while (len--) { 1943 s32 osd; 1944 struct ceph_entity_addr addr; 1945 1946 osd = ceph_decode_32(p); 1947 if (osd >= map->max_osd) 1948 goto e_inval; 1949 1950 if (struct_v >= 7) 1951 ret = ceph_decode_entity_addrvec(p, end, msgr2, &addr); 1952 else 1953 ret = ceph_decode_entity_addr(p, end, &addr); 1954 if (ret) 1955 return ret; 1956 1957 dout("%s osd%d addr %s\n", __func__, osd, ceph_pr_addr(&addr)); 1958 1959 osdmap_info(map, "osd%d up\n", osd); 1960 map->osd_state[osd] |= CEPH_OSD_EXISTS | CEPH_OSD_UP; 1961 map->osd_addr[osd] = addr; 1962 } 1963 1964 *p = new_weight_end; 1965 return 0; 1966 1967 e_inval: 1968 return -EINVAL; 1969 } 1970 1971 /* 1972 * decode and apply an incremental map update. 1973 */ 1974 struct ceph_osdmap *osdmap_apply_incremental(void **p, void *end, bool msgr2, 1975 struct ceph_osdmap *map) 1976 { 1977 struct ceph_fsid fsid; 1978 u32 epoch = 0; 1979 struct ceph_timespec modified; 1980 s32 len; 1981 u64 pool; 1982 __s64 new_pool_max; 1983 __s32 new_flags, max; 1984 void *start = *p; 1985 int err; 1986 u8 struct_v; 1987 1988 dout("%s %p to %p len %d\n", __func__, *p, end, (int)(end - *p)); 1989 1990 err = get_osdmap_client_data_v(p, end, "inc", &struct_v); 1991 if (err) 1992 goto bad; 1993 1994 /* fsid, epoch, modified, new_pool_max, new_flags */ 1995 ceph_decode_need(p, end, sizeof(fsid) + sizeof(u32) + sizeof(modified) + 1996 sizeof(u64) + sizeof(u32), e_inval); 1997 ceph_decode_copy(p, &fsid, sizeof(fsid)); 1998 epoch = ceph_decode_32(p); 1999 ceph_decode_copy(p, &modified, sizeof(modified)); 2000 new_pool_max = ceph_decode_64(p); 2001 new_flags = ceph_decode_32(p); 2002 2003 if (epoch != map->epoch + 1) 2004 goto e_inval; 2005 2006 /* full map? */ 2007 ceph_decode_32_safe(p, end, len, e_inval); 2008 if (len > 0) { 2009 dout("apply_incremental full map len %d, %p to %p\n", 2010 len, *p, end); 2011 return ceph_osdmap_decode(p, min(*p+len, end), msgr2); 2012 } 2013 2014 /* new crush? */ 2015 ceph_decode_32_safe(p, end, len, e_inval); 2016 if (len > 0) { 2017 err = osdmap_set_crush(map, 2018 crush_decode(*p, min(*p + len, end))); 2019 if (err) 2020 goto bad; 2021 *p += len; 2022 } 2023 2024 /* new flags? */ 2025 if (new_flags >= 0) 2026 map->flags = new_flags; 2027 if (new_pool_max >= 0) 2028 map->pool_max = new_pool_max; 2029 2030 /* new max? */ 2031 ceph_decode_32_safe(p, end, max, e_inval); 2032 if (max >= 0) { 2033 err = osdmap_set_max_osd(map, max); 2034 if (err) 2035 goto bad; 2036 } 2037 2038 map->epoch++; 2039 map->modified = modified; 2040 2041 /* new_pools */ 2042 err = decode_new_pools(p, end, map); 2043 if (err) 2044 goto bad; 2045 2046 /* new_pool_names */ 2047 err = decode_pool_names(p, end, map); 2048 if (err) 2049 goto bad; 2050 2051 /* old_pool */ 2052 ceph_decode_32_safe(p, end, len, e_inval); 2053 while (len--) { 2054 struct ceph_pg_pool_info *pi; 2055 2056 ceph_decode_64_safe(p, end, pool, e_inval); 2057 pi = lookup_pg_pool(&map->pg_pools, pool); 2058 if (pi) 2059 __remove_pg_pool(&map->pg_pools, pi); 2060 } 2061 2062 /* new_up_client, new_state, new_weight */ 2063 err = decode_new_up_state_weight(p, end, struct_v, msgr2, map); 2064 if (err) 2065 goto bad; 2066 2067 /* new_pg_temp */ 2068 err = decode_new_pg_temp(p, end, map); 2069 if (err) 2070 goto bad; 2071 2072 /* new_primary_temp */ 2073 if (struct_v >= 1) { 2074 err = decode_new_primary_temp(p, end, map); 2075 if (err) 2076 goto bad; 2077 } 2078 2079 /* new_primary_affinity */ 2080 if (struct_v >= 2) { 2081 err = decode_new_primary_affinity(p, end, map); 2082 if (err) 2083 goto bad; 2084 } 2085 2086 if (struct_v >= 3) { 2087 /* new_erasure_code_profiles */ 2088 ceph_decode_skip_map_of_map(p, end, string, string, string, 2089 e_inval); 2090 /* old_erasure_code_profiles */ 2091 ceph_decode_skip_set(p, end, string, e_inval); 2092 } 2093 2094 if (struct_v >= 4) { 2095 err = decode_new_pg_upmap(p, end, map); 2096 if (err) 2097 goto bad; 2098 2099 err = decode_old_pg_upmap(p, end, map); 2100 if (err) 2101 goto bad; 2102 2103 err = decode_new_pg_upmap_items(p, end, map); 2104 if (err) 2105 goto bad; 2106 2107 err = decode_old_pg_upmap_items(p, end, map); 2108 if (err) 2109 goto bad; 2110 } 2111 2112 /* ignore the rest */ 2113 *p = end; 2114 2115 dout("inc osdmap epoch %d max_osd %d\n", map->epoch, map->max_osd); 2116 return map; 2117 2118 e_inval: 2119 err = -EINVAL; 2120 bad: 2121 pr_err("corrupt inc osdmap (%d) epoch %d off %d (%p of %p-%p)\n", 2122 err, epoch, (int)(*p - start), *p, start, end); 2123 print_hex_dump(KERN_DEBUG, "osdmap: ", 2124 DUMP_PREFIX_OFFSET, 16, 1, 2125 start, end - start, true); 2126 return ERR_PTR(err); 2127 } 2128 2129 void ceph_oloc_copy(struct ceph_object_locator *dest, 2130 const struct ceph_object_locator *src) 2131 { 2132 ceph_oloc_destroy(dest); 2133 2134 dest->pool = src->pool; 2135 if (src->pool_ns) 2136 dest->pool_ns = ceph_get_string(src->pool_ns); 2137 else 2138 dest->pool_ns = NULL; 2139 } 2140 EXPORT_SYMBOL(ceph_oloc_copy); 2141 2142 void ceph_oloc_destroy(struct ceph_object_locator *oloc) 2143 { 2144 ceph_put_string(oloc->pool_ns); 2145 } 2146 EXPORT_SYMBOL(ceph_oloc_destroy); 2147 2148 void ceph_oid_copy(struct ceph_object_id *dest, 2149 const struct ceph_object_id *src) 2150 { 2151 ceph_oid_destroy(dest); 2152 2153 if (src->name != src->inline_name) { 2154 /* very rare, see ceph_object_id definition */ 2155 dest->name = kmalloc(src->name_len + 1, 2156 GFP_NOIO | __GFP_NOFAIL); 2157 } else { 2158 dest->name = dest->inline_name; 2159 } 2160 memcpy(dest->name, src->name, src->name_len + 1); 2161 dest->name_len = src->name_len; 2162 } 2163 EXPORT_SYMBOL(ceph_oid_copy); 2164 2165 static __printf(2, 0) 2166 int oid_printf_vargs(struct ceph_object_id *oid, const char *fmt, va_list ap) 2167 { 2168 int len; 2169 2170 WARN_ON(!ceph_oid_empty(oid)); 2171 2172 len = vsnprintf(oid->inline_name, sizeof(oid->inline_name), fmt, ap); 2173 if (len >= sizeof(oid->inline_name)) 2174 return len; 2175 2176 oid->name_len = len; 2177 return 0; 2178 } 2179 2180 /* 2181 * If oid doesn't fit into inline buffer, BUG. 2182 */ 2183 void ceph_oid_printf(struct ceph_object_id *oid, const char *fmt, ...) 2184 { 2185 va_list ap; 2186 2187 va_start(ap, fmt); 2188 BUG_ON(oid_printf_vargs(oid, fmt, ap)); 2189 va_end(ap); 2190 } 2191 EXPORT_SYMBOL(ceph_oid_printf); 2192 2193 static __printf(3, 0) 2194 int oid_aprintf_vargs(struct ceph_object_id *oid, gfp_t gfp, 2195 const char *fmt, va_list ap) 2196 { 2197 va_list aq; 2198 int len; 2199 2200 va_copy(aq, ap); 2201 len = oid_printf_vargs(oid, fmt, aq); 2202 va_end(aq); 2203 2204 if (len) { 2205 char *external_name; 2206 2207 external_name = kmalloc(len + 1, gfp); 2208 if (!external_name) 2209 return -ENOMEM; 2210 2211 oid->name = external_name; 2212 WARN_ON(vsnprintf(oid->name, len + 1, fmt, ap) != len); 2213 oid->name_len = len; 2214 } 2215 2216 return 0; 2217 } 2218 2219 /* 2220 * If oid doesn't fit into inline buffer, allocate. 2221 */ 2222 int ceph_oid_aprintf(struct ceph_object_id *oid, gfp_t gfp, 2223 const char *fmt, ...) 2224 { 2225 va_list ap; 2226 int ret; 2227 2228 va_start(ap, fmt); 2229 ret = oid_aprintf_vargs(oid, gfp, fmt, ap); 2230 va_end(ap); 2231 2232 return ret; 2233 } 2234 EXPORT_SYMBOL(ceph_oid_aprintf); 2235 2236 void ceph_oid_destroy(struct ceph_object_id *oid) 2237 { 2238 if (oid->name != oid->inline_name) 2239 kfree(oid->name); 2240 } 2241 EXPORT_SYMBOL(ceph_oid_destroy); 2242 2243 /* 2244 * osds only 2245 */ 2246 static bool __osds_equal(const struct ceph_osds *lhs, 2247 const struct ceph_osds *rhs) 2248 { 2249 if (lhs->size == rhs->size && 2250 !memcmp(lhs->osds, rhs->osds, rhs->size * sizeof(rhs->osds[0]))) 2251 return true; 2252 2253 return false; 2254 } 2255 2256 /* 2257 * osds + primary 2258 */ 2259 static bool osds_equal(const struct ceph_osds *lhs, 2260 const struct ceph_osds *rhs) 2261 { 2262 if (__osds_equal(lhs, rhs) && 2263 lhs->primary == rhs->primary) 2264 return true; 2265 2266 return false; 2267 } 2268 2269 static bool osds_valid(const struct ceph_osds *set) 2270 { 2271 /* non-empty set */ 2272 if (set->size > 0 && set->primary >= 0) 2273 return true; 2274 2275 /* empty can_shift_osds set */ 2276 if (!set->size && set->primary == -1) 2277 return true; 2278 2279 /* empty !can_shift_osds set - all NONE */ 2280 if (set->size > 0 && set->primary == -1) { 2281 int i; 2282 2283 for (i = 0; i < set->size; i++) { 2284 if (set->osds[i] != CRUSH_ITEM_NONE) 2285 break; 2286 } 2287 if (i == set->size) 2288 return true; 2289 } 2290 2291 return false; 2292 } 2293 2294 void ceph_osds_copy(struct ceph_osds *dest, const struct ceph_osds *src) 2295 { 2296 memcpy(dest->osds, src->osds, src->size * sizeof(src->osds[0])); 2297 dest->size = src->size; 2298 dest->primary = src->primary; 2299 } 2300 2301 bool ceph_pg_is_split(const struct ceph_pg *pgid, u32 old_pg_num, 2302 u32 new_pg_num) 2303 { 2304 int old_bits = calc_bits_of(old_pg_num); 2305 int old_mask = (1 << old_bits) - 1; 2306 int n; 2307 2308 WARN_ON(pgid->seed >= old_pg_num); 2309 if (new_pg_num <= old_pg_num) 2310 return false; 2311 2312 for (n = 1; ; n++) { 2313 int next_bit = n << (old_bits - 1); 2314 u32 s = next_bit | pgid->seed; 2315 2316 if (s < old_pg_num || s == pgid->seed) 2317 continue; 2318 if (s >= new_pg_num) 2319 break; 2320 2321 s = ceph_stable_mod(s, old_pg_num, old_mask); 2322 if (s == pgid->seed) 2323 return true; 2324 } 2325 2326 return false; 2327 } 2328 2329 bool ceph_is_new_interval(const struct ceph_osds *old_acting, 2330 const struct ceph_osds *new_acting, 2331 const struct ceph_osds *old_up, 2332 const struct ceph_osds *new_up, 2333 int old_size, 2334 int new_size, 2335 int old_min_size, 2336 int new_min_size, 2337 u32 old_pg_num, 2338 u32 new_pg_num, 2339 bool old_sort_bitwise, 2340 bool new_sort_bitwise, 2341 bool old_recovery_deletes, 2342 bool new_recovery_deletes, 2343 const struct ceph_pg *pgid) 2344 { 2345 return !osds_equal(old_acting, new_acting) || 2346 !osds_equal(old_up, new_up) || 2347 old_size != new_size || 2348 old_min_size != new_min_size || 2349 ceph_pg_is_split(pgid, old_pg_num, new_pg_num) || 2350 old_sort_bitwise != new_sort_bitwise || 2351 old_recovery_deletes != new_recovery_deletes; 2352 } 2353 2354 static int calc_pg_rank(int osd, const struct ceph_osds *acting) 2355 { 2356 int i; 2357 2358 for (i = 0; i < acting->size; i++) { 2359 if (acting->osds[i] == osd) 2360 return i; 2361 } 2362 2363 return -1; 2364 } 2365 2366 static bool primary_changed(const struct ceph_osds *old_acting, 2367 const struct ceph_osds *new_acting) 2368 { 2369 if (!old_acting->size && !new_acting->size) 2370 return false; /* both still empty */ 2371 2372 if (!old_acting->size ^ !new_acting->size) 2373 return true; /* was empty, now not, or vice versa */ 2374 2375 if (old_acting->primary != new_acting->primary) 2376 return true; /* primary changed */ 2377 2378 if (calc_pg_rank(old_acting->primary, old_acting) != 2379 calc_pg_rank(new_acting->primary, new_acting)) 2380 return true; 2381 2382 return false; /* same primary (tho replicas may have changed) */ 2383 } 2384 2385 bool ceph_osds_changed(const struct ceph_osds *old_acting, 2386 const struct ceph_osds *new_acting, 2387 bool any_change) 2388 { 2389 if (primary_changed(old_acting, new_acting)) 2390 return true; 2391 2392 if (any_change && !__osds_equal(old_acting, new_acting)) 2393 return true; 2394 2395 return false; 2396 } 2397 2398 /* 2399 * Map an object into a PG. 2400 * 2401 * Should only be called with target_oid and target_oloc (as opposed to 2402 * base_oid and base_oloc), since tiering isn't taken into account. 2403 */ 2404 void __ceph_object_locator_to_pg(struct ceph_pg_pool_info *pi, 2405 const struct ceph_object_id *oid, 2406 const struct ceph_object_locator *oloc, 2407 struct ceph_pg *raw_pgid) 2408 { 2409 WARN_ON(pi->id != oloc->pool); 2410 2411 if (!oloc->pool_ns) { 2412 raw_pgid->pool = oloc->pool; 2413 raw_pgid->seed = ceph_str_hash(pi->object_hash, oid->name, 2414 oid->name_len); 2415 dout("%s %s -> raw_pgid %llu.%x\n", __func__, oid->name, 2416 raw_pgid->pool, raw_pgid->seed); 2417 } else { 2418 char stack_buf[256]; 2419 char *buf = stack_buf; 2420 int nsl = oloc->pool_ns->len; 2421 size_t total = nsl + 1 + oid->name_len; 2422 2423 if (total > sizeof(stack_buf)) 2424 buf = kmalloc(total, GFP_NOIO | __GFP_NOFAIL); 2425 memcpy(buf, oloc->pool_ns->str, nsl); 2426 buf[nsl] = '\037'; 2427 memcpy(buf + nsl + 1, oid->name, oid->name_len); 2428 raw_pgid->pool = oloc->pool; 2429 raw_pgid->seed = ceph_str_hash(pi->object_hash, buf, total); 2430 if (buf != stack_buf) 2431 kfree(buf); 2432 dout("%s %s ns %.*s -> raw_pgid %llu.%x\n", __func__, 2433 oid->name, nsl, oloc->pool_ns->str, 2434 raw_pgid->pool, raw_pgid->seed); 2435 } 2436 } 2437 2438 int ceph_object_locator_to_pg(struct ceph_osdmap *osdmap, 2439 const struct ceph_object_id *oid, 2440 const struct ceph_object_locator *oloc, 2441 struct ceph_pg *raw_pgid) 2442 { 2443 struct ceph_pg_pool_info *pi; 2444 2445 pi = ceph_pg_pool_by_id(osdmap, oloc->pool); 2446 if (!pi) 2447 return -ENOENT; 2448 2449 __ceph_object_locator_to_pg(pi, oid, oloc, raw_pgid); 2450 return 0; 2451 } 2452 EXPORT_SYMBOL(ceph_object_locator_to_pg); 2453 2454 /* 2455 * Map a raw PG (full precision ps) into an actual PG. 2456 */ 2457 static void raw_pg_to_pg(struct ceph_pg_pool_info *pi, 2458 const struct ceph_pg *raw_pgid, 2459 struct ceph_pg *pgid) 2460 { 2461 pgid->pool = raw_pgid->pool; 2462 pgid->seed = ceph_stable_mod(raw_pgid->seed, pi->pg_num, 2463 pi->pg_num_mask); 2464 } 2465 2466 /* 2467 * Map a raw PG (full precision ps) into a placement ps (placement 2468 * seed). Include pool id in that value so that different pools don't 2469 * use the same seeds. 2470 */ 2471 static u32 raw_pg_to_pps(struct ceph_pg_pool_info *pi, 2472 const struct ceph_pg *raw_pgid) 2473 { 2474 if (pi->flags & CEPH_POOL_FLAG_HASHPSPOOL) { 2475 /* hash pool id and seed so that pool PGs do not overlap */ 2476 return crush_hash32_2(CRUSH_HASH_RJENKINS1, 2477 ceph_stable_mod(raw_pgid->seed, 2478 pi->pgp_num, 2479 pi->pgp_num_mask), 2480 raw_pgid->pool); 2481 } else { 2482 /* 2483 * legacy behavior: add ps and pool together. this is 2484 * not a great approach because the PGs from each pool 2485 * will overlap on top of each other: 0.5 == 1.4 == 2486 * 2.3 == ... 2487 */ 2488 return ceph_stable_mod(raw_pgid->seed, pi->pgp_num, 2489 pi->pgp_num_mask) + 2490 (unsigned)raw_pgid->pool; 2491 } 2492 } 2493 2494 /* 2495 * Magic value used for a "default" fallback choose_args, used if the 2496 * crush_choose_arg_map passed to do_crush() does not exist. If this 2497 * also doesn't exist, fall back to canonical weights. 2498 */ 2499 #define CEPH_DEFAULT_CHOOSE_ARGS -1 2500 2501 static int do_crush(struct ceph_osdmap *map, int ruleno, int x, 2502 int *result, int result_max, 2503 const __u32 *weight, int weight_max, 2504 s64 choose_args_index) 2505 { 2506 struct crush_choose_arg_map *arg_map; 2507 struct crush_work *work; 2508 int r; 2509 2510 BUG_ON(result_max > CEPH_PG_MAX_SIZE); 2511 2512 arg_map = lookup_choose_arg_map(&map->crush->choose_args, 2513 choose_args_index); 2514 if (!arg_map) 2515 arg_map = lookup_choose_arg_map(&map->crush->choose_args, 2516 CEPH_DEFAULT_CHOOSE_ARGS); 2517 2518 work = get_workspace(&map->crush_wsm, map->crush); 2519 r = crush_do_rule(map->crush, ruleno, x, result, result_max, 2520 weight, weight_max, work, 2521 arg_map ? arg_map->args : NULL); 2522 put_workspace(&map->crush_wsm, work); 2523 return r; 2524 } 2525 2526 static void remove_nonexistent_osds(struct ceph_osdmap *osdmap, 2527 struct ceph_pg_pool_info *pi, 2528 struct ceph_osds *set) 2529 { 2530 int i; 2531 2532 if (ceph_can_shift_osds(pi)) { 2533 int removed = 0; 2534 2535 /* shift left */ 2536 for (i = 0; i < set->size; i++) { 2537 if (!ceph_osd_exists(osdmap, set->osds[i])) { 2538 removed++; 2539 continue; 2540 } 2541 if (removed) 2542 set->osds[i - removed] = set->osds[i]; 2543 } 2544 set->size -= removed; 2545 } else { 2546 /* set dne devices to NONE */ 2547 for (i = 0; i < set->size; i++) { 2548 if (!ceph_osd_exists(osdmap, set->osds[i])) 2549 set->osds[i] = CRUSH_ITEM_NONE; 2550 } 2551 } 2552 } 2553 2554 /* 2555 * Calculate raw set (CRUSH output) for given PG and filter out 2556 * nonexistent OSDs. ->primary is undefined for a raw set. 2557 * 2558 * Placement seed (CRUSH input) is returned through @ppps. 2559 */ 2560 static void pg_to_raw_osds(struct ceph_osdmap *osdmap, 2561 struct ceph_pg_pool_info *pi, 2562 const struct ceph_pg *raw_pgid, 2563 struct ceph_osds *raw, 2564 u32 *ppps) 2565 { 2566 u32 pps = raw_pg_to_pps(pi, raw_pgid); 2567 int ruleno; 2568 int len; 2569 2570 ceph_osds_init(raw); 2571 if (ppps) 2572 *ppps = pps; 2573 2574 ruleno = crush_find_rule(osdmap->crush, pi->crush_ruleset, pi->type, 2575 pi->size); 2576 if (ruleno < 0) { 2577 pr_err("no crush rule: pool %lld ruleset %d type %d size %d\n", 2578 pi->id, pi->crush_ruleset, pi->type, pi->size); 2579 return; 2580 } 2581 2582 if (pi->size > ARRAY_SIZE(raw->osds)) { 2583 pr_err_ratelimited("pool %lld ruleset %d type %d too wide: size %d > %zu\n", 2584 pi->id, pi->crush_ruleset, pi->type, pi->size, 2585 ARRAY_SIZE(raw->osds)); 2586 return; 2587 } 2588 2589 len = do_crush(osdmap, ruleno, pps, raw->osds, pi->size, 2590 osdmap->osd_weight, osdmap->max_osd, pi->id); 2591 if (len < 0) { 2592 pr_err("error %d from crush rule %d: pool %lld ruleset %d type %d size %d\n", 2593 len, ruleno, pi->id, pi->crush_ruleset, pi->type, 2594 pi->size); 2595 return; 2596 } 2597 2598 raw->size = len; 2599 remove_nonexistent_osds(osdmap, pi, raw); 2600 } 2601 2602 /* apply pg_upmap[_items] mappings */ 2603 static void apply_upmap(struct ceph_osdmap *osdmap, 2604 const struct ceph_pg *pgid, 2605 struct ceph_osds *raw) 2606 { 2607 struct ceph_pg_mapping *pg; 2608 int i, j; 2609 2610 pg = lookup_pg_mapping(&osdmap->pg_upmap, pgid); 2611 if (pg) { 2612 /* make sure targets aren't marked out */ 2613 for (i = 0; i < pg->pg_upmap.len; i++) { 2614 int osd = pg->pg_upmap.osds[i]; 2615 2616 if (osd != CRUSH_ITEM_NONE && 2617 osd < osdmap->max_osd && 2618 osdmap->osd_weight[osd] == 0) { 2619 /* reject/ignore explicit mapping */ 2620 return; 2621 } 2622 } 2623 for (i = 0; i < pg->pg_upmap.len; i++) 2624 raw->osds[i] = pg->pg_upmap.osds[i]; 2625 raw->size = pg->pg_upmap.len; 2626 /* check and apply pg_upmap_items, if any */ 2627 } 2628 2629 pg = lookup_pg_mapping(&osdmap->pg_upmap_items, pgid); 2630 if (pg) { 2631 /* 2632 * Note: this approach does not allow a bidirectional swap, 2633 * e.g., [[1,2],[2,1]] applied to [0,1,2] -> [0,2,1]. 2634 */ 2635 for (i = 0; i < pg->pg_upmap_items.len; i++) { 2636 int from = pg->pg_upmap_items.from_to[i][0]; 2637 int to = pg->pg_upmap_items.from_to[i][1]; 2638 int pos = -1; 2639 bool exists = false; 2640 2641 /* make sure replacement doesn't already appear */ 2642 for (j = 0; j < raw->size; j++) { 2643 int osd = raw->osds[j]; 2644 2645 if (osd == to) { 2646 exists = true; 2647 break; 2648 } 2649 /* ignore mapping if target is marked out */ 2650 if (osd == from && pos < 0 && 2651 !(to != CRUSH_ITEM_NONE && 2652 to < osdmap->max_osd && 2653 osdmap->osd_weight[to] == 0)) { 2654 pos = j; 2655 } 2656 } 2657 if (!exists && pos >= 0) 2658 raw->osds[pos] = to; 2659 } 2660 } 2661 } 2662 2663 /* 2664 * Given raw set, calculate up set and up primary. By definition of an 2665 * up set, the result won't contain nonexistent or down OSDs. 2666 * 2667 * This is done in-place - on return @set is the up set. If it's 2668 * empty, ->primary will remain undefined. 2669 */ 2670 static void raw_to_up_osds(struct ceph_osdmap *osdmap, 2671 struct ceph_pg_pool_info *pi, 2672 struct ceph_osds *set) 2673 { 2674 int i; 2675 2676 /* ->primary is undefined for a raw set */ 2677 BUG_ON(set->primary != -1); 2678 2679 if (ceph_can_shift_osds(pi)) { 2680 int removed = 0; 2681 2682 /* shift left */ 2683 for (i = 0; i < set->size; i++) { 2684 if (ceph_osd_is_down(osdmap, set->osds[i])) { 2685 removed++; 2686 continue; 2687 } 2688 if (removed) 2689 set->osds[i - removed] = set->osds[i]; 2690 } 2691 set->size -= removed; 2692 if (set->size > 0) 2693 set->primary = set->osds[0]; 2694 } else { 2695 /* set down/dne devices to NONE */ 2696 for (i = set->size - 1; i >= 0; i--) { 2697 if (ceph_osd_is_down(osdmap, set->osds[i])) 2698 set->osds[i] = CRUSH_ITEM_NONE; 2699 else 2700 set->primary = set->osds[i]; 2701 } 2702 } 2703 } 2704 2705 static void apply_primary_affinity(struct ceph_osdmap *osdmap, 2706 struct ceph_pg_pool_info *pi, 2707 u32 pps, 2708 struct ceph_osds *up) 2709 { 2710 int i; 2711 int pos = -1; 2712 2713 /* 2714 * Do we have any non-default primary_affinity values for these 2715 * osds? 2716 */ 2717 if (!osdmap->osd_primary_affinity) 2718 return; 2719 2720 for (i = 0; i < up->size; i++) { 2721 int osd = up->osds[i]; 2722 2723 if (osd != CRUSH_ITEM_NONE && 2724 osdmap->osd_primary_affinity[osd] != 2725 CEPH_OSD_DEFAULT_PRIMARY_AFFINITY) { 2726 break; 2727 } 2728 } 2729 if (i == up->size) 2730 return; 2731 2732 /* 2733 * Pick the primary. Feed both the seed (for the pg) and the 2734 * osd into the hash/rng so that a proportional fraction of an 2735 * osd's pgs get rejected as primary. 2736 */ 2737 for (i = 0; i < up->size; i++) { 2738 int osd = up->osds[i]; 2739 u32 aff; 2740 2741 if (osd == CRUSH_ITEM_NONE) 2742 continue; 2743 2744 aff = osdmap->osd_primary_affinity[osd]; 2745 if (aff < CEPH_OSD_MAX_PRIMARY_AFFINITY && 2746 (crush_hash32_2(CRUSH_HASH_RJENKINS1, 2747 pps, osd) >> 16) >= aff) { 2748 /* 2749 * We chose not to use this primary. Note it 2750 * anyway as a fallback in case we don't pick 2751 * anyone else, but keep looking. 2752 */ 2753 if (pos < 0) 2754 pos = i; 2755 } else { 2756 pos = i; 2757 break; 2758 } 2759 } 2760 if (pos < 0) 2761 return; 2762 2763 up->primary = up->osds[pos]; 2764 2765 if (ceph_can_shift_osds(pi) && pos > 0) { 2766 /* move the new primary to the front */ 2767 for (i = pos; i > 0; i--) 2768 up->osds[i] = up->osds[i - 1]; 2769 up->osds[0] = up->primary; 2770 } 2771 } 2772 2773 /* 2774 * Get pg_temp and primary_temp mappings for given PG. 2775 * 2776 * Note that a PG may have none, only pg_temp, only primary_temp or 2777 * both pg_temp and primary_temp mappings. This means @temp isn't 2778 * always a valid OSD set on return: in the "only primary_temp" case, 2779 * @temp will have its ->primary >= 0 but ->size == 0. 2780 */ 2781 static void get_temp_osds(struct ceph_osdmap *osdmap, 2782 struct ceph_pg_pool_info *pi, 2783 const struct ceph_pg *pgid, 2784 struct ceph_osds *temp) 2785 { 2786 struct ceph_pg_mapping *pg; 2787 int i; 2788 2789 ceph_osds_init(temp); 2790 2791 /* pg_temp? */ 2792 pg = lookup_pg_mapping(&osdmap->pg_temp, pgid); 2793 if (pg) { 2794 for (i = 0; i < pg->pg_temp.len; i++) { 2795 if (ceph_osd_is_down(osdmap, pg->pg_temp.osds[i])) { 2796 if (ceph_can_shift_osds(pi)) 2797 continue; 2798 2799 temp->osds[temp->size++] = CRUSH_ITEM_NONE; 2800 } else { 2801 temp->osds[temp->size++] = pg->pg_temp.osds[i]; 2802 } 2803 } 2804 2805 /* apply pg_temp's primary */ 2806 for (i = 0; i < temp->size; i++) { 2807 if (temp->osds[i] != CRUSH_ITEM_NONE) { 2808 temp->primary = temp->osds[i]; 2809 break; 2810 } 2811 } 2812 } 2813 2814 /* primary_temp? (shouldn't ever be a nonexistent or down OSD) */ 2815 pg = lookup_pg_mapping(&osdmap->primary_temp, pgid); 2816 if (pg && !WARN_ON_ONCE(ceph_osd_is_down(osdmap, 2817 pg->primary_temp.osd))) 2818 temp->primary = pg->primary_temp.osd; 2819 } 2820 2821 /* 2822 * Map a PG to its acting set as well as its up set. 2823 * 2824 * Acting set is used for data mapping purposes, while up set can be 2825 * recorded for detecting interval changes and deciding whether to 2826 * resend a request. 2827 */ 2828 void ceph_pg_to_up_acting_osds(struct ceph_osdmap *osdmap, 2829 struct ceph_pg_pool_info *pi, 2830 const struct ceph_pg *raw_pgid, 2831 struct ceph_osds *up, 2832 struct ceph_osds *acting) 2833 { 2834 struct ceph_pg pgid; 2835 u32 pps; 2836 2837 WARN_ON(pi->id != raw_pgid->pool); 2838 raw_pg_to_pg(pi, raw_pgid, &pgid); 2839 2840 pg_to_raw_osds(osdmap, pi, raw_pgid, up, &pps); 2841 apply_upmap(osdmap, &pgid, up); 2842 raw_to_up_osds(osdmap, pi, up); 2843 apply_primary_affinity(osdmap, pi, pps, up); 2844 get_temp_osds(osdmap, pi, &pgid, acting); 2845 if (!acting->size) { 2846 memcpy(acting->osds, up->osds, up->size * sizeof(up->osds[0])); 2847 acting->size = up->size; 2848 if (acting->primary == -1) 2849 acting->primary = up->primary; 2850 } 2851 WARN_ON(!osds_valid(up) || !osds_valid(acting)); 2852 } 2853 2854 bool ceph_pg_to_primary_shard(struct ceph_osdmap *osdmap, 2855 struct ceph_pg_pool_info *pi, 2856 const struct ceph_pg *raw_pgid, 2857 struct ceph_spg *spgid) 2858 { 2859 struct ceph_pg pgid; 2860 struct ceph_osds up, acting; 2861 int i; 2862 2863 WARN_ON(pi->id != raw_pgid->pool); 2864 raw_pg_to_pg(pi, raw_pgid, &pgid); 2865 2866 if (ceph_can_shift_osds(pi)) { 2867 spgid->pgid = pgid; /* struct */ 2868 spgid->shard = CEPH_SPG_NOSHARD; 2869 return true; 2870 } 2871 2872 ceph_pg_to_up_acting_osds(osdmap, pi, &pgid, &up, &acting); 2873 for (i = 0; i < acting.size; i++) { 2874 if (acting.osds[i] == acting.primary) { 2875 spgid->pgid = pgid; /* struct */ 2876 spgid->shard = i; 2877 return true; 2878 } 2879 } 2880 2881 return false; 2882 } 2883 2884 /* 2885 * Return acting primary for given PG, or -1 if none. 2886 */ 2887 int ceph_pg_to_acting_primary(struct ceph_osdmap *osdmap, 2888 const struct ceph_pg *raw_pgid) 2889 { 2890 struct ceph_pg_pool_info *pi; 2891 struct ceph_osds up, acting; 2892 2893 pi = ceph_pg_pool_by_id(osdmap, raw_pgid->pool); 2894 if (!pi) 2895 return -1; 2896 2897 ceph_pg_to_up_acting_osds(osdmap, pi, raw_pgid, &up, &acting); 2898 return acting.primary; 2899 } 2900 EXPORT_SYMBOL(ceph_pg_to_acting_primary); 2901 2902 static struct crush_loc_node *alloc_crush_loc(size_t type_name_len, 2903 size_t name_len) 2904 { 2905 struct crush_loc_node *loc; 2906 2907 loc = kmalloc(sizeof(*loc) + type_name_len + name_len + 2, GFP_NOIO); 2908 if (!loc) 2909 return NULL; 2910 2911 RB_CLEAR_NODE(&loc->cl_node); 2912 return loc; 2913 } 2914 2915 static void free_crush_loc(struct crush_loc_node *loc) 2916 { 2917 WARN_ON(!RB_EMPTY_NODE(&loc->cl_node)); 2918 2919 kfree(loc); 2920 } 2921 2922 static int crush_loc_compare(const struct crush_loc *loc1, 2923 const struct crush_loc *loc2) 2924 { 2925 return strcmp(loc1->cl_type_name, loc2->cl_type_name) ?: 2926 strcmp(loc1->cl_name, loc2->cl_name); 2927 } 2928 2929 DEFINE_RB_FUNCS2(crush_loc, struct crush_loc_node, cl_loc, crush_loc_compare, 2930 RB_BYPTR, const struct crush_loc *, cl_node) 2931 2932 /* 2933 * Parses a set of <bucket type name>':'<bucket name> pairs separated 2934 * by '|', e.g. "rack:foo1|rack:foo2|datacenter:bar". 2935 * 2936 * Note that @crush_location is modified by strsep(). 2937 */ 2938 int ceph_parse_crush_location(char *crush_location, struct rb_root *locs) 2939 { 2940 struct crush_loc_node *loc; 2941 const char *type_name, *name, *colon; 2942 size_t type_name_len, name_len; 2943 2944 dout("%s '%s'\n", __func__, crush_location); 2945 while ((type_name = strsep(&crush_location, "|"))) { 2946 colon = strchr(type_name, ':'); 2947 if (!colon) 2948 return -EINVAL; 2949 2950 type_name_len = colon - type_name; 2951 if (type_name_len == 0) 2952 return -EINVAL; 2953 2954 name = colon + 1; 2955 name_len = strlen(name); 2956 if (name_len == 0) 2957 return -EINVAL; 2958 2959 loc = alloc_crush_loc(type_name_len, name_len); 2960 if (!loc) 2961 return -ENOMEM; 2962 2963 loc->cl_loc.cl_type_name = loc->cl_data; 2964 memcpy(loc->cl_loc.cl_type_name, type_name, type_name_len); 2965 loc->cl_loc.cl_type_name[type_name_len] = '\0'; 2966 2967 loc->cl_loc.cl_name = loc->cl_data + type_name_len + 1; 2968 memcpy(loc->cl_loc.cl_name, name, name_len); 2969 loc->cl_loc.cl_name[name_len] = '\0'; 2970 2971 if (!__insert_crush_loc(locs, loc)) { 2972 free_crush_loc(loc); 2973 return -EEXIST; 2974 } 2975 2976 dout("%s type_name '%s' name '%s'\n", __func__, 2977 loc->cl_loc.cl_type_name, loc->cl_loc.cl_name); 2978 } 2979 2980 return 0; 2981 } 2982 2983 int ceph_compare_crush_locs(struct rb_root *locs1, struct rb_root *locs2) 2984 { 2985 struct rb_node *n1 = rb_first(locs1); 2986 struct rb_node *n2 = rb_first(locs2); 2987 int ret; 2988 2989 for ( ; n1 && n2; n1 = rb_next(n1), n2 = rb_next(n2)) { 2990 struct crush_loc_node *loc1 = 2991 rb_entry(n1, struct crush_loc_node, cl_node); 2992 struct crush_loc_node *loc2 = 2993 rb_entry(n2, struct crush_loc_node, cl_node); 2994 2995 ret = crush_loc_compare(&loc1->cl_loc, &loc2->cl_loc); 2996 if (ret) 2997 return ret; 2998 } 2999 3000 if (!n1 && n2) 3001 return -1; 3002 if (n1 && !n2) 3003 return 1; 3004 return 0; 3005 } 3006 3007 void ceph_clear_crush_locs(struct rb_root *locs) 3008 { 3009 while (!RB_EMPTY_ROOT(locs)) { 3010 struct crush_loc_node *loc = 3011 rb_entry(rb_first(locs), struct crush_loc_node, cl_node); 3012 3013 erase_crush_loc(locs, loc); 3014 free_crush_loc(loc); 3015 } 3016 } 3017 3018 /* 3019 * [a-zA-Z0-9-_.]+ 3020 */ 3021 static bool is_valid_crush_name(const char *name) 3022 { 3023 do { 3024 if (!('a' <= *name && *name <= 'z') && 3025 !('A' <= *name && *name <= 'Z') && 3026 !('0' <= *name && *name <= '9') && 3027 *name != '-' && *name != '_' && *name != '.') 3028 return false; 3029 } while (*++name != '\0'); 3030 3031 return true; 3032 } 3033 3034 /* 3035 * Gets the parent of an item. Returns its id (<0 because the 3036 * parent is always a bucket), type id (>0 for the same reason, 3037 * via @parent_type_id) and location (via @parent_loc). If no 3038 * parent, returns 0. 3039 * 3040 * Does a linear search, as there are no parent pointers of any 3041 * kind. Note that the result is ambiguous for items that occur 3042 * multiple times in the map. 3043 */ 3044 static int get_immediate_parent(struct crush_map *c, int id, 3045 u16 *parent_type_id, 3046 struct crush_loc *parent_loc) 3047 { 3048 struct crush_bucket *b; 3049 struct crush_name_node *type_cn, *cn; 3050 int i, j; 3051 3052 for (i = 0; i < c->max_buckets; i++) { 3053 b = c->buckets[i]; 3054 if (!b) 3055 continue; 3056 3057 /* ignore per-class shadow hierarchy */ 3058 cn = lookup_crush_name(&c->names, b->id); 3059 if (!cn || !is_valid_crush_name(cn->cn_name)) 3060 continue; 3061 3062 for (j = 0; j < b->size; j++) { 3063 if (b->items[j] != id) 3064 continue; 3065 3066 type_cn = lookup_crush_name(&c->type_names, b->type); 3067 if (WARN_ON_ONCE(!type_cn)) 3068 continue; 3069 3070 *parent_type_id = b->type; 3071 parent_loc->cl_type_name = type_cn->cn_name; 3072 parent_loc->cl_name = cn->cn_name; 3073 return b->id; 3074 } 3075 } 3076 3077 return 0; /* no parent */ 3078 } 3079 3080 /* 3081 * Calculates the locality/distance from an item to a client 3082 * location expressed in terms of CRUSH hierarchy as a set of 3083 * (bucket type name, bucket name) pairs. Specifically, looks 3084 * for the lowest-valued bucket type for which the location of 3085 * @id matches one of the locations in @locs, so for standard 3086 * bucket types (host = 1, rack = 3, datacenter = 8, zone = 9) 3087 * a matching host is closer than a matching rack and a matching 3088 * data center is closer than a matching zone. 3089 * 3090 * Specifying multiple locations (a "multipath" location) such 3091 * as "rack=foo1 rack=foo2 datacenter=bar" is allowed -- @locs 3092 * is a multimap. The locality will be: 3093 * 3094 * - 3 for OSDs in racks foo1 and foo2 3095 * - 8 for OSDs in data center bar 3096 * - -1 for all other OSDs 3097 * 3098 * The lowest possible bucket type is 1, so the best locality 3099 * for an OSD is 1 (i.e. a matching host). Locality 0 would be 3100 * the OSD itself. 3101 */ 3102 int ceph_get_crush_locality(struct ceph_osdmap *osdmap, int id, 3103 struct rb_root *locs) 3104 { 3105 struct crush_loc loc; 3106 u16 type_id; 3107 3108 /* 3109 * Instead of repeated get_immediate_parent() calls, 3110 * the location of @id could be obtained with a single 3111 * depth-first traversal. 3112 */ 3113 for (;;) { 3114 id = get_immediate_parent(osdmap->crush, id, &type_id, &loc); 3115 if (id >= 0) 3116 return -1; /* not local */ 3117 3118 if (lookup_crush_loc(locs, &loc)) 3119 return type_id; 3120 } 3121 } 3122