compare.c

map_with_crush():

void map_with_crush(int replication_count, int hosts_count, int object_map[][NUMBER_OF_OBJECTS]) {
  struct crush_map *m = crush_create();
  m->choose_local_tries = 0;
  m->choose_local_fallback_tries = 0;
  m->choose_total_tries = 50;
  m->chooseleaf_descend_once = 1;
  m->chooseleaf_vary_r = 1;
  m->chooseleaf_stable = 1;
  m->allowed_bucket_algs =
    (1 << CRUSH_BUCKET_UNIFORM) |
    (1 << CRUSH_BUCKET_LIST) |
    (1 << CRUSH_BUCKET_STRAW2);
  int root_type = 1;
  int host_type = 2;  
  int bucketno = 0;

  int hosts[hosts_count];
  int weights[hosts_count];
  int disk = 0;
  int weight = 0x10000;  // 1.0
  // 创建一个多个host类型的bucket，并给每个bucket添加一个disk的item
  for(int host = 0; host < hosts_count; host++) {
    struct crush_bucket *b;
    // 函数创建一个bucket结构。需要指定CRUSH算法(uniform、list、tree、straw)、hash算法、
    // bucket类型(自定义的type)、size(包括多少个item)、这些items的id数组、这些items的weights
    b = crush_make_bucket(m, CRUSH_BUCKET_STRAW2, CRUSH_HASH_DEFAULT, host_type,
                          0, NULL, NULL);
    assert(b != NULL);
    // 往bucket中添加item, disk为item的id，weight为item的权重
    assert(crush_bucket_add_item(m, b, disk, weight) == 0);
    // 将b添加到m中，m->buckets[],bucket的id为bucketno
    assert(crush_add_bucket(m, 0, b, &bucketno) == 0);
    hosts[host] = bucketno;
    weights[host] = weight;
    disk++;
  }

  struct crush_bucket *root;
  int bucket_root;
  // 创建一个root类型的bucket，bucket中添加的item是hosts
  root = crush_make_bucket(m, CRUSH_BUCKET_STRAW2, CRUSH_HASH_DEFAULT, root_type,
                           hosts_count, hosts, weights);
  assert(root != NULL);
  // 将bucket添加带map中
  assert(crush_add_bucket(m, 0, root, &bucket_root) == 0);
  // 调整权重分布
  assert(crush_reweight_bucket(m, root) == 0);
  
  struct crush_rule *r;
  int minsize = 1;
  int maxsize = 5;
  int number_of_steps = 3;
  r = crush_make_rule(number_of_steps, 0, 0, minsize, maxsize);
  assert(r != NULL);
  crush_rule_set_step(r, 0, CRUSH_RULE_TAKE, bucket_root, 0);
  crush_rule_set_step(r, 1, CRUSH_RULE_CHOOSELEAF_FIRSTN, replication_count, host_type);
  crush_rule_set_step(r, 2, CRUSH_RULE_EMIT, 0, 0);
  int ruleno = crush_add_rule(m, r, -1);
  assert(ruleno >= 0);

  crush_finalize(m);

  {
    int result[replication_count];
    __u32 weights[hosts_count];
    for(int i = 0; i < hosts_count; i++)
      weights[i] = 0x10000;
    int cwin_size = crush_work_size(m, replication_count);
    char cwin[cwin_size];
    crush_init_workspace(m, cwin);
    for(int x = 0; x < NUMBER_OF_OBJECTS; x++) {
      memset(result, '', sizeof(int) * replication_count);
      assert(crush_do_rule(m, ruleno, x, result, replication_count, weights, hosts_count, cwin) == replication_count);
      for(int i = 0; i < replication_count; i++) {
        object_map[i][x] = result[i];
      }
    }
  }
  crush_destroy(m);
}

构造crush map

1. 创建crush map

 struct crush_map *m = crush_create();

struct crush_map *crush_create()
{
    struct crush_map *m;
    m = malloc(sizeof(*m));
        if (!m)
                return NULL;
    memset(m, 0, sizeof(*m));

    /* initialize legacy tunable values */
    m->choose_local_tries = 2;
    m->choose_local_fallback_tries = 5;
    m->choose_total_tries = 19;
    m->chooseleaf_descend_once = 0;
    m->chooseleaf_vary_r = 0;
    m->straw_calc_version = 0;

    // by default, use legacy types, and also exclude tree,
    // since it was buggy.
    m->allowed_bucket_algs = CRUSH_LEGACY_ALLOWED_BUCKET_ALGS;
    return m;
}

2. 添加bucket

 分为两步：先make，后add

 // 创建一个多个host类型的bucket，并给每个bucket添加一个disk的item
  for(int host = 0; host < hosts_count; host++) {
    struct crush_bucket *b;
    // 函数创建一个bucket结构。需要指定CRUSH算法(uniform、list、tree、straw)、hash算法、
    // bucket类型(自定义的type)、size(包括多少个item)、这些items的id数组、这些items的weights
    b = crush_make_bucket(m, CRUSH_BUCKET_STRAW2, CRUSH_HASH_DEFAULT, host_type,
                          0, NULL, NULL);
    assert(b != NULL);
    // 往bucket中添加item, disk为item的id，weight为item的权重
    assert(crush_bucket_add_item(m, b, disk, weight) == 0);
    // 将b添加到m中，m->buckets[],bucket的id为bucketno
    assert(crush_add_bucket(m, 0, b, &bucketno) == 0);
    hosts[host] = bucketno;
    weights[host] = weight;
    disk++;
  }

2.1 make bucket

先看一下bucket的数据结构：

/** @ingroup API
 *
 * A bucket contains __size__ __items__ which are either positive
 * numbers or negative numbers that reference other buckets and is
 * uniquely identified with __id__ which is a negative number.  The
 * __weight__ of a bucket is the cumulative weight of all its
 * children.  A bucket is assigned a ::crush_algorithm that is used by
 * crush_do_rule() to draw an item depending on its weight.  A bucket
 * can be assigned a strictly positive (> 0) __type__ defined by the
 * caller. The __type__ can be used by crush_do_rule(), when it is
 * given as an argument of a rule step.
 *
 * A pointer to crush_bucket can safely be cast into the following
 * structure, depending on the value of __alg__:
 *
 * - __alg__ == ::CRUSH_BUCKET_UNIFORM cast to crush_bucket_uniform
 * - __alg__ == ::CRUSH_BUCKET_LIST cast to crush_bucket_list
 * - __alg__ == ::CRUSH_BUCKET_STRAW2 cast to crush_bucket_straw2
 *
 * The weight of each item depends on the algorithm and the
 * information about it is available in the corresponding structure
 * (crush_bucket_uniform, crush_bucket_list or crush_bucket_straw2).
 *
 * See crush_map for more information on how __id__ is used
 * to reference the bucket.
 */
struct crush_bucket {
    __s32 id;        /*!< bucket identifier, < 0 and unique within a crush_map 小于0的唯一标记 */
    __u16 type;      /*!< > 0 bucket type, defined by the caller > 0的值，用户自定义的类型 */
    __u8 alg;        /*!< the item selection ::crush_algorithm 所采用的crush算法 */
        /*! @cond INTERNAL */
    __u8 hash;       /* which hash function to use, CRUSH_HASH_* 使用的hash算法类型 */
    /*! @endcond */
    __u32 weight;    /*!< 16.16 fixed point cumulated children weight */
    __u32 size;      /*!< size of the __items__ array，如果size为0，则不包含item*/
        __s32 *items;    /*!< array of children: < 0 are buckets, >= 0 items， 这是个数组，
        记录所有的item， 如果item的值是负数，则为buket， 为自然数则是device */
};

产生一个bucket：

b = crush_make_bucket(m, CRUSH_BUCKET_STRAW2, CRUSH_HASH_DEFAULT, host_type, 0, NULL, NULL);

struct crush_bucket*
crush_make_bucket(struct crush_map *map,
          int alg, int hash, int type, int size,
          int *items,
          int *weights)
{
    int item_weight;

    switch (alg) {
    case CRUSH_BUCKET_UNIFORM:
        if (size && weights)
            item_weight = weights[0];
        else
            item_weight = 0;
        return (struct crush_bucket *)crush_make_uniform_bucket(hash, type, size, items, item_weight);

    case CRUSH_BUCKET_LIST:
        return (struct crush_bucket *)crush_make_list_bucket(hash, type, size, items, weights);

    case CRUSH_BUCKET_TREE:
        return (struct crush_bucket *)crush_make_tree_bucket(hash, type, size, items, weights);

    case CRUSH_BUCKET_STRAW:
        return (struct crush_bucket *)crush_make_straw_bucket(map, hash, type, size, items, weights);
    case CRUSH_BUCKET_STRAW2:
        return (struct crush_bucket *)crush_make_straw2_bucket(map, hash, type, size, items, weights);
    }
    return 0;
}

struct crush_bucket_straw2 *
crush_make_straw2_bucket(struct crush_map *map,
             int hash,          // hash值
             int type,          // 类型
             int size,          // item的大小
             int *items,        // items
             int *weights)      // item的weight值
{
    struct crush_bucket_straw2 *bucket;
    int i;

    bucket = malloc(sizeof(*bucket));
        if (!bucket)
                return NULL;
    memset(bucket, 0, sizeof(*bucket));
    bucket->h.alg = CRUSH_BUCKET_STRAW2;
    bucket->h.hash = hash;
    bucket->h.type = type;
    bucket->h.size = size;

    bucket->h.items = malloc(sizeof(__s32)*size);
        if (!bucket->h.items)
                goto err;
    bucket->item_weights = malloc(sizeof(__u32)*size);
        if (!bucket->item_weights)
                goto err;

    bucket->h.weight = 0;
    for (i=0; i<size; i++) {
        bucket->h.items[i] = items[i];        // item中是item的id
        bucket->h.weight += weights[i];       // 总的weight值
        bucket->item_weights[i] = weights[i];
    }

    return bucket;
err:
        free(bucket->item_weights);
        free(bucket->h.items);
        free(bucket);
        return NULL;
}

2.2 向bucket中添加item

入参：

1. struct crush_map *map  map

2. struct crush_bucket *b   bucket

3. int item       item的id

4. int weight   item的weight

assert(crush_bucket_add_item(m, b, disk, weight) == 0);

int crush_bucket_add_item(struct crush_map *map,
              struct crush_bucket *b, int item, int weight)
{
    switch (b->alg) {
    case CRUSH_BUCKET_UNIFORM:
        return crush_add_uniform_bucket_item((struct crush_bucket_uniform *)b, item, weight);
    case CRUSH_BUCKET_LIST:
        return crush_add_list_bucket_item((struct crush_bucket_list *)b, item, weight);
    case CRUSH_BUCKET_TREE:
        return crush_add_tree_bucket_item((struct crush_bucket_tree *)b, item, weight);
    case CRUSH_BUCKET_STRAW:
        return crush_add_straw_bucket_item(map, (struct crush_bucket_straw *)b, item, weight);
    case CRUSH_BUCKET_STRAW2:
        return crush_add_straw2_bucket_item(map, (struct crush_bucket_straw2 *)b, item, weight);
    default:
        return -1;
    }
}

int crush_add_straw2_bucket_item(struct crush_map *map,
                 struct crush_bucket_straw2 *bucket,
                 int item, int weight)
{
    int newsize = bucket->h.size + 1;          // 添加一个item

    void *_realloc = NULL;

    if ((_realloc = realloc(bucket->h.items, sizeof(__s32)*newsize)) == NULL) {
        return -ENOMEM;
    } else {
        bucket->h.items = _realloc;
    }
    if ((_realloc = realloc(bucket->item_weights, sizeof(__u32)*newsize)) == NULL) {
        return -ENOMEM;
    } else {
        bucket->item_weights = _realloc;
    }

    bucket->h.items[newsize-1] = item;
    bucket->item_weights[newsize-1] = weight;

    if (crush_addition_is_unsafe(bucket->h.weight, weight))
                return -ERANGE;

    bucket->h.weight += weight;
    bucket->h.size++;

    return 0;
}

2.3  把bucket添加到map中

入参：

1. struct crush_map *map  

2. int id   bucket的id

3. struct crush_bucket *bucket   bucket的结构体

4. int *idout 和id值一致

crush_add_bucket(m, 0, b, &bucketno) == 0

int crush_add_bucket(struct crush_map *map,
             int id,
             struct crush_bucket *bucket,
             int *idout)
{
    int pos;

    /* find a bucket id */
    if (id == 0)
        id = crush_get_next_bucket_id(map);
    pos = -1 - id;

    while (pos >= map->max_buckets) {     // 扩充map的bucket数组大小
        /* expand array */
        int oldsize = map->max_buckets;
        if (map->max_buckets)             // 如果已经设置最大的bucket个数，则加倍扩充
            map->max_buckets *= 2;
        else
            map->max_buckets = 8;         // 未进行设置，则默认为8
        void *_realloc = NULL;
        if ((_realloc = realloc(map->buckets, map->max_buckets * sizeof(map->buckets[0]))) == NULL) {
            return -ENOMEM; 
        } else {
            map->buckets = _realloc;
        }
        memset(map->buckets + oldsize, 0, (map->max_buckets-oldsize) * sizeof(map->buckets[0]));  // 初始化
    }

    if (map->buckets[pos] != 0) {   // 判断是否已经存在
        return -EEXIST;
    }

    /* add it */
    bucket->id = id;
    map->buckets[pos] = bucket;

    if (idout) *idout = id;    // 返回id
    return 0;
}

2.4 构造root bucket

构造一个root类型的bucket，并将hosts添加到该bucket中；

root = crush_make_bucket(m, CRUSH_BUCKET_STRAW2, CRUSH_HASH_DEFAULT, root_type, hosts_count, hosts, weights);

// 入参
// struct crush_map *map
// int alg 算法类型 CRUSH_BUCKET_STRAW2
// int hash 选择使用的是那种hash函数类型
// int type类型，root type
// 包含的item的大小， hosts_count
// int *items item的id数组 hosts
// int *weights item的weight集合 weights
struct crush_bucket*
crush_make_bucket(struct crush_map *map,
          int alg, int hash, int type, int size,
          int *items,
          int *weights)

// 将bucket添加带map中
assert(crush_add_bucket(m, 0, root, &bucket_root) == 0);

// 调整权重分布
assert(crush_reweight_bucket(m, root) == 0);

int crush_reweight_bucket(struct crush_map *crush, struct crush_bucket *b)
{
    switch (b->alg) {
    case CRUSH_BUCKET_UNIFORM:
        return crush_reweight_uniform_bucket(crush, (struct crush_bucket_uniform *)b);
    case CRUSH_BUCKET_LIST:
        return crush_reweight_list_bucket(crush, (struct crush_bucket_list *)b);
    case CRUSH_BUCKET_TREE:
        return crush_reweight_tree_bucket(crush, (struct crush_bucket_tree *)b);
    case CRUSH_BUCKET_STRAW:
        return crush_reweight_straw_bucket(crush, (struct crush_bucket_straw *)b);
    case CRUSH_BUCKET_STRAW2:
        return crush_reweight_straw2_bucket(crush, (struct crush_bucket_straw2 *)b);
    default:
        return -1;
    }
}

static int crush_reweight_straw2_bucket(struct crush_map *crush, struct crush_bucket_straw2 *bucket)
{
    unsigned i;

    bucket->h.weight = 0;
    for (i = 0; i < bucket->h.size; i++) {
        int id = bucket->h.items[i];
        if (id < 0) {
            struct crush_bucket *c = crush->buckets[-1-id];
            crush_reweight_bucket(crush, c);        // 递归调用，逐级调整其实就是重新计算下每一层的bucket的weight
            bucket->item_weights[i] = c->weight;
        }

        if (crush_addition_is_unsafe(bucket->h.weight, bucket->item_weights[i]))
            return -ERANGE;

        bucket->h.weight += bucket->item_weights[i];
    }

    return 0;
}

至此，crush_map已经创建完成

梳理一下：

1. 创建一个空map；

2. 创建host类型的bucket，向bucket添加disk类型的item； bucket, bucket->h.items[]

3. 将host添加到bucket中； map->buckets[]

4. 创建root类型的bucket，将host类型的bucket添加到root下，并调整相应的weight值；

接下来分析ruleset规则的创建，它决定了rule的选路规则

rule的数据结构

/*
 * The rule mask is used to describe what the rule is intended for.
 * Given a ruleset and size of output set, we search through the
 * rule list for a matching rule_mask.
 */
struct crush_rule_mask {
    __u8 ruleset;  // ruleset的编号
    __u8 type;     // 类型
    __u8 min_size;
    __u8 max_size;
};

struct crush_rule {
    __u32 len;   //steps数组的长度
    struct crush_rule_mask mask;      // ruleset的相关配置参数
    struct crush_rule_step steps[0];  // 每一个操作步骤，操作类型+参数
};

1. 首先，创建一个rule的数据结构，用于存储rule信息

r = crush_make_rule(number_of_steps, 0, 0, minsize, maxsize);

// step的number，ruleset的id，type的id
struct crush_rule *crush_make_rule(int len, int ruleset, int type, int minsize, int maxsize)
{
    struct crush_rule *rule;
    rule = malloc(crush_rule_size(len));
        if (!rule)
                return NULL;
    rule->len = len;
    rule->mask.ruleset = ruleset;
    rule->mask.type = type;
    rule->mask.min_size = minsize;
    rule->mask.max_size = maxsize;
    return rule;
}

2. 接着，添加step：

crush_rule_set_step(r, 0, CRUSH_RULE_TAKE, bucket_root, 0);

crush_rule_set_step(r, 1, CRUSH_RULE_CHOOSELEAF_FIRSTN, replication_count, host_type);

crush_rule_set_step(r, 2, CRUSH_RULE_EMIT, 0, 0);

/*  入参：n为step的编号，op为操作码，arg为参数
 * be careful; this doesn't verify that the buffer you allocated is big enough!
 */
void crush_rule_set_step(struct crush_rule *rule, int n, int op, int arg1, int arg2)
{
    assert((__u32)n < rule->len);
    rule->steps[n].op = op;
    rule->steps[n].arg1 = arg1;
    rule->steps[n].arg2 = arg2;
}

3. 添加rule：

int crush_add_rule(struct crush_map *map, struct crush_rule *rule, int ruleno)
{
    __u32 r;
    // 如果ruleno小于0，则自己分配连续的rule，否则，按照指定的ruleno
    if (ruleno < 0) {
        for (r=0; r < map->max_rules; r++)
            if (map->rules[r] == 0)
                break;
        assert(r < CRUSH_MAX_RULES);
    }
    else
        r = ruleno;

    if (r >= map->max_rules) {
        /* expand array */
        int oldsize;
        void *_realloc = NULL;
        if (map->max_rules +1 > CRUSH_MAX_RULES)
            return -ENOSPC;
        oldsize = map->max_rules;
        map->max_rules = r+1;  
        if ((_realloc = realloc(map->rules, map->max_rules * sizeof(map->rules[0]))) == NULL) {
            return -ENOMEM; 
        } else {
            map->rules = _realloc;
        } 
        memset(map->rules + oldsize, 0, (map->max_rules-oldsize) * sizeof(map->rules[0]));
    }

    /* add it */
    map->rules[r] = rule;
    return r;
}

看一下map的数据结构：

/** @ingroup API
 *
 * A crush map define a hierarchy of crush_bucket that end with leaves
 * (buckets and leaves are called items) and a set of crush_rule to
 * map an integer to items with the crush_do_rule() function.
 *
 */
struct crush_map {
        /*! An array of crush_bucket pointers of size __max_buckets__.
         * An element of the array may be NULL if the bucket was removed with
         * crush_remove_bucket(). The buckets must be added with crush_add_bucket().
         * The bucket found at __buckets[i]__ must have a crush_bucket.id == -1-i.
         */
    struct crush_bucket **buckets;     // 动态二维数组，保存所有的bucket结构
        /*! An array of crush_rule pointers of size __max_rules__.
         * An element of the array may be NULL if the rule was removed (there is
         * no API to do so but there may be one in the future). The rules must be added
         * with crush_add_rule().
         */
    struct crush_rule **rules;        // 保存所有的Crush_rule 
        __s32 max_buckets; /*!< the size of __buckets__ */
    __u32 max_rules; /*!< the size of __rules__ */
        /*! The value of the highest item stored in the crush_map + 1
         */
    __s32 max_devices;

    /*! choose local retries before re-descent */
    __u32 choose_local_tries;
    /*! choose local attempts using a fallback permutation before
     *! re-descent */
    __u32 choose_local_fallback_tries;
    /*! choose attempts before giving up */
    __u32 choose_total_tries;
    /*! attempt chooseleaf inner descent once for firstn mode; on
     *! reject retry outer descent.  Note that this does *not*
     *! apply to a collision: in that case we will retry as we used
     *! to. */
    __u32 chooseleaf_descend_once;

    /*! if non-zero, feed r into chooseleaf, bit-shifted right by (r-1)
     *! bits.  a value of 1 is best for new clusters.  for legacy clusters
     *! that want to limit reshuffling, a value of 3 or 4 will make the
     *! mappings line up a bit better with previous mappings. */
    __u8 chooseleaf_vary_r;

    /*! if true, it makes chooseleaf firstn to return stable results (if
     *! no local retry) so that data migrations would be optimal when some
     *! device fails. */
    __u8 chooseleaf_stable;

        /*! @cond INTERNAL */
    /* This value is calculated after decode or construction by
       the builder. It is exposed here (rather than having a
       'build CRUSH working space' function) so that callers can
       reserve a static buffer, allocate space on the stack, or
       otherwise avoid calling into the heap allocator if they
       want to. The size of the working space depends on the map,
       while the size of the scratch vector passed to the mapper
       depends on the size of the desired result set.

       Nothing stops the caller from allocating both in one swell
       foop and passing in two points, though. */
    size_t working_size;

#ifndef __KERNEL__
    /*
     * version 0 (original) of straw_calc has various flaws.  version 1
     * fixes a few of them.
     */
    __u8 straw_calc_version;

    /*
     * allowed bucket algs is a bitmask, here the bit positions
     * are CRUSH_BUCKET_*.  note that these are *bits* and
     * CRUSH_BUCKET_* values are not, so we need to or together (1
     * << CRUSH_BUCKET_WHATEVER).  The 0th bit is not used to
     * minimize confusion (bucket type values start at 1).
     */
    __u32 allowed_bucket_algs;

    __u32 *choose_tries;
#endif
    /*! @endcond */
};

4. finalize rule

/* 更新working_size和max_devices
 * finalize should be called _after_ all buckets are added to the map.
 */
void crush_finalize(struct crush_map *map)
{
    int b;
    __u32 i;

    /* Calculate the needed working space while we do other
       finalization tasks. */
    map->working_size = sizeof(struct crush_work);
    /* Space for the array of pointers to per-bucket workspace */
    map->working_size += map->max_buckets *
        sizeof(struct crush_work_bucket *);

    /* calc max_devices */
    map->max_devices = 0;
    for (b=0; b<map->max_buckets; b++) {
        if (map->buckets[b] == 0)
            continue;
        // 更新map->max_devices
        for (i=0; i<map->buckets[b]->size; i++)
            if (map->buckets[b]->items[i] >= map->max_devices)
                map->max_devices = map->buckets[b]->items[i] + 1;

        switch (map->buckets[b]->alg) {
        default:
            /* The base case, permutation variables and
               the pointer to the permutation array. */
            map->working_size += sizeof(struct crush_work_bucket);
            break;
        }
        /* Every bucket has a permutation array. */
        map->working_size += map->buckets[b]->size * sizeof(__u32);
    }
}

如何调用

先初始化空间：

int cwin_size = crush_work_size(m, replication_count);  // 3倍空间，o和w

crush_init_workspace(m, cwin);

for(int x = 0; x < NUMBER_OF_OBJECTS; x++) {
      memset(result, '', sizeof(int) * replication_count);
      assert(crush_do_rule(m, ruleno, x, result, replication_count, weights, hosts_count, cwin) == replication_count);
      for(int i = 0; i < replication_count; i++) {
        object_map[i][x] = result[i];
      }
 }

crush do rule

/**
 * crush_do_rule - calculate a mapping with the given input and rule
 * @map: the crush_map
 * @ruleno: the rule id
 * @x: hash input
 * @result: pointer to result vector
 * @result_max: maximum result size
 * @weight: weight vector (for map leaves)
 * @weight_max: size of weight vector
 * @cwin: Pointer to at least map->working_size bytes of memory or NULL.
 */
int crush_do_rule(const struct crush_map *map,
          int ruleno, int x, int *result, int result_max,
          const __u32 *weight, int weight_max,
          void *cwin)
{
    int result_len;
    struct crush_work *cw = cwin;
    int *a = (int *)((char *)cw + map->working_size);  // 用了一个scratch空间来完成item的搜索
    int *b = a + result_max;
    int *c = b + result_max;
    int *w = a;
    int *o = b;
    int recurse_to_leaf;
    int wsize = 0;
    int osize;
    int *tmp;
    const struct crush_rule *rule;
    __u32 step;
    int i, j;
    int numrep;
    int out_size;
    /*
     * the original choose_total_tries value was off by one (it
     * counted "retries" and not "tries").  add one.
     */
    int choose_tries = map->choose_total_tries + 1;
    int choose_leaf_tries = 0;
    /*
     * the local tries values were counted as "retries", though,
     * and need no adjustment
     */
    int choose_local_retries = map->choose_local_tries;
    int choose_local_fallback_retries = map->choose_local_fallback_tries;

    int vary_r = map->chooseleaf_vary_r;
    int stable = map->chooseleaf_stable;

    if ((__u32)ruleno >= map->max_rules) {
        dprintk(" bad ruleno %d
", ruleno);
        return 0;
    }

    rule = map->rules[ruleno];
    result_len = 0;

    for (step = 0; step < rule->len; step++) {
        int firstn = 0;
        const struct crush_rule_step *curstep = &rule->steps[step];

        switch (curstep->op) {
        case CRUSH_RULE_TAKE:
            if ((curstep->arg1 >= 0 &&
                 curstep->arg1 < map->max_devices) ||
                (-1-curstep->arg1 >= 0 &&
                 -1-curstep->arg1 < map->max_buckets &&
                 map->buckets[-1-curstep->arg1])) {
                w[0] = curstep->arg1;
                wsize = 1;
            } else {
                dprintk(" bad take value %d
", curstep->arg1);
            }
            break;

        case CRUSH_RULE_SET_CHOOSE_TRIES:
            if (curstep->arg1 > 0)
                choose_tries = curstep->arg1;
            break;

        case CRUSH_RULE_SET_CHOOSELEAF_TRIES:
            if (curstep->arg1 > 0)
                choose_leaf_tries = curstep->arg1;
            break;

        case CRUSH_RULE_SET_CHOOSE_LOCAL_TRIES:
            if (curstep->arg1 >= 0)
                choose_local_retries = curstep->arg1;
            break;

        case CRUSH_RULE_SET_CHOOSE_LOCAL_FALLBACK_TRIES:
            if (curstep->arg1 >= 0)
                choose_local_fallback_retries = curstep->arg1;
            break;

        case CRUSH_RULE_SET_CHOOSELEAF_VARY_R:
            if (curstep->arg1 >= 0)
                vary_r = curstep->arg1;
            break;

        case CRUSH_RULE_SET_CHOOSELEAF_STABLE:
            if (curstep->arg1 >= 0)
                stable = curstep->arg1;
            break;

        case CRUSH_RULE_CHOOSELEAF_FIRSTN:
        case CRUSH_RULE_CHOOSE_FIRSTN:
            firstn = 1;
            /* fall through */
        case CRUSH_RULE_CHOOSELEAF_INDEP:
        case CRUSH_RULE_CHOOSE_INDEP:
            if (wsize == 0)
                break;

            recurse_to_leaf =
                curstep->op ==
                 CRUSH_RULE_CHOOSELEAF_FIRSTN ||
                curstep->op ==
                CRUSH_RULE_CHOOSELEAF_INDEP;

            /* reset output */
            osize = 0;

            for (i = 0; i < wsize; i++) {
                int bno;
                /*
                 * see CRUSH_N, CRUSH_N_MINUS macros.
                 * basically, numrep <= 0 means relative to
                 * the provided result_max
                 */
                numrep = curstep->arg1;
                if (numrep <= 0) {
                    numrep += result_max;
                    if (numrep <= 0)
                        continue;
                }
                j = 0;
                /* make sure bucket id is valid */
                bno = -1 - w[i];
                if (bno < 0 || bno >= map->max_buckets) {
                    // w[i] is probably CRUSH_ITEM_NONE
                    dprintk("  bad w[i] %d
", w[i]);
                    continue;
                }
                if (firstn) {
                    int recurse_tries;
                    if (choose_leaf_tries)
                        recurse_tries =
                            choose_leaf_tries;
                    else if (map->chooseleaf_descend_once)
                        recurse_tries = 1;
                    else
                        recurse_tries = choose_tries;
                    osize += crush_choose_firstn(
                        map,
                        cw,
                        map->buckets[bno],
                        weight, weight_max,
                        x, numrep,
                        curstep->arg2,
                        o+osize, j,
                        result_max-osize,
                        choose_tries,
                        recurse_tries,
                        choose_local_retries,
                        choose_local_fallback_retries,
                        recurse_to_leaf,
                        vary_r,
                        stable,
                        c+osize,
                        0);
                } else {
                    out_size = ((numrep < (result_max-osize)) ?
                            numrep : (result_max-osize));
                    crush_choose_indep(
                        map,
                        cw,
                        map->buckets[bno],
                        weight, weight_max,
                        x, out_size, numrep,
                        curstep->arg2,
                        o+osize, j,
                        choose_tries,
                        choose_leaf_tries ?
                           choose_leaf_tries : 1,
                        recurse_to_leaf,
                        c+osize,
                        0);
                    osize += out_size;
                }
            }

            if (recurse_to_leaf)
                /* copy final _leaf_ values to output set */
                memcpy(o, c, osize*sizeof(*o));

            /* swap o and w arrays */
            tmp = o;
            o = w;
            w = tmp;
            wsize = osize;
            break;


        case CRUSH_RULE_EMIT:
            for (i = 0; i < wsize && result_len < result_max; i++) {
                result[result_len] = w[i];
                result_len++;
            }
            wsize = 0;
            break;

        default:
            dprintk(" unknown op %d at step %d
",
                curstep->op, step);
            break;
        }
    }

    return result_len;
}