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