net/ceph/osdmap.c

   1
   2 #include <linux/ceph/ceph_debug.h>
   3
   4 #include <linux/module.h>
   5 #include <linux/slab.h>
   6 #include <asm/div64.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 char *ceph_osdmap_state_str(char *str, int len, int state)
  15 {
  16         if (!len)
  17                 return str;
  18
  19         if ((state & CEPH_OSD_EXISTS) && (state & CEPH_OSD_UP))
  20                 snprintf(str, len, "exists, up");
  21         else if (state & CEPH_OSD_EXISTS)
  22                 snprintf(str, len, "exists");
  23         else if (state & CEPH_OSD_UP)
  24                 snprintf(str, len, "up");
  25         else
  26                 snprintf(str, len, "doesn't exist");
  27
  28         return str;
  29 }
  30
  31 /* maps */
  32
  33 static int calc_bits_of(unsigned int t)
  34 {
  35         int b = 0;
  36         while (t) {
  37                 t = t >> 1;
  38                 b++;
  39         }
  40         return b;
  41 }
  42
  43 /*
  44  * the foo_mask is the smallest value 2^n-1 that is >= foo.
  45  */
  46 static void calc_pg_masks(struct ceph_pg_pool_info *pi)
  47 {
  48         pi->pg_num_mask = (1 << calc_bits_of(pi->pg_num-1)) - 1;
  49         pi->pgp_num_mask = (1 << calc_bits_of(pi->pgp_num-1)) - 1;
  50 }
  51
  52 /*
  53  * decode crush map
  54  */
  55 static int crush_decode_uniform_bucket(void **p, void *end,
  56                                        struct crush_bucket_uniform *b)
  57 {
  58         dout("crush_decode_uniform_bucket %p to %p\n", *p, end);
  59         ceph_decode_need(p, end, (1+b->h.size) * sizeof(u32), bad);
  60         b->item_weight = ceph_decode_32(p);
  61         return 0;
  62 bad:
  63         return -EINVAL;
  64 }
  65
  66 static int crush_decode_list_bucket(void **p, void *end,
  67                                     struct crush_bucket_list *b)
  68 {
  69         int j;
  70         dout("crush_decode_list_bucket %p to %p\n", *p, end);
  71         b->item_weights = kcalloc(b->h.size, sizeof(u32), GFP_NOFS);
  72         if (b->item_weights == NULL)
  73                 return -ENOMEM;
  74         b->sum_weights = kcalloc(b->h.size, sizeof(u32), GFP_NOFS);
  75         if (b->sum_weights == NULL)
  76                 return -ENOMEM;
  77         ceph_decode_need(p, end, 2 * b->h.size * sizeof(u32), bad);
  78         for (j = 0; j < b->h.size; j++) {
  79                 b->item_weights[j] = ceph_decode_32(p);
  80                 b->sum_weights[j] = ceph_decode_32(p);
  81         }
  82         return 0;
  83 bad:
  84         return -EINVAL;
  85 }
  86
  87 static int crush_decode_tree_bucket(void **p, void *end,
  88                                     struct crush_bucket_tree *b)
  89 {
  90         int j;
  91         dout("crush_decode_tree_bucket %p to %p\n", *p, end);
  92         ceph_decode_8_safe(p, end, b->num_nodes, bad);
  93         b->node_weights = kcalloc(b->num_nodes, sizeof(u32), GFP_NOFS);
  94         if (b->node_weights == NULL)
  95                 return -ENOMEM;
  96         ceph_decode_need(p, end, b->num_nodes * sizeof(u32), bad);
  97         for (j = 0; j < b->num_nodes; j++)
  98                 b->node_weights[j] = ceph_decode_32(p);
  99         return 0;
 100 bad:
 101         return -EINVAL;
 102 }
 103
 104 static int crush_decode_straw_bucket(void **p, void *end,
 105                                      struct crush_bucket_straw *b)
 106 {
 107         int j;
 108         dout("crush_decode_straw_bucket %p to %p\n", *p, end);
 109         b->item_weights = kcalloc(b->h.size, sizeof(u32), GFP_NOFS);
 110         if (b->item_weights == NULL)
 111                 return -ENOMEM;
 112         b->straws = kcalloc(b->h.size, sizeof(u32), GFP_NOFS);
 113         if (b->straws == NULL)
 114                 return -ENOMEM;
 115         ceph_decode_need(p, end, 2 * b->h.size * sizeof(u32), bad);
 116         for (j = 0; j < b->h.size; j++) {
 117                 b->item_weights[j] = ceph_decode_32(p);
 118                 b->straws[j] = ceph_decode_32(p);
 119         }
 120         return 0;
 121 bad:
 122         return -EINVAL;
 123 }
 124
 125 static int crush_decode_straw2_bucket(void **p, void *end,
 126                                       struct crush_bucket_straw2 *b)
 127 {
 128         int j;
 129         dout("crush_decode_straw2_bucket %p to %p\n", *p, end);
 130         b->item_weights = kcalloc(b->h.size, sizeof(u32), GFP_NOFS);
 131         if (b->item_weights == NULL)
 132                 return -ENOMEM;
 133         ceph_decode_need(p, end, b->h.size * sizeof(u32), bad);
 134         for (j = 0; j < b->h.size; j++)
 135                 b->item_weights[j] = ceph_decode_32(p);
 136         return 0;
 137 bad:
 138         return -EINVAL;
 139 }
 140
 141 static int skip_name_map(void **p, void *end)
 142 {
 143         int len;
 144         ceph_decode_32_safe(p, end, len ,bad);
 145         while (len--) {
 146                 int strlen;
 147                 *p += sizeof(u32);
 148                 ceph_decode_32_safe(p, end, strlen, bad);
 149                 *p += strlen;
 150 }
 151         return 0;
 152 bad:
 153         return -EINVAL;
 154 }
 155
 156 static void crush_finalize(struct crush_map *c)
 157 {
 158         __s32 b;
 159
 160         /* Space for the array of pointers to per-bucket workspace */
 161         c->working_size = sizeof(struct crush_work) +
 162             c->max_buckets * sizeof(struct crush_work_bucket *);
 163
 164         for (b = 0; b < c->max_buckets; b++) {
 165                 if (!c->buckets[b])
 166                         continue;
 167
 168                 switch (c->buckets[b]->alg) {
 169                 default:
 170                         /*
 171                          * The base case, permutation variables and
 172                          * the pointer to the permutation array.
 173                          */
 174                         c->working_size += sizeof(struct crush_work_bucket);
 175                         break;
 176                 }
 177                 /* Every bucket has a permutation array. */
 178                 c->working_size += c->buckets[b]->size * sizeof(__u32);
 179         }
 180 }
 181
 182 static struct crush_map *crush_decode(void *pbyval, void *end)
 183 {
 184         struct crush_map *c;
 185         int err = -EINVAL;
 186         int i, j;
 187         void **p = &pbyval;
 188         void *start = pbyval;
 189         u32 magic;
 190         u32 num_name_maps;
 191
 192         dout("crush_decode %p to %p len %d\n", *p, end, (int)(end - *p));
 193
 194         c = kzalloc(sizeof(*c), GFP_NOFS);
 195         if (c == NULL)
 196                 return ERR_PTR(-ENOMEM);
 197
 198         /* set tunables to default values */
 199         c->choose_local_tries = 2;
 200         c->choose_local_fallback_tries = 5;
 201         c->choose_total_tries = 19;
 202         c->chooseleaf_descend_once = 0;
 203
 204         ceph_decode_need(p, end, 4*sizeof(u32), bad);
 205         magic = ceph_decode_32(p);
 206         if (magic != CRUSH_MAGIC) {
 207                 pr_err("crush_decode magic %x != current %x\n",
 208                        (unsigned int)magic, (unsigned int)CRUSH_MAGIC);
 209                 goto bad;
 210         }
 211         c->max_buckets = ceph_decode_32(p);
 212         c->max_rules = ceph_decode_32(p);
 213         c->max_devices = ceph_decode_32(p);
 214
 215         c->buckets = kcalloc(c->max_buckets, sizeof(*c->buckets), GFP_NOFS);
 216         if (c->buckets == NULL)
 217                 goto badmem;
 218         c->rules = kcalloc(c->max_rules, sizeof(*c->rules), GFP_NOFS);
 219         if (c->rules == NULL)
 220                 goto badmem;
 221
 222         /* buckets */
 223         for (i = 0; i < c->max_buckets; i++) {
 224                 int size = 0;
 225                 u32 alg;
 226                 struct crush_bucket *b;
 227
 228                 ceph_decode_32_safe(p, end, alg, bad);
 229                 if (alg == 0) {
 230                         c->buckets[i] = NULL;
 231                         continue;
 232                 }
 233                 dout("crush_decode bucket %d off %x %p to %p\n",
 234                      i, (int)(*p-start), *p, end);
 235
 236                 switch (alg) {
 237                 case CRUSH_BUCKET_UNIFORM:
 238                         size = sizeof(struct crush_bucket_uniform);
 239                         break;
 240                 case CRUSH_BUCKET_LIST:
 241                         size = sizeof(struct crush_bucket_list);
 242                         break;
 243                 case CRUSH_BUCKET_TREE:
 244                         size = sizeof(struct crush_bucket_tree);
 245                         break;
 246                 case CRUSH_BUCKET_STRAW:
 247                         size = sizeof(struct crush_bucket_straw);
 248                         break;
 249                 case CRUSH_BUCKET_STRAW2:
 250                         size = sizeof(struct crush_bucket_straw2);
 251                         break;
 252                 default:
 253                         err = -EINVAL;
 254                         goto bad;
 255                 }
 256                 BUG_ON(size == 0);
 257                 b = c->buckets[i] = kzalloc(size, GFP_NOFS);
 258                 if (b == NULL)
 259                         goto badmem;
 260
 261                 ceph_decode_need(p, end, 4*sizeof(u32), bad);
 262                 b->id = ceph_decode_32(p);
 263                 b->type = ceph_decode_16(p);
 264                 b->alg = ceph_decode_8(p);
 265                 b->hash = ceph_decode_8(p);
 266                 b->weight = ceph_decode_32(p);
 267                 b->size = ceph_decode_32(p);
 268
 269                 dout("crush_decode bucket size %d off %x %p to %p\n",
 270                      b->size, (int)(*p-start), *p, end);
 271
 272                 b->items = kcalloc(b->size, sizeof(__s32), GFP_NOFS);
 273                 if (b->items == NULL)
 274                         goto badmem;
 275
 276                 ceph_decode_need(p, end, b->size*sizeof(u32), bad);
 277                 for (j = 0; j < b->size; j++)
 278                         b->items[j] = ceph_decode_32(p);
 279
 280                 switch (b->alg) {
 281                 case CRUSH_BUCKET_UNIFORM:
 282                         err = crush_decode_uniform_bucket(p, end,
 283                                   (struct crush_bucket_uniform *)b);
 284                         if (err < 0)
 285                                 goto bad;
 286                         break;
 287                 case CRUSH_BUCKET_LIST:
 288                         err = crush_decode_list_bucket(p, end,
 289                                (struct crush_bucket_list *)b);
 290                         if (err < 0)
 291                                 goto bad;
 292                         break;
 293                 case CRUSH_BUCKET_TREE:
 294                         err = crush_decode_tree_bucket(p, end,
 295                                 (struct crush_bucket_tree *)b);
 296                         if (err < 0)
 297                                 goto bad;
 298                         break;
 299                 case CRUSH_BUCKET_STRAW:
 300                         err = crush_decode_straw_bucket(p, end,
 301                                 (struct crush_bucket_straw *)b);
 302                         if (err < 0)
 303                                 goto bad;
 304                         break;
 305                 case CRUSH_BUCKET_STRAW2:
 306                         err = crush_decode_straw2_bucket(p, end,
 307                                 (struct crush_bucket_straw2 *)b);
 308                         if (err < 0)
 309                                 goto bad;
 310                         break;
 311                 }
 312         }
 313
 314         /* rules */
 315         dout("rule vec is %p\n", c->rules);
 316         for (i = 0; i < c->max_rules; i++) {
 317                 u32 yes;
 318                 struct crush_rule *r;
 319
 320                 err = -EINVAL;
 321                 ceph_decode_32_safe(p, end, yes, bad);
 322                 if (!yes) {
 323                         dout("crush_decode NO rule %d off %x %p to %p\n",
 324                              i, (int)(*p-start), *p, end);
 325                         c->rules[i] = NULL;
 326                         continue;
 327                 }
 328
 329                 dout("crush_decode rule %d off %x %p to %p\n",
 330                      i, (int)(*p-start), *p, end);
 331
 332                 /* len */
 333                 ceph_decode_32_safe(p, end, yes, bad);
 334 #if BITS_PER_LONG == 32
 335                 err = -EINVAL;
 336                 if (yes > (ULONG_MAX - sizeof(*r))
 337                           / sizeof(struct crush_rule_step))
 338                         goto bad;
 339 #endif
 340                 r = c->rules[i] = kmalloc(sizeof(*r) +
 341                                           yes*sizeof(struct crush_rule_step),
 342                                           GFP_NOFS);
 343                 if (r == NULL)
 344                         goto badmem;
 345                 dout(" rule %d is at %p\n", i, r);
 346                 r->len = yes;
 347                 ceph_decode_copy_safe(p, end, &r->mask, 4, bad); /* 4 u8's */
 348                 ceph_decode_need(p, end, r->len*3*sizeof(u32), bad);
 349                 for (j = 0; j < r->len; j++) {
 350                         r->steps[j].op = ceph_decode_32(p);
 351                         r->steps[j].arg1 = ceph_decode_32(p);
 352                         r->steps[j].arg2 = ceph_decode_32(p);
 353                 }
 354         }
 355
 356         /* ignore trailing name maps. */
 357         for (num_name_maps = 0; num_name_maps < 3; num_name_maps++) {
 358                 err = skip_name_map(p, end);
 359                 if (err < 0)
 360                         goto done;
 361         }
 362
 363         /* tunables */
 364         ceph_decode_need(p, end, 3*sizeof(u32), done);
 365         c->choose_local_tries = ceph_decode_32(p);
 366         c->choose_local_fallback_tries =  ceph_decode_32(p);
 367         c->choose_total_tries = ceph_decode_32(p);
 368         dout("crush decode tunable choose_local_tries = %d\n",
 369              c->choose_local_tries);
 370         dout("crush decode tunable choose_local_fallback_tries = %d\n",
 371              c->choose_local_fallback_tries);
 372         dout("crush decode tunable choose_total_tries = %d\n",
 373              c->choose_total_tries);
 374
 375         ceph_decode_need(p, end, sizeof(u32), done);
 376         c->chooseleaf_descend_once = ceph_decode_32(p);
 377         dout("crush decode tunable chooseleaf_descend_once = %d\n",
 378              c->chooseleaf_descend_once);
 379
 380         ceph_decode_need(p, end, sizeof(u8), done);
 381         c->chooseleaf_vary_r = ceph_decode_8(p);
 382         dout("crush decode tunable chooseleaf_vary_r = %d\n",
 383              c->chooseleaf_vary_r);
 384
 385         /* skip straw_calc_version, allowed_bucket_algs */
 386         ceph_decode_need(p, end, sizeof(u8) + sizeof(u32), done);
 387         *p += sizeof(u8) + sizeof(u32);
 388
 389         ceph_decode_need(p, end, sizeof(u8), done);
 390         c->chooseleaf_stable = ceph_decode_8(p);
 391         dout("crush decode tunable chooseleaf_stable = %d\n",
 392              c->chooseleaf_stable);
 393
 394 done:
 395         crush_finalize(c);
 396         dout("crush_decode success\n");
 397         return c;
 398
 399 badmem:
 400         err = -ENOMEM;
 401 bad:
 402         dout("crush_decode fail %d\n", err);
 403         crush_destroy(c);
 404         return ERR_PTR(err);
 405 }
 406
 407 int ceph_pg_compare(const struct ceph_pg *lhs, const struct ceph_pg *rhs)
 408 {
 409         if (lhs->pool < rhs->pool)
 410                 return -1;
 411         if (lhs->pool > rhs->pool)
 412                 return 1;
 413         if (lhs->seed < rhs->seed)
 414                 return -1;
 415         if (lhs->seed > rhs->seed)
 416                 return 1;
 417
 418         return 0;
 419 }
 420
 421 /*
 422  * rbtree of pg_mapping for handling pg_temp (explicit mapping of pgid
 423  * to a set of osds) and primary_temp (explicit primary setting)
 424  */
 425 static int __insert_pg_mapping(struct ceph_pg_mapping *new,
 426                                struct rb_root *root)
 427 {
 428         struct rb_node **p = &root->rb_node;
 429         struct rb_node *parent = NULL;
 430         struct ceph_pg_mapping *pg = NULL;
 431         int c;
 432
 433         dout("__insert_pg_mapping %llx %p\n", *(u64 *)&new->pgid, new);
 434         while (*p) {
 435                 parent = *p;
 436                 pg = rb_entry(parent, struct ceph_pg_mapping, node);
 437                 c = ceph_pg_compare(&new->pgid, &pg->pgid);
 438                 if (c < 0)
 439                         p = &(*p)->rb_left;
 440                 else if (c > 0)
 441                         p = &(*p)->rb_right;
 442                 else
 443                         return -EEXIST;
 444         }
 445
 446         rb_link_node(&new->node, parent, p);
 447         rb_insert_color(&new->node, root);
 448         return 0;
 449 }
 450
 451 static struct ceph_pg_mapping *__lookup_pg_mapping(struct rb_root *root,
 452                                                    struct ceph_pg pgid)
 453 {
 454         struct rb_node *n = root->rb_node;
 455         struct ceph_pg_mapping *pg;
 456         int c;
 457
 458         while (n) {
 459                 pg = rb_entry(n, struct ceph_pg_mapping, node);
 460                 c = ceph_pg_compare(&pgid, &pg->pgid);
 461                 if (c < 0) {
 462                         n = n->rb_left;
 463                 } else if (c > 0) {
 464                         n = n->rb_right;
 465                 } else {
 466                         dout("__lookup_pg_mapping %lld.%x got %p\n",
 467                              pgid.pool, pgid.seed, pg);
 468                         return pg;
 469                 }
 470         }
 471         return NULL;
 472 }
 473
 474 static int __remove_pg_mapping(struct rb_root *root, struct ceph_pg pgid)
 475 {
 476         struct ceph_pg_mapping *pg = __lookup_pg_mapping(root, pgid);
 477
 478         if (pg) {
 479                 dout("__remove_pg_mapping %lld.%x %p\n", pgid.pool, pgid.seed,
 480                      pg);
 481                 rb_erase(&pg->node, root);
 482                 kfree(pg);
 483                 return 0;
 484         }
 485         dout("__remove_pg_mapping %lld.%x dne\n", pgid.pool, pgid.seed);
 486         return -ENOENT;
 487 }
 488
 489 /*
 490  * rbtree of pg pool info
 491  */
 492 static int __insert_pg_pool(struct rb_root *root, struct ceph_pg_pool_info *new)
 493 {
 494         struct rb_node **p = &root->rb_node;
 495         struct rb_node *parent = NULL;
 496         struct ceph_pg_pool_info *pi = NULL;
 497
 498         while (*p) {
 499                 parent = *p;
 500                 pi = rb_entry(parent, struct ceph_pg_pool_info, node);
 501                 if (new->id < pi->id)
 502                         p = &(*p)->rb_left;
 503                 else if (new->id > pi->id)
 504                         p = &(*p)->rb_right;
 505                 else
 506                         return -EEXIST;
 507         }
 508
 509         rb_link_node(&new->node, parent, p);
 510         rb_insert_color(&new->node, root);
 511         return 0;
 512 }
 513
 514 static struct ceph_pg_pool_info *__lookup_pg_pool(struct rb_root *root, u64 id)
 515 {
 516         struct ceph_pg_pool_info *pi;
 517         struct rb_node *n = root->rb_node;
 518
 519         while (n) {
 520                 pi = rb_entry(n, struct ceph_pg_pool_info, node);
 521                 if (id < pi->id)
 522                         n = n->rb_left;
 523                 else if (id > pi->id)
 524                         n = n->rb_right;
 525                 else
 526                         return pi;
 527         }
 528         return NULL;
 529 }
 530
 531 struct ceph_pg_pool_info *ceph_pg_pool_by_id(struct ceph_osdmap *map, u64 id)
 532 {
 533         return __lookup_pg_pool(&map->pg_pools, id);
 534 }
 535
 536 const char *ceph_pg_pool_name_by_id(struct ceph_osdmap *map, u64 id)
 537 {
 538         struct ceph_pg_pool_info *pi;
 539
 540         if (id == CEPH_NOPOOL)
 541                 return NULL;
 542
 543         if (WARN_ON_ONCE(id > (u64) INT_MAX))
 544                 return NULL;
 545
 546         pi = __lookup_pg_pool(&map->pg_pools, (int) id);
 547
 548         return pi ? pi->name : NULL;
 549 }
 550 EXPORT_SYMBOL(ceph_pg_pool_name_by_id);
 551
 552 int ceph_pg_poolid_by_name(struct ceph_osdmap *map, const char *name)
 553 {
 554         struct rb_node *rbp;
 555
 556         for (rbp = rb_first(&map->pg_pools); rbp; rbp = rb_next(rbp)) {
 557                 struct ceph_pg_pool_info *pi =
 558                         rb_entry(rbp, struct ceph_pg_pool_info, node);
 559                 if (pi->name && strcmp(pi->name, name) == 0)
 560                         return pi->id;
 561         }
 562         return -ENOENT;
 563 }
 564 EXPORT_SYMBOL(ceph_pg_poolid_by_name);
 565
 566 static void __remove_pg_pool(struct rb_root *root, struct ceph_pg_pool_info *pi)
 567 {
 568         rb_erase(&pi->node, root);
 569         kfree(pi->name);
 570         kfree(pi);
 571 }
 572
 573 static int decode_pool(void **p, void *end, struct ceph_pg_pool_info *pi)
 574 {
 575         u8 ev, cv;
 576         unsigned len, num;
 577         void *pool_end;
 578
 579         ceph_decode_need(p, end, 2 + 4, bad);
 580         ev = ceph_decode_8(p);  /* encoding version */
 581         cv = ceph_decode_8(p); /* compat version */
 582         if (ev < 5) {
 583                 pr_warn("got v %d < 5 cv %d of ceph_pg_pool\n", ev, cv);
 584                 return -EINVAL;
 585         }
 586         if (cv > 9) {
 587                 pr_warn("got v %d cv %d > 9 of ceph_pg_pool\n", ev, cv);
 588                 return -EINVAL;
 589         }
 590         len = ceph_decode_32(p);
 591         ceph_decode_need(p, end, len, bad);
 592         pool_end = *p + len;
 593
 594         pi->type = ceph_decode_8(p);
 595         pi->size = ceph_decode_8(p);
 596         pi->crush_ruleset = ceph_decode_8(p);
 597         pi->object_hash = ceph_decode_8(p);
 598
 599         pi->pg_num = ceph_decode_32(p);
 600         pi->pgp_num = ceph_decode_32(p);
 601
 602         *p += 4 + 4;  /* skip lpg* */
 603         *p += 4;      /* skip last_change */
 604         *p += 8 + 4;  /* skip snap_seq, snap_epoch */
 605
 606         /* skip snaps */
 607         num = ceph_decode_32(p);
 608         while (num--) {
 609                 *p += 8;  /* snapid key */
 610                 *p += 1 + 1; /* versions */
 611                 len = ceph_decode_32(p);
 612                 *p += len;
 613         }
 614
 615         /* skip removed_snaps */
 616         num = ceph_decode_32(p);
 617         *p += num * (8 + 8);
 618
 619         *p += 8;  /* skip auid */
 620         pi->flags = ceph_decode_64(p);
 621         *p += 4;  /* skip crash_replay_interval */
 622
 623         if (ev >= 7)
 624                 pi->min_size = ceph_decode_8(p);
 625         else
 626                 pi->min_size = pi->size - pi->size / 2;
 627
 628         if (ev >= 8)
 629                 *p += 8 + 8;  /* skip quota_max_* */
 630
 631         if (ev >= 9) {
 632                 /* skip tiers */
 633                 num = ceph_decode_32(p);
 634                 *p += num * 8;
 635
 636                 *p += 8;  /* skip tier_of */
 637                 *p += 1;  /* skip cache_mode */
 638
 639                 pi->read_tier = ceph_decode_64(p);
 640                 pi->write_tier = ceph_decode_64(p);
 641         } else {
 642                 pi->read_tier = -1;
 643                 pi->write_tier = -1;
 644         }
 645
 646         if (ev >= 10) {
 647                 /* skip properties */
 648                 num = ceph_decode_32(p);
 649                 while (num--) {
 650                         len = ceph_decode_32(p);
 651                         *p += len; /* key */
 652                         len = ceph_decode_32(p);
 653                         *p += len; /* val */
 654                 }
 655         }
 656
 657         if (ev >= 11) {
 658                 /* skip hit_set_params */
 659                 *p += 1 + 1; /* versions */
 660                 len = ceph_decode_32(p);
 661                 *p += len;
 662
 663                 *p += 4; /* skip hit_set_period */
 664                 *p += 4; /* skip hit_set_count */
 665         }
 666
 667         if (ev >= 12)
 668                 *p += 4; /* skip stripe_width */
 669
 670         if (ev >= 13) {
 671                 *p += 8; /* skip target_max_bytes */
 672                 *p += 8; /* skip target_max_objects */
 673                 *p += 4; /* skip cache_target_dirty_ratio_micro */
 674                 *p += 4; /* skip cache_target_full_ratio_micro */
 675                 *p += 4; /* skip cache_min_flush_age */
 676                 *p += 4; /* skip cache_min_evict_age */
 677         }
 678
 679         if (ev >=  14) {
 680                 /* skip erasure_code_profile */
 681                 len = ceph_decode_32(p);
 682                 *p += len;
 683         }
 684
 685         if (ev >= 15)
 686                 pi->last_force_request_resend = ceph_decode_32(p);
 687         else
 688                 pi->last_force_request_resend = 0;
 689
 690         /* ignore the rest */
 691
 692         *p = pool_end;
 693         calc_pg_masks(pi);
 694         return 0;
 695
 696 bad:
 697         return -EINVAL;
 698 }
 699
 700 static int decode_pool_names(void **p, void *end, struct ceph_osdmap *map)
 701 {
 702         struct ceph_pg_pool_info *pi;
 703         u32 num, len;
 704         u64 pool;
 705
 706         ceph_decode_32_safe(p, end, num, bad);
 707         dout(" %d pool names\n", num);
 708         while (num--) {
 709                 ceph_decode_64_safe(p, end, pool, bad);
 710                 ceph_decode_32_safe(p, end, len, bad);
 711                 dout("  pool %llu len %d\n", pool, len);
 712                 ceph_decode_need(p, end, len, bad);
 713                 pi = __lookup_pg_pool(&map->pg_pools, pool);
 714                 if (pi) {
 715                         char *name = kstrndup(*p, len, GFP_NOFS);
 716
 717                         if (!name)
 718                                 return -ENOMEM;
 719                         kfree(pi->name);
 720                         pi->name = name;
 721                         dout("  name is %s\n", pi->name);
 722                 }
 723                 *p += len;
 724         }
 725         return 0;
 726
 727 bad:
 728         return -EINVAL;
 729 }
 730
 731 /*
 732  * osd map
 733  */
 734 struct ceph_osdmap *ceph_osdmap_alloc(void)
 735 {
 736         struct ceph_osdmap *map;
 737
 738         map = kzalloc(sizeof(*map), GFP_NOIO);
 739         if (!map)
 740                 return NULL;
 741
 742         map->pg_pools = RB_ROOT;
 743         map->pool_max = -1;
 744         map->pg_temp = RB_ROOT;
 745         map->primary_temp = RB_ROOT;
 746         mutex_init(&map->crush_workspace_mutex);
 747
 748         return map;
 749 }
 750
 751 void ceph_osdmap_destroy(struct ceph_osdmap *map)
 752 {
 753         dout("osdmap_destroy %p\n", map);
 754         if (map->crush)
 755                 crush_destroy(map->crush);
 756         while (!RB_EMPTY_ROOT(&map->pg_temp)) {
 757                 struct ceph_pg_mapping *pg =
 758                         rb_entry(rb_first(&map->pg_temp),
 759                                  struct ceph_pg_mapping, node);
 760                 rb_erase(&pg->node, &map->pg_temp);
 761                 kfree(pg);
 762         }
 763         while (!RB_EMPTY_ROOT(&map->primary_temp)) {
 764                 struct ceph_pg_mapping *pg =
 765                         rb_entry(rb_first(&map->primary_temp),
 766                                  struct ceph_pg_mapping, node);
 767                 rb_erase(&pg->node, &map->primary_temp);
 768                 kfree(pg);
 769         }
 770         while (!RB_EMPTY_ROOT(&map->pg_pools)) {
 771                 struct ceph_pg_pool_info *pi =
 772                         rb_entry(rb_first(&map->pg_pools),
 773                                  struct ceph_pg_pool_info, node);
 774                 __remove_pg_pool(&map->pg_pools, pi);
 775         }
 776         kfree(map->osd_state);
 777         kfree(map->osd_weight);
 778         kfree(map->osd_addr);
 779         kfree(map->osd_primary_affinity);
 780         kfree(map->crush_workspace);
 781         kfree(map);
 782 }
 783
 784 /*
 785  * Adjust max_osd value, (re)allocate arrays.
 786  *
 787  * The new elements are properly initialized.
 788  */
 789 static int osdmap_set_max_osd(struct ceph_osdmap *map, int max)
 790 {
 791         u8 *state;
 792         u32 *weight;
 793         struct ceph_entity_addr *addr;
 794         int i;
 795
 796         state = krealloc(map->osd_state, max*sizeof(*state), GFP_NOFS);
 797         if (!state)
 798                 return -ENOMEM;
 799         map->osd_state = state;
 800
 801         weight = krealloc(map->osd_weight, max*sizeof(*weight), GFP_NOFS);
 802         if (!weight)
 803                 return -ENOMEM;
 804         map->osd_weight = weight;
 805
 806         addr = krealloc(map->osd_addr, max*sizeof(*addr), GFP_NOFS);
 807         if (!addr)
 808                 return -ENOMEM;
 809         map->osd_addr = addr;
 810
 811         for (i = map->max_osd; i < max; i++) {
 812                 map->osd_state[i] = 0;
 813                 map->osd_weight[i] = CEPH_OSD_OUT;
 814                 memset(map->osd_addr + i, 0, sizeof(*map->osd_addr));
 815         }
 816
 817         if (map->osd_primary_affinity) {
 818                 u32 *affinity;
 819
 820                 affinity = krealloc(map->osd_primary_affinity,
 821                                     max*sizeof(*affinity), GFP_NOFS);
 822                 if (!affinity)
 823                         return -ENOMEM;
 824                 map->osd_primary_affinity = affinity;
 825
 826                 for (i = map->max_osd; i < max; i++)
 827                         map->osd_primary_affinity[i] =
 828                             CEPH_OSD_DEFAULT_PRIMARY_AFFINITY;
 829         }
 830
 831         map->max_osd = max;
 832
 833         return 0;
 834 }
 835
 836 static int osdmap_set_crush(struct ceph_osdmap *map, struct crush_map *crush)
 837 {
 838         void *workspace;
 839         size_t work_size;
 840
 841         if (IS_ERR(crush))
 842                 return PTR_ERR(crush);
 843
 844         work_size = crush_work_size(crush, CEPH_PG_MAX_SIZE);
 845         dout("%s work_size %zu bytes\n", __func__, work_size);
 846         workspace = kmalloc(work_size, GFP_NOIO);
 847         if (!workspace) {
 848                 crush_destroy(crush);
 849                 return -ENOMEM;
 850         }
 851         crush_init_workspace(crush, workspace);
 852
 853         if (map->crush)
 854                 crush_destroy(map->crush);
 855         kfree(map->crush_workspace);
 856         map->crush = crush;
 857         map->crush_workspace = workspace;
 858         return 0;
 859 }
 860
 861 #define OSDMAP_WRAPPER_COMPAT_VER       7
 862 #define OSDMAP_CLIENT_DATA_COMPAT_VER   1
 863
 864 /*
 865  * Return 0 or error.  On success, *v is set to 0 for old (v6) osdmaps,
 866  * to struct_v of the client_data section for new (v7 and above)
 867  * osdmaps.
 868  */
 869 static int get_osdmap_client_data_v(void **p, void *end,
 870                                     const char *prefix, u8 *v)
 871 {
 872         u8 struct_v;
 873
 874         ceph_decode_8_safe(p, end, struct_v, e_inval);
 875         if (struct_v >= 7) {
 876                 u8 struct_compat;
 877
 878                 ceph_decode_8_safe(p, end, struct_compat, e_inval);
 879                 if (struct_compat > OSDMAP_WRAPPER_COMPAT_VER) {
 880                         pr_warn("got v %d cv %d > %d of %s ceph_osdmap\n",
 881                                 struct_v, struct_compat,
 882                                 OSDMAP_WRAPPER_COMPAT_VER, prefix);
 883                         return -EINVAL;
 884                 }
 885                 *p += 4; /* ignore wrapper struct_len */
 886
 887                 ceph_decode_8_safe(p, end, struct_v, e_inval);
 888                 ceph_decode_8_safe(p, end, struct_compat, e_inval);
 889                 if (struct_compat > OSDMAP_CLIENT_DATA_COMPAT_VER) {
 890                         pr_warn("got v %d cv %d > %d of %s ceph_osdmap client data\n",
 891                                 struct_v, struct_compat,
 892                                 OSDMAP_CLIENT_DATA_COMPAT_VER, prefix);
 893                         return -EINVAL;
 894                 }
 895                 *p += 4; /* ignore client data struct_len */
 896         } else {
 897                 u16 version;
 898
 899                 *p -= 1;
 900                 ceph_decode_16_safe(p, end, version, e_inval);
 901                 if (version < 6) {
 902                         pr_warn("got v %d < 6 of %s ceph_osdmap\n",
 903                                 version, prefix);
 904                         return -EINVAL;
 905                 }
 906
 907                 /* old osdmap enconding */
 908                 struct_v = 0;
 909         }
 910
 911         *v = struct_v;
 912         return 0;
 913
 914 e_inval:
 915         return -EINVAL;
 916 }
 917
 918 static int __decode_pools(void **p, void *end, struct ceph_osdmap *map,
 919                           bool incremental)
 920 {
 921         u32 n;
 922
 923         ceph_decode_32_safe(p, end, n, e_inval);
 924         while (n--) {
 925                 struct ceph_pg_pool_info *pi;
 926                 u64 pool;
 927                 int ret;
 928
 929                 ceph_decode_64_safe(p, end, pool, e_inval);
 930
 931                 pi = __lookup_pg_pool(&map->pg_pools, pool);
 932                 if (!incremental || !pi) {
 933                         pi = kzalloc(sizeof(*pi), GFP_NOFS);
 934                         if (!pi)
 935                                 return -ENOMEM;
 936
 937                         pi->id = pool;
 938
 939                         ret = __insert_pg_pool(&map->pg_pools, pi);
 940                         if (ret) {
 941                                 kfree(pi);
 942                                 return ret;
 943                         }
 944                 }
 945
 946                 ret = decode_pool(p, end, pi);
 947                 if (ret)
 948                         return ret;
 949         }
 950
 951         return 0;
 952
 953 e_inval:
 954         return -EINVAL;
 955 }
 956
 957 static int decode_pools(void **p, void *end, struct ceph_osdmap *map)
 958 {
 959         return __decode_pools(p, end, map, false);
 960 }
 961
 962 static int decode_new_pools(void **p, void *end, struct ceph_osdmap *map)
 963 {
 964         return __decode_pools(p, end, map, true);
 965 }
 966
 967 static int __decode_pg_temp(void **p, void *end, struct ceph_osdmap *map,
 968                             bool incremental)
 969 {
 970         u32 n;
 971
 972         ceph_decode_32_safe(p, end, n, e_inval);
 973         while (n--) {
 974                 struct ceph_pg pgid;
 975                 u32 len, i;
 976                 int ret;
 977
 978                 ret = ceph_decode_pgid(p, end, &pgid);
 979                 if (ret)
 980                         return ret;
 981
 982                 ceph_decode_32_safe(p, end, len, e_inval);
 983
 984                 ret = __remove_pg_mapping(&map->pg_temp, pgid);
 985                 BUG_ON(!incremental && ret != -ENOENT);
 986
 987                 if (!incremental || len > 0) {
 988                         struct ceph_pg_mapping *pg;
 989
 990                         ceph_decode_need(p, end, len*sizeof(u32), e_inval);
 991
 992                         if (len > (UINT_MAX - sizeof(*pg)) / sizeof(u32))
 993                                 return -EINVAL;
 994
 995                         pg = kzalloc(sizeof(*pg) + len*sizeof(u32), GFP_NOFS);
 996                         if (!pg)
 997                                 return -ENOMEM;
 998
 999                         pg->pgid = pgid;
1000                         pg->pg_temp.len = len;
1001                         for (i = 0; i < len; i++)
1002                                 pg->pg_temp.osds[i] = ceph_decode_32(p);
1003
1004                         ret = __insert_pg_mapping(pg, &map->pg_temp);
1005                         if (ret) {
1006                                 kfree(pg);
1007                                 return ret;
1008                         }
1009                 }
1010         }
1011
1012         return 0;
1013
1014 e_inval:
1015         return -EINVAL;
1016 }
1017
1018 static int decode_pg_temp(void **p, void *end, struct ceph_osdmap *map)
1019 {
1020         return __decode_pg_temp(p, end, map, false);
1021 }
1022
1023 static int decode_new_pg_temp(void **p, void *end, struct ceph_osdmap *map)
1024 {
1025         return __decode_pg_temp(p, end, map, true);
1026 }
1027
1028 static int __decode_primary_temp(void **p, void *end, struct ceph_osdmap *map,
1029                                  bool incremental)
1030 {
1031         u32 n;
1032
1033         ceph_decode_32_safe(p, end, n, e_inval);
1034         while (n--) {
1035                 struct ceph_pg pgid;
1036                 u32 osd;
1037                 int ret;
1038
1039                 ret = ceph_decode_pgid(p, end, &pgid);
1040                 if (ret)
1041                         return ret;
1042
1043                 ceph_decode_32_safe(p, end, osd, e_inval);
1044
1045                 ret = __remove_pg_mapping(&map->primary_temp, pgid);
1046                 BUG_ON(!incremental && ret != -ENOENT);
1047
1048                 if (!incremental || osd != (u32)-1) {
1049                         struct ceph_pg_mapping *pg;
1050
1051                         pg = kzalloc(sizeof(*pg), GFP_NOFS);
1052                         if (!pg)
1053                                 return -ENOMEM;
1054
1055                         pg->pgid = pgid;
1056                         pg->primary_temp.osd = osd;
1057
1058                         ret = __insert_pg_mapping(pg, &map->primary_temp);
1059                         if (ret) {
1060                                 kfree(pg);
1061                                 return ret;
1062                         }
1063                 }
1064         }
1065
1066         return 0;
1067
1068 e_inval:
1069         return -EINVAL;
1070 }
1071
1072 static int decode_primary_temp(void **p, void *end, struct ceph_osdmap *map)
1073 {
1074         return __decode_primary_temp(p, end, map, false);
1075 }
1076
1077 static int decode_new_primary_temp(void **p, void *end,
1078                                    struct ceph_osdmap *map)
1079 {
1080         return __decode_primary_temp(p, end, map, true);
1081 }
1082
1083 u32 ceph_get_primary_affinity(struct ceph_osdmap *map, int osd)
1084 {
1085         BUG_ON(osd >= map->max_osd);
1086
1087         if (!map->osd_primary_affinity)
1088                 return CEPH_OSD_DEFAULT_PRIMARY_AFFINITY;
1089
1090         return map->osd_primary_affinity[osd];
1091 }
1092
1093 static int set_primary_affinity(struct ceph_osdmap *map, int osd, u32 aff)
1094 {
1095         BUG_ON(osd >= map->max_osd);
1096
1097         if (!map->osd_primary_affinity) {
1098                 int i;
1099
1100                 map->osd_primary_affinity = kmalloc(map->max_osd*sizeof(u32),
1101                                                     GFP_NOFS);
1102                 if (!map->osd_primary_affinity)
1103                         return -ENOMEM;
1104
1105                 for (i = 0; i < map->max_osd; i++)
1106                         map->osd_primary_affinity[i] =
1107                             CEPH_OSD_DEFAULT_PRIMARY_AFFINITY;
1108         }
1109
1110         map->osd_primary_affinity[osd] = aff;
1111
1112         return 0;
1113 }
1114
1115 static int decode_primary_affinity(void **p, void *end,
1116                                    struct ceph_osdmap *map)
1117 {
1118         u32 len, i;
1119
1120         ceph_decode_32_safe(p, end, len, e_inval);
1121         if (len == 0) {
1122                 kfree(map->osd_primary_affinity);
1123                 map->osd_primary_affinity = NULL;
1124                 return 0;
1125         }
1126         if (len != map->max_osd)
1127                 goto e_inval;
1128
1129         ceph_decode_need(p, end, map->max_osd*sizeof(u32), e_inval);
1130
1131         for (i = 0; i < map->max_osd; i++) {
1132                 int ret;
1133
1134                 ret = set_primary_affinity(map, i, ceph_decode_32(p));
1135                 if (ret)
1136                         return ret;
1137         }
1138
1139         return 0;
1140
1141 e_inval:
1142         return -EINVAL;
1143 }
1144
1145 static int decode_new_primary_affinity(void **p, void *end,
1146                                        struct ceph_osdmap *map)
1147 {
1148         u32 n;
1149
1150         ceph_decode_32_safe(p, end, n, e_inval);
1151         while (n--) {
1152                 u32 osd, aff;
1153                 int ret;
1154
1155                 ceph_decode_32_safe(p, end, osd, e_inval);
1156                 ceph_decode_32_safe(p, end, aff, e_inval);
1157
1158                 ret = set_primary_affinity(map, osd, aff);
1159                 if (ret)
1160                         return ret;
1161
1162                 pr_info("osd%d primary-affinity 0x%x\n", osd, aff);
1163         }
1164
1165         return 0;
1166
1167 e_inval:
1168         return -EINVAL;
1169 }
1170
1171 /*
1172  * decode a full map.
1173  */
1174 static int osdmap_decode(void **p, void *end, struct ceph_osdmap *map)
1175 {
1176         u8 struct_v;
1177         u32 epoch = 0;
1178         void *start = *p;
1179         u32 max;
1180         u32 len, i;
1181         int err;
1182
1183         dout("%s %p to %p len %d\n", __func__, *p, end, (int)(end - *p));
1184
1185         err = get_osdmap_client_data_v(p, end, "full", &struct_v);
1186         if (err)
1187                 goto bad;
1188
1189         /* fsid, epoch, created, modified */
1190         ceph_decode_need(p, end, sizeof(map->fsid) + sizeof(u32) +
1191                          sizeof(map->created) + sizeof(map->modified), e_inval);
1192         ceph_decode_copy(p, &map->fsid, sizeof(map->fsid));
1193         epoch = map->epoch = ceph_decode_32(p);
1194         ceph_decode_copy(p, &map->created, sizeof(map->created));
1195         ceph_decode_copy(p, &map->modified, sizeof(map->modified));
1196
1197         /* pools */
1198         err = decode_pools(p, end, map);
1199         if (err)
1200                 goto bad;
1201
1202         /* pool_name */
1203         err = decode_pool_names(p, end, map);
1204         if (err)
1205                 goto bad;
1206
1207         ceph_decode_32_safe(p, end, map->pool_max, e_inval);
1208
1209         ceph_decode_32_safe(p, end, map->flags, e_inval);
1210
1211         /* max_osd */
1212         ceph_decode_32_safe(p, end, max, e_inval);
1213
1214         /* (re)alloc osd arrays */
1215         err = osdmap_set_max_osd(map, max);
1216         if (err)
1217                 goto bad;
1218
1219         /* osd_state, osd_weight, osd_addrs->client_addr */
1220         ceph_decode_need(p, end, 3*sizeof(u32) +
1221                          map->max_osd*(1 + sizeof(*map->osd_weight) +
1222                                        sizeof(*map->osd_addr)), e_inval);
1223
1224         if (ceph_decode_32(p) != map->max_osd)
1225                 goto e_inval;
1226
1227         ceph_decode_copy(p, map->osd_state, map->max_osd);
1228
1229         if (ceph_decode_32(p) != map->max_osd)
1230                 goto e_inval;
1231
1232         for (i = 0; i < map->max_osd; i++)
1233                 map->osd_weight[i] = ceph_decode_32(p);
1234
1235         if (ceph_decode_32(p) != map->max_osd)
1236                 goto e_inval;
1237
1238         ceph_decode_copy(p, map->osd_addr, map->max_osd*sizeof(*map->osd_addr));
1239         for (i = 0; i < map->max_osd; i++)
1240                 ceph_decode_addr(&map->osd_addr[i]);
1241
1242         /* pg_temp */
1243         err = decode_pg_temp(p, end, map);
1244         if (err)
1245                 goto bad;
1246
1247         /* primary_temp */
1248         if (struct_v >= 1) {
1249                 err = decode_primary_temp(p, end, map);
1250                 if (err)
1251                         goto bad;
1252         }
1253
1254         /* primary_affinity */
1255         if (struct_v >= 2) {
1256                 err = decode_primary_affinity(p, end, map);
1257                 if (err)
1258                         goto bad;
1259         } else {
1260                 /* XXX can this happen? */
1261                 kfree(map->osd_primary_affinity);
1262                 map->osd_primary_affinity = NULL;
1263         }
1264
1265         /* crush */
1266         ceph_decode_32_safe(p, end, len, e_inval);
1267         err = osdmap_set_crush(map, crush_decode(*p, min(*p + len, end)));
1268         if (err)
1269                 goto bad;
1270
1271         /* ignore the rest */
1272         *p = end;
1273
1274         dout("full osdmap epoch %d max_osd %d\n", map->epoch, map->max_osd);
1275         return 0;
1276
1277 e_inval:
1278         err = -EINVAL;
1279 bad:
1280         pr_err("corrupt full osdmap (%d) epoch %d off %d (%p of %p-%p)\n",
1281                err, epoch, (int)(*p - start), *p, start, end);
1282         print_hex_dump(KERN_DEBUG, "osdmap: ",
1283                        DUMP_PREFIX_OFFSET, 16, 1,
1284                        start, end - start, true);
1285         return err;
1286 }
1287
1288 /*
1289  * Allocate and decode a full map.
1290  */
1291 struct ceph_osdmap *ceph_osdmap_decode(void **p, void *end)
1292 {
1293         struct ceph_osdmap *map;
1294         int ret;
1295
1296         map = ceph_osdmap_alloc();
1297         if (!map)
1298                 return ERR_PTR(-ENOMEM);
1299
1300         ret = osdmap_decode(p, end, map);
1301         if (ret) {
1302                 ceph_osdmap_destroy(map);
1303                 return ERR_PTR(ret);
1304         }
1305
1306         return map;
1307 }
1308
1309 /*
1310  * Encoding order is (new_up_client, new_state, new_weight).  Need to
1311  * apply in the (new_weight, new_state, new_up_client) order, because
1312  * an incremental map may look like e.g.
1313  *
1314  *     new_up_client: { osd=6, addr=... } # set osd_state and addr
1315  *     new_state: { osd=6, xorstate=EXISTS } # clear osd_state
1316  */
1317 static int decode_new_up_state_weight(void **p, void *end,
1318                                       struct ceph_osdmap *map)
1319 {
1320         void *new_up_client;
1321         void *new_state;
1322         void *new_weight_end;
1323         u32 len;
1324
1325         new_up_client = *p;
1326         ceph_decode_32_safe(p, end, len, e_inval);
1327         len *= sizeof(u32) + sizeof(struct ceph_entity_addr);
1328         ceph_decode_need(p, end, len, e_inval);
1329         *p += len;
1330
1331         new_state = *p;
1332         ceph_decode_32_safe(p, end, len, e_inval);
1333         len *= sizeof(u32) + sizeof(u8);
1334         ceph_decode_need(p, end, len, e_inval);
1335         *p += len;
1336
1337         /* new_weight */
1338         ceph_decode_32_safe(p, end, len, e_inval);
1339         while (len--) {
1340                 s32 osd;
1341                 u32 w;
1342
1343                 ceph_decode_need(p, end, 2*sizeof(u32), e_inval);
1344                 osd = ceph_decode_32(p);
1345                 w = ceph_decode_32(p);
1346                 BUG_ON(osd >= map->max_osd);
1347                 pr_info("osd%d weight 0x%x %s\n", osd, w,
1348                      w == CEPH_OSD_IN ? "(in)" :
1349                      (w == CEPH_OSD_OUT ? "(out)" : ""));
1350                 map->osd_weight[osd] = w;
1351
1352                 /*
1353                  * If we are marking in, set the EXISTS, and clear the
1354                  * AUTOOUT and NEW bits.
1355                  */
1356                 if (w) {
1357                         map->osd_state[osd] |= CEPH_OSD_EXISTS;
1358                         map->osd_state[osd] &= ~(CEPH_OSD_AUTOOUT |
1359                                                  CEPH_OSD_NEW);
1360                 }
1361         }
1362         new_weight_end = *p;
1363
1364         /* new_state (up/down) */
1365         *p = new_state;
1366         len = ceph_decode_32(p);
1367         while (len--) {
1368                 s32 osd;
1369                 u8 xorstate;
1370                 int ret;
1371
1372                 osd = ceph_decode_32(p);
1373                 xorstate = ceph_decode_8(p);
1374                 if (xorstate == 0)
1375                         xorstate = CEPH_OSD_UP;
1376                 BUG_ON(osd >= map->max_osd);
1377                 if ((map->osd_state[osd] & CEPH_OSD_UP) &&
1378                     (xorstate & CEPH_OSD_UP))
1379                         pr_info("osd%d down\n", osd);
1380                 if ((map->osd_state[osd] & CEPH_OSD_EXISTS) &&
1381                     (xorstate & CEPH_OSD_EXISTS)) {
1382                         pr_info("osd%d does not exist\n", osd);
1383                         ret = set_primary_affinity(map, osd,
1384                                                    CEPH_OSD_DEFAULT_PRIMARY_AFFINITY);
1385                         if (ret)
1386                                 return ret;
1387                         memset(map->osd_addr + osd, 0, sizeof(*map->osd_addr));
1388                         map->osd_state[osd] = 0;
1389                 } else {
1390                         map->osd_state[osd] ^= xorstate;
1391                 }
1392         }
1393
1394         /* new_up_client */
1395         *p = new_up_client;
1396         len = ceph_decode_32(p);
1397         while (len--) {
1398                 s32 osd;
1399                 struct ceph_entity_addr addr;
1400
1401                 osd = ceph_decode_32(p);
1402                 ceph_decode_copy(p, &addr, sizeof(addr));
1403                 ceph_decode_addr(&addr);
1404                 BUG_ON(osd >= map->max_osd);
1405                 pr_info("osd%d up\n", osd);
1406                 map->osd_state[osd] |= CEPH_OSD_EXISTS | CEPH_OSD_UP;
1407                 map->osd_addr[osd] = addr;
1408         }
1409
1410         *p = new_weight_end;
1411         return 0;
1412
1413 e_inval:
1414         return -EINVAL;
1415 }
1416
1417 /*
1418  * decode and apply an incremental map update.
1419  */
1420 struct ceph_osdmap *osdmap_apply_incremental(void **p, void *end,
1421                                              struct ceph_osdmap *map)
1422 {
1423         struct ceph_fsid fsid;
1424         u32 epoch = 0;
1425         struct ceph_timespec modified;
1426         s32 len;
1427         u64 pool;
1428         __s64 new_pool_max;
1429         __s32 new_flags, max;
1430         void *start = *p;
1431         int err;
1432         u8 struct_v;
1433
1434         dout("%s %p to %p len %d\n", __func__, *p, end, (int)(end - *p));
1435
1436         err = get_osdmap_client_data_v(p, end, "inc", &struct_v);
1437         if (err)
1438                 goto bad;
1439
1440         /* fsid, epoch, modified, new_pool_max, new_flags */
1441         ceph_decode_need(p, end, sizeof(fsid) + sizeof(u32) + sizeof(modified) +
1442                          sizeof(u64) + sizeof(u32), e_inval);
1443         ceph_decode_copy(p, &fsid, sizeof(fsid));
1444         epoch = ceph_decode_32(p);
1445         BUG_ON(epoch != map->epoch+1);
1446         ceph_decode_copy(p, &modified, sizeof(modified));
1447         new_pool_max = ceph_decode_64(p);
1448         new_flags = ceph_decode_32(p);
1449
1450         /* full map? */
1451         ceph_decode_32_safe(p, end, len, e_inval);
1452         if (len > 0) {
1453                 dout("apply_incremental full map len %d, %p to %p\n",
1454                      len, *p, end);
1455                 return ceph_osdmap_decode(p, min(*p+len, end));
1456         }
1457
1458         /* new crush? */
1459         ceph_decode_32_safe(p, end, len, e_inval);
1460         if (len > 0) {
1461                 err = osdmap_set_crush(map,
1462                                        crush_decode(*p, min(*p + len, end)));
1463                 if (err)
1464                         goto bad;
1465                 *p += len;
1466         }
1467
1468         /* new flags? */
1469         if (new_flags >= 0)
1470                 map->flags = new_flags;
1471         if (new_pool_max >= 0)
1472                 map->pool_max = new_pool_max;
1473
1474         /* new max? */
1475         ceph_decode_32_safe(p, end, max, e_inval);
1476         if (max >= 0) {
1477                 err = osdmap_set_max_osd(map, max);
1478                 if (err)
1479                         goto bad;
1480         }
1481
1482         map->epoch++;
1483         map->modified = modified;
1484
1485         /* new_pools */
1486         err = decode_new_pools(p, end, map);
1487         if (err)
1488                 goto bad;
1489
1490         /* new_pool_names */
1491         err = decode_pool_names(p, end, map);
1492         if (err)
1493                 goto bad;
1494
1495         /* old_pool */
1496         ceph_decode_32_safe(p, end, len, e_inval);
1497         while (len--) {
1498                 struct ceph_pg_pool_info *pi;
1499
1500                 ceph_decode_64_safe(p, end, pool, e_inval);
1501                 pi = __lookup_pg_pool(&map->pg_pools, pool);
1502                 if (pi)
1503                         __remove_pg_pool(&map->pg_pools, pi);
1504         }
1505
1506         /* new_up_client, new_state, new_weight */
1507         err = decode_new_up_state_weight(p, end, map);
1508         if (err)
1509                 goto bad;
1510
1511         /* new_pg_temp */
1512         err = decode_new_pg_temp(p, end, map);
1513         if (err)
1514                 goto bad;
1515
1516         /* new_primary_temp */
1517         if (struct_v >= 1) {
1518                 err = decode_new_primary_temp(p, end, map);
1519                 if (err)
1520                         goto bad;
1521         }
1522
1523         /* new_primary_affinity */
1524         if (struct_v >= 2) {
1525                 err = decode_new_primary_affinity(p, end, map);
1526                 if (err)
1527                         goto bad;
1528         }
1529
1530         /* ignore the rest */
1531         *p = end;
1532
1533         dout("inc osdmap epoch %d max_osd %d\n", map->epoch, map->max_osd);
1534         return map;
1535
1536 e_inval:
1537         err = -EINVAL;
1538 bad:
1539         pr_err("corrupt inc osdmap (%d) epoch %d off %d (%p of %p-%p)\n",
1540                err, epoch, (int)(*p - start), *p, start, end);
1541         print_hex_dump(KERN_DEBUG, "osdmap: ",
1542                        DUMP_PREFIX_OFFSET, 16, 1,
1543                        start, end - start, true);
1544         return ERR_PTR(err);
1545 }
1546
1547 void ceph_oloc_copy(struct ceph_object_locator *dest,
1548                     const struct ceph_object_locator *src)
1549 {
1550         WARN_ON(!ceph_oloc_empty(dest));
1551         WARN_ON(dest->pool_ns); /* empty() only covers ->pool */
1552
1553         dest->pool = src->pool;
1554         if (src->pool_ns)
1555                 dest->pool_ns = ceph_get_string(src->pool_ns);
1556 }
1557 EXPORT_SYMBOL(ceph_oloc_copy);
1558
1559 void ceph_oloc_destroy(struct ceph_object_locator *oloc)
1560 {
1561         ceph_put_string(oloc->pool_ns);
1562 }
1563 EXPORT_SYMBOL(ceph_oloc_destroy);
1564
1565 void ceph_oid_copy(struct ceph_object_id *dest,
1566                    const struct ceph_object_id *src)
1567 {
1568         WARN_ON(!ceph_oid_empty(dest));
1569
1570         if (src->name != src->inline_name) {
1571                 /* very rare, see ceph_object_id definition */
1572                 dest->name = kmalloc(src->name_len + 1,
1573                                      GFP_NOIO | __GFP_NOFAIL);
1574         }
1575
1576         memcpy(dest->name, src->name, src->name_len + 1);
1577         dest->name_len = src->name_len;
1578 }
1579 EXPORT_SYMBOL(ceph_oid_copy);
1580
1581 static __printf(2, 0)
1582 int oid_printf_vargs(struct ceph_object_id *oid, const char *fmt, va_list ap)
1583 {
1584         int len;
1585
1586         WARN_ON(!ceph_oid_empty(oid));
1587
1588         len = vsnprintf(oid->inline_name, sizeof(oid->inline_name), fmt, ap);
1589         if (len >= sizeof(oid->inline_name))
1590                 return len;
1591
1592         oid->name_len = len;
1593         return 0;
1594 }
1595
1596 /*
1597  * If oid doesn't fit into inline buffer, BUG.
1598  */
1599 void ceph_oid_printf(struct ceph_object_id *oid, const char *fmt, ...)
1600 {
1601         va_list ap;
1602
1603         va_start(ap, fmt);
1604         BUG_ON(oid_printf_vargs(oid, fmt, ap));
1605         va_end(ap);
1606 }
1607 EXPORT_SYMBOL(ceph_oid_printf);
1608
1609 static __printf(3, 0)
1610 int oid_aprintf_vargs(struct ceph_object_id *oid, gfp_t gfp,
1611                       const char *fmt, va_list ap)
1612 {
1613         va_list aq;
1614         int len;
1615
1616         va_copy(aq, ap);
1617         len = oid_printf_vargs(oid, fmt, aq);
1618         va_end(aq);
1619
1620         if (len) {
1621                 char *external_name;
1622
1623                 external_name = kmalloc(len + 1, gfp);
1624                 if (!external_name)
1625                         return -ENOMEM;
1626
1627                 oid->name = external_name;
1628                 WARN_ON(vsnprintf(oid->name, len + 1, fmt, ap) != len);
1629                 oid->name_len = len;
1630         }
1631
1632         return 0;
1633 }
1634
1635 /*
1636  * If oid doesn't fit into inline buffer, allocate.
1637  */
1638 int ceph_oid_aprintf(struct ceph_object_id *oid, gfp_t gfp,
1639                      const char *fmt, ...)
1640 {
1641         va_list ap;
1642         int ret;
1643
1644         va_start(ap, fmt);
1645         ret = oid_aprintf_vargs(oid, gfp, fmt, ap);
1646         va_end(ap);
1647
1648         return ret;
1649 }
1650 EXPORT_SYMBOL(ceph_oid_aprintf);
1651
1652 void ceph_oid_destroy(struct ceph_object_id *oid)
1653 {
1654         if (oid->name != oid->inline_name)
1655                 kfree(oid->name);
1656 }
1657 EXPORT_SYMBOL(ceph_oid_destroy);
1658
1659 /*
1660  * osds only
1661  */
1662 static bool __osds_equal(const struct ceph_osds *lhs,
1663                          const struct ceph_osds *rhs)
1664 {
1665         if (lhs->size == rhs->size &&
1666             !memcmp(lhs->osds, rhs->osds, rhs->size * sizeof(rhs->osds[0])))
1667                 return true;
1668
1669         return false;
1670 }
1671
1672 /*
1673  * osds + primary
1674  */
1675 static bool osds_equal(const struct ceph_osds *lhs,
1676                        const struct ceph_osds *rhs)
1677 {
1678         if (__osds_equal(lhs, rhs) &&
1679             lhs->primary == rhs->primary)
1680                 return true;
1681
1682         return false;
1683 }
1684
1685 static bool osds_valid(const struct ceph_osds *set)
1686 {
1687         /* non-empty set */
1688         if (set->size > 0 && set->primary >= 0)
1689                 return true;
1690
1691         /* empty can_shift_osds set */
1692         if (!set->size && set->primary == -1)
1693                 return true;
1694
1695         /* empty !can_shift_osds set - all NONE */
1696         if (set->size > 0 && set->primary == -1) {
1697                 int i;
1698
1699                 for (i = 0; i < set->size; i++) {
1700                         if (set->osds[i] != CRUSH_ITEM_NONE)
1701                                 break;
1702                 }
1703                 if (i == set->size)
1704                         return true;
1705         }
1706
1707         return false;
1708 }
1709
1710 void ceph_osds_copy(struct ceph_osds *dest, const struct ceph_osds *src)
1711 {
1712         memcpy(dest->osds, src->osds, src->size * sizeof(src->osds[0]));
1713         dest->size = src->size;
1714         dest->primary = src->primary;
1715 }
1716
1717 static bool is_split(const struct ceph_pg *pgid,
1718                      u32 old_pg_num,
1719                      u32 new_pg_num)
1720 {
1721         int old_bits = calc_bits_of(old_pg_num);
1722         int old_mask = (1 << old_bits) - 1;
1723         int n;
1724
1725         WARN_ON(pgid->seed >= old_pg_num);
1726         if (new_pg_num <= old_pg_num)
1727                 return false;
1728
1729         for (n = 1; ; n++) {
1730                 int next_bit = n << (old_bits - 1);
1731                 u32 s = next_bit | pgid->seed;
1732
1733                 if (s < old_pg_num || s == pgid->seed)
1734                         continue;
1735                 if (s >= new_pg_num)
1736                         break;
1737
1738                 s = ceph_stable_mod(s, old_pg_num, old_mask);
1739                 if (s == pgid->seed)
1740                         return true;
1741         }
1742
1743         return false;
1744 }
1745
1746 bool ceph_is_new_interval(const struct ceph_osds *old_acting,
1747                           const struct ceph_osds *new_acting,
1748                           const struct ceph_osds *old_up,
1749                           const struct ceph_osds *new_up,
1750                           int old_size,
1751                           int new_size,
1752                           int old_min_size,
1753                           int new_min_size,
1754                           u32 old_pg_num,
1755                           u32 new_pg_num,
1756                           bool old_sort_bitwise,
1757                           bool new_sort_bitwise,
1758                           const struct ceph_pg *pgid)
1759 {
1760         return !osds_equal(old_acting, new_acting) ||
1761                !osds_equal(old_up, new_up) ||
1762                old_size != new_size ||
1763                old_min_size != new_min_size ||
1764                is_split(pgid, old_pg_num, new_pg_num) ||
1765                old_sort_bitwise != new_sort_bitwise;
1766 }
1767
1768 static int calc_pg_rank(int osd, const struct ceph_osds *acting)
1769 {
1770         int i;
1771
1772         for (i = 0; i < acting->size; i++) {
1773                 if (acting->osds[i] == osd)
1774                         return i;
1775         }
1776
1777         return -1;
1778 }
1779
1780 static bool primary_changed(const struct ceph_osds *old_acting,
1781                             const struct ceph_osds *new_acting)
1782 {
1783         if (!old_acting->size && !new_acting->size)
1784                 return false; /* both still empty */
1785
1786         if (!old_acting->size ^ !new_acting->size)
1787                 return true; /* was empty, now not, or vice versa */
1788
1789         if (old_acting->primary != new_acting->primary)
1790                 return true; /* primary changed */
1791
1792         if (calc_pg_rank(old_acting->primary, old_acting) !=
1793             calc_pg_rank(new_acting->primary, new_acting))
1794                 return true;
1795
1796         return false; /* same primary (tho replicas may have changed) */
1797 }
1798
1799 bool ceph_osds_changed(const struct ceph_osds *old_acting,
1800                        const struct ceph_osds *new_acting,
1801                        bool any_change)
1802 {
1803         if (primary_changed(old_acting, new_acting))
1804                 return true;
1805
1806         if (any_change && !__osds_equal(old_acting, new_acting))
1807                 return true;
1808
1809         return false;
1810 }
1811
1812 /*
1813  * calculate file layout from given offset, length.
1814  * fill in correct oid, logical length, and object extent
1815  * offset, length.
1816  *
1817  * for now, we write only a single su, until we can
1818  * pass a stride back to the caller.
1819  */
1820 int ceph_calc_file_object_mapping(struct ceph_file_layout *layout,
1821                                    u64 off, u64 len,
1822                                    u64 *ono,
1823                                    u64 *oxoff, u64 *oxlen)
1824 {
1825         u32 osize = layout->object_size;
1826         u32 su = layout->stripe_unit;
1827         u32 sc = layout->stripe_count;
1828         u32 bl, stripeno, stripepos, objsetno;
1829         u32 su_per_object;
1830         u64 t, su_offset;
1831
1832         dout("mapping %llu~%llu  osize %u fl_su %u\n", off, len,
1833              osize, su);
1834         if (su == 0 || sc == 0)
1835                 goto invalid;
1836         su_per_object = osize / su;
1837         if (su_per_object == 0)
1838                 goto invalid;
1839         dout("osize %u / su %u = su_per_object %u\n", osize, su,
1840              su_per_object);
1841
1842         if ((su & ~PAGE_MASK) != 0)
1843                 goto invalid;
1844
1845         /* bl = *off / su; */
1846         t = off;
1847         do_div(t, su);
1848         bl = t;
1849         dout("off %llu / su %u = bl %u\n", off, su, bl);
1850
1851         stripeno = bl / sc;
1852         stripepos = bl % sc;
1853         objsetno = stripeno / su_per_object;
1854
1855         *ono = objsetno * sc + stripepos;
1856         dout("objset %u * sc %u = ono %u\n", objsetno, sc, (unsigned int)*ono);
1857
1858         /* *oxoff = *off % layout->fl_stripe_unit;  # offset in su */
1859         t = off;
1860         su_offset = do_div(t, su);
1861         *oxoff = su_offset + (stripeno % su_per_object) * su;
1862
1863         /*
1864          * Calculate the length of the extent being written to the selected
1865          * object. This is the minimum of the full length requested (len) or
1866          * the remainder of the current stripe being written to.
1867          */
1868         *oxlen = min_t(u64, len, su - su_offset);
1869
1870         dout(" obj extent %llu~%llu\n", *oxoff, *oxlen);
1871         return 0;
1872
1873 invalid:
1874         dout(" invalid layout\n");
1875         *ono = 0;
1876         *oxoff = 0;
1877         *oxlen = 0;
1878         return -EINVAL;
1879 }
1880 EXPORT_SYMBOL(ceph_calc_file_object_mapping);
1881
1882 /*
1883  * Map an object into a PG.
1884  *
1885  * Should only be called with target_oid and target_oloc (as opposed to
1886  * base_oid and base_oloc), since tiering isn't taken into account.
1887  */
1888 int ceph_object_locator_to_pg(struct ceph_osdmap *osdmap,
1889                               struct ceph_object_id *oid,
1890                               struct ceph_object_locator *oloc,
1891                               struct ceph_pg *raw_pgid)
1892 {
1893         struct ceph_pg_pool_info *pi;
1894
1895         pi = ceph_pg_pool_by_id(osdmap, oloc->pool);
1896         if (!pi)
1897                 return -ENOENT;
1898
1899         if (!oloc->pool_ns) {
1900                 raw_pgid->pool = oloc->pool;
1901                 raw_pgid->seed = ceph_str_hash(pi->object_hash, oid->name,
1902                                              oid->name_len);
1903                 dout("%s %s -> raw_pgid %llu.%x\n", __func__, oid->name,
1904                      raw_pgid->pool, raw_pgid->seed);
1905         } else {
1906                 char stack_buf[256];
1907                 char *buf = stack_buf;
1908                 int nsl = oloc->pool_ns->len;
1909                 size_t total = nsl + 1 + oid->name_len;
1910
1911                 if (total > sizeof(stack_buf)) {
1912                         buf = kmalloc(total, GFP_NOIO);
1913                         if (!buf)
1914                                 return -ENOMEM;
1915                 }
1916                 memcpy(buf, oloc->pool_ns->str, nsl);
1917                 buf[nsl] = '\037';
1918                 memcpy(buf + nsl + 1, oid->name, oid->name_len);
1919                 raw_pgid->pool = oloc->pool;
1920                 raw_pgid->seed = ceph_str_hash(pi->object_hash, buf, total);
1921                 if (buf != stack_buf)
1922                         kfree(buf);
1923                 dout("%s %s ns %.*s -> raw_pgid %llu.%x\n", __func__,
1924                      oid->name, nsl, oloc->pool_ns->str,
1925                      raw_pgid->pool, raw_pgid->seed);
1926         }
1927         return 0;
1928 }
1929 EXPORT_SYMBOL(ceph_object_locator_to_pg);
1930
1931 /*
1932  * Map a raw PG (full precision ps) into an actual PG.
1933  */
1934 static void raw_pg_to_pg(struct ceph_pg_pool_info *pi,
1935                          const struct ceph_pg *raw_pgid,
1936                          struct ceph_pg *pgid)
1937 {
1938         pgid->pool = raw_pgid->pool;
1939         pgid->seed = ceph_stable_mod(raw_pgid->seed, pi->pg_num,
1940                                      pi->pg_num_mask);
1941 }
1942
1943 /*
1944  * Map a raw PG (full precision ps) into a placement ps (placement
1945  * seed).  Include pool id in that value so that different pools don't
1946  * use the same seeds.
1947  */
1948 static u32 raw_pg_to_pps(struct ceph_pg_pool_info *pi,
1949                          const struct ceph_pg *raw_pgid)
1950 {
1951         if (pi->flags & CEPH_POOL_FLAG_HASHPSPOOL) {
1952                 /* hash pool id and seed so that pool PGs do not overlap */
1953                 return crush_hash32_2(CRUSH_HASH_RJENKINS1,
1954                                       ceph_stable_mod(raw_pgid->seed,
1955                                                       pi->pgp_num,
1956                                                       pi->pgp_num_mask),
1957                                       raw_pgid->pool);
1958         } else {
1959                 /*
1960                  * legacy behavior: add ps and pool together.  this is
1961                  * not a great approach because the PGs from each pool
1962                  * will overlap on top of each other: 0.5 == 1.4 ==
1963                  * 2.3 == ...
1964                  */
1965                 return ceph_stable_mod(raw_pgid->seed, pi->pgp_num,
1966                                        pi->pgp_num_mask) +
1967                        (unsigned)raw_pgid->pool;
1968         }
1969 }
1970
1971 static int do_crush(struct ceph_osdmap *map, int ruleno, int x,
1972                     int *result, int result_max,
1973                     const __u32 *weight, int weight_max)
1974 {
1975         int r;
1976
1977         BUG_ON(result_max > CEPH_PG_MAX_SIZE);
1978
1979         mutex_lock(&map->crush_workspace_mutex);
1980         r = crush_do_rule(map->crush, ruleno, x, result, result_max,
1981                           weight, weight_max, map->crush_workspace);
1982         mutex_unlock(&map->crush_workspace_mutex);
1983
1984         return r;
1985 }
1986
1987 /*
1988  * Calculate raw set (CRUSH output) for given PG.  The result may
1989  * contain nonexistent OSDs.  ->primary is undefined for a raw set.
1990  *
1991  * Placement seed (CRUSH input) is returned through @ppps.
1992  */
1993 static void pg_to_raw_osds(struct ceph_osdmap *osdmap,
1994                            struct ceph_pg_pool_info *pi,
1995                            const struct ceph_pg *raw_pgid,
1996                            struct ceph_osds *raw,
1997                            u32 *ppps)
1998 {
1999         u32 pps = raw_pg_to_pps(pi, raw_pgid);
2000         int ruleno;
2001         int len;
2002
2003         ceph_osds_init(raw);
2004         if (ppps)
2005                 *ppps = pps;
2006
2007         ruleno = crush_find_rule(osdmap->crush, pi->crush_ruleset, pi->type,
2008                                  pi->size);
2009         if (ruleno < 0) {
2010                 pr_err("no crush rule: pool %lld ruleset %d type %d size %d\n",
2011                        pi->id, pi->crush_ruleset, pi->type, pi->size);
2012                 return;
2013         }
2014
2015         if (pi->size > ARRAY_SIZE(raw->osds)) {
2016                 pr_err_ratelimited("pool %lld ruleset %d type %d too wide: size %d > %zu\n",
2017                        pi->id, pi->crush_ruleset, pi->type, pi->size,
2018                        ARRAY_SIZE(raw->osds));
2019                 return;
2020         }
2021
2022         len = do_crush(osdmap, ruleno, pps, raw->osds, pi->size,
2023                        osdmap->osd_weight, osdmap->max_osd);
2024         if (len < 0) {
2025                 pr_err("error %d from crush rule %d: pool %lld ruleset %d type %d size %d\n",
2026                        len, ruleno, pi->id, pi->crush_ruleset, pi->type,
2027                        pi->size);
2028                 return;
2029         }
2030
2031         raw->size = len;
2032 }
2033
2034 /*
2035  * Given raw set, calculate up set and up primary.  By definition of an
2036  * up set, the result won't contain nonexistent or down OSDs.
2037  *
2038  * This is done in-place - on return @set is the up set.  If it's
2039  * empty, ->primary will remain undefined.
2040  */
2041 static void raw_to_up_osds(struct ceph_osdmap *osdmap,
2042                            struct ceph_pg_pool_info *pi,
2043                            struct ceph_osds *set)
2044 {
2045         int i;
2046
2047         /* ->primary is undefined for a raw set */
2048         BUG_ON(set->primary != -1);
2049
2050         if (ceph_can_shift_osds(pi)) {
2051                 int removed = 0;
2052
2053                 /* shift left */
2054                 for (i = 0; i < set->size; i++) {
2055                         if (ceph_osd_is_down(osdmap, set->osds[i])) {
2056                                 removed++;
2057                                 continue;
2058                         }
2059                         if (removed)
2060                                 set->osds[i - removed] = set->osds[i];
2061                 }
2062                 set->size -= removed;
2063                 if (set->size > 0)
2064                         set->primary = set->osds[0];
2065         } else {
2066                 /* set down/dne devices to NONE */
2067                 for (i = set->size - 1; i >= 0; i--) {
2068                         if (ceph_osd_is_down(osdmap, set->osds[i]))
2069                                 set->osds[i] = CRUSH_ITEM_NONE;
2070                         else
2071                                 set->primary = set->osds[i];
2072                 }
2073         }
2074 }
2075
2076 static void apply_primary_affinity(struct ceph_osdmap *osdmap,
2077                                    struct ceph_pg_pool_info *pi,
2078                                    u32 pps,
2079                                    struct ceph_osds *up)
2080 {
2081         int i;
2082         int pos = -1;
2083
2084         /*
2085          * Do we have any non-default primary_affinity values for these
2086          * osds?
2087          */
2088         if (!osdmap->osd_primary_affinity)
2089                 return;
2090
2091         for (i = 0; i < up->size; i++) {
2092                 int osd = up->osds[i];
2093
2094                 if (osd != CRUSH_ITEM_NONE &&
2095                     osdmap->osd_primary_affinity[osd] !=
2096                                         CEPH_OSD_DEFAULT_PRIMARY_AFFINITY) {
2097                         break;
2098                 }
2099         }
2100         if (i == up->size)
2101                 return;
2102
2103         /*
2104          * Pick the primary.  Feed both the seed (for the pg) and the
2105          * osd into the hash/rng so that a proportional fraction of an
2106          * osd's pgs get rejected as primary.
2107          */
2108         for (i = 0; i < up->size; i++) {
2109                 int osd = up->osds[i];
2110                 u32 aff;
2111
2112                 if (osd == CRUSH_ITEM_NONE)
2113                         continue;
2114
2115                 aff = osdmap->osd_primary_affinity[osd];
2116                 if (aff < CEPH_OSD_MAX_PRIMARY_AFFINITY &&
2117                     (crush_hash32_2(CRUSH_HASH_RJENKINS1,
2118                                     pps, osd) >> 16) >= aff) {
2119                         /*
2120                          * We chose not to use this primary.  Note it
2121                          * anyway as a fallback in case we don't pick
2122                          * anyone else, but keep looking.
2123                          */
2124                         if (pos < 0)
2125                                 pos = i;
2126                 } else {
2127                         pos = i;
2128                         break;
2129                 }
2130         }
2131         if (pos < 0)
2132                 return;
2133
2134         up->primary = up->osds[pos];
2135
2136         if (ceph_can_shift_osds(pi) && pos > 0) {
2137                 /* move the new primary to the front */
2138                 for (i = pos; i > 0; i--)
2139                         up->osds[i] = up->osds[i - 1];
2140                 up->osds[0] = up->primary;
2141         }
2142 }
2143
2144 /*
2145  * Get pg_temp and primary_temp mappings for given PG.
2146  *
2147  * Note that a PG may have none, only pg_temp, only primary_temp or
2148  * both pg_temp and primary_temp mappings.  This means @temp isn't
2149  * always a valid OSD set on return: in the "only primary_temp" case,
2150  * @temp will have its ->primary >= 0 but ->size == 0.
2151  */
2152 static void get_temp_osds(struct ceph_osdmap *osdmap,
2153                           struct ceph_pg_pool_info *pi,
2154                           const struct ceph_pg *raw_pgid,
2155                           struct ceph_osds *temp)
2156 {
2157         struct ceph_pg pgid;
2158         struct ceph_pg_mapping *pg;
2159         int i;
2160
2161         raw_pg_to_pg(pi, raw_pgid, &pgid);
2162         ceph_osds_init(temp);
2163
2164         /* pg_temp? */
2165         pg = __lookup_pg_mapping(&osdmap->pg_temp, pgid);
2166         if (pg) {
2167                 for (i = 0; i < pg->pg_temp.len; i++) {
2168                         if (ceph_osd_is_down(osdmap, pg->pg_temp.osds[i])) {
2169                                 if (ceph_can_shift_osds(pi))
2170                                         continue;
2171
2172                                 temp->osds[temp->size++] = CRUSH_ITEM_NONE;
2173                         } else {
2174                                 temp->osds[temp->size++] = pg->pg_temp.osds[i];
2175                         }
2176                 }
2177
2178                 /* apply pg_temp's primary */
2179                 for (i = 0; i < temp->size; i++) {
2180                         if (temp->osds[i] != CRUSH_ITEM_NONE) {
2181                                 temp->primary = temp->osds[i];
2182                                 break;
2183                         }
2184                 }
2185         }
2186
2187         /* primary_temp? */
2188         pg = __lookup_pg_mapping(&osdmap->primary_temp, pgid);
2189         if (pg)
2190                 temp->primary = pg->primary_temp.osd;
2191 }
2192
2193 /*
2194  * Map a PG to its acting set as well as its up set.
2195  *
2196  * Acting set is used for data mapping purposes, while up set can be
2197  * recorded for detecting interval changes and deciding whether to
2198  * resend a request.
2199  */
2200 void ceph_pg_to_up_acting_osds(struct ceph_osdmap *osdmap,
2201                                const struct ceph_pg *raw_pgid,
2202                                struct ceph_osds *up,
2203                                struct ceph_osds *acting)
2204 {
2205         struct ceph_pg_pool_info *pi;
2206         u32 pps;
2207
2208         pi = ceph_pg_pool_by_id(osdmap, raw_pgid->pool);
2209         if (!pi) {
2210                 ceph_osds_init(up);
2211                 ceph_osds_init(acting);
2212                 goto out;
2213         }
2214
2215         pg_to_raw_osds(osdmap, pi, raw_pgid, up, &pps);
2216         raw_to_up_osds(osdmap, pi, up);
2217         apply_primary_affinity(osdmap, pi, pps, up);
2218         get_temp_osds(osdmap, pi, raw_pgid, acting);
2219         if (!acting->size) {
2220                 memcpy(acting->osds, up->osds, up->size * sizeof(up->osds[0]));
2221                 acting->size = up->size;
2222                 if (acting->primary == -1)
2223                         acting->primary = up->primary;
2224         }
2225 out:
2226         WARN_ON(!osds_valid(up) || !osds_valid(acting));
2227 }
2228
2229 /*
2230  * Return acting primary for given PG, or -1 if none.
2231  */
2232 int ceph_pg_to_acting_primary(struct ceph_osdmap *osdmap,
2233                               const struct ceph_pg *raw_pgid)
2234 {
2235         struct ceph_osds up, acting;
2236
2237         ceph_pg_to_up_acting_osds(osdmap, raw_pgid, &up, &acting);
2238         return acting.primary;
2239 }
2240 EXPORT_SYMBOL(ceph_pg_to_acting_primary);