2 #include <linux/ceph/ceph_debug.h>
4 #include <linux/module.h>
5 #include <linux/slab.h>
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>
14 char *ceph_osdmap_state_str(char *str
, int len
, u32 state
)
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");
26 snprintf(str
, len
, "doesn't exist");
33 static int calc_bits_of(unsigned int t
)
44 * the foo_mask is the smallest value 2^n-1 that is >= foo.
46 static void calc_pg_masks(struct ceph_pg_pool_info
*pi
)
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;
55 static int crush_decode_uniform_bucket(void **p
, void *end
,
56 struct crush_bucket_uniform
*b
)
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
);
66 static int crush_decode_list_bucket(void **p
, void *end
,
67 struct crush_bucket_list
*b
)
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
)
74 b
->sum_weights
= kcalloc(b
->h
.size
, sizeof(u32
), GFP_NOFS
);
75 if (b
->sum_weights
== NULL
)
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
);
87 static int crush_decode_tree_bucket(void **p
, void *end
,
88 struct crush_bucket_tree
*b
)
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
)
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
);
104 static int crush_decode_straw_bucket(void **p
, void *end
,
105 struct crush_bucket_straw
*b
)
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
)
112 b
->straws
= kcalloc(b
->h
.size
, sizeof(u32
), GFP_NOFS
);
113 if (b
->straws
== NULL
)
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
);
125 static int crush_decode_straw2_bucket(void **p
, void *end
,
126 struct crush_bucket_straw2
*b
)
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
)
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
);
141 static struct crush_choose_arg_map
*alloc_choose_arg_map(void)
143 struct crush_choose_arg_map
*arg_map
;
145 arg_map
= kzalloc(sizeof(*arg_map
), GFP_NOIO
);
149 RB_CLEAR_NODE(&arg_map
->node
);
153 static void free_choose_arg_map(struct crush_choose_arg_map
*arg_map
)
158 WARN_ON(!RB_EMPTY_NODE(&arg_map
->node
));
160 for (i
= 0; i
< arg_map
->size
; i
++) {
161 struct crush_choose_arg
*arg
= &arg_map
->args
[i
];
163 for (j
= 0; j
< arg
->weight_set_size
; j
++)
164 kfree(arg
->weight_set
[j
].weights
);
165 kfree(arg
->weight_set
);
168 kfree(arg_map
->args
);
173 DEFINE_RB_FUNCS(choose_arg_map
, struct crush_choose_arg_map
, choose_args_index
,
176 void clear_choose_args(struct crush_map
*c
)
178 while (!RB_EMPTY_ROOT(&c
->choose_args
)) {
179 struct crush_choose_arg_map
*arg_map
=
180 rb_entry(rb_first(&c
->choose_args
),
181 struct crush_choose_arg_map
, node
);
183 erase_choose_arg_map(&c
->choose_args
, arg_map
);
184 free_choose_arg_map(arg_map
);
188 static u32
*decode_array_32_alloc(void **p
, void *end
, u32
*plen
)
194 ceph_decode_32_safe(p
, end
, len
, e_inval
);
198 a
= kmalloc_array(len
, sizeof(u32
), GFP_NOIO
);
204 ceph_decode_need(p
, end
, len
* sizeof(u32
), e_inval
);
205 for (i
= 0; i
< len
; i
++)
206 a
[i
] = ceph_decode_32(p
);
220 * Assumes @arg is zero-initialized.
222 static int decode_choose_arg(void **p
, void *end
, struct crush_choose_arg
*arg
)
226 ceph_decode_32_safe(p
, end
, arg
->weight_set_size
, e_inval
);
227 if (arg
->weight_set_size
) {
230 arg
->weight_set
= kmalloc_array(arg
->weight_set_size
,
231 sizeof(*arg
->weight_set
),
233 if (!arg
->weight_set
)
236 for (i
= 0; i
< arg
->weight_set_size
; i
++) {
237 struct crush_weight_set
*w
= &arg
->weight_set
[i
];
239 w
->weights
= decode_array_32_alloc(p
, end
, &w
->size
);
240 if (IS_ERR(w
->weights
)) {
241 ret
= PTR_ERR(w
->weights
);
248 arg
->ids
= decode_array_32_alloc(p
, end
, &arg
->ids_size
);
249 if (IS_ERR(arg
->ids
)) {
250 ret
= PTR_ERR(arg
->ids
);
261 static int decode_choose_args(void **p
, void *end
, struct crush_map
*c
)
263 struct crush_choose_arg_map
*arg_map
= NULL
;
264 u32 num_choose_arg_maps
, num_buckets
;
267 ceph_decode_32_safe(p
, end
, num_choose_arg_maps
, e_inval
);
268 while (num_choose_arg_maps
--) {
269 arg_map
= alloc_choose_arg_map();
275 ceph_decode_64_safe(p
, end
, arg_map
->choose_args_index
,
277 arg_map
->size
= c
->max_buckets
;
278 arg_map
->args
= kcalloc(arg_map
->size
, sizeof(*arg_map
->args
),
280 if (!arg_map
->args
) {
285 ceph_decode_32_safe(p
, end
, num_buckets
, e_inval
);
286 while (num_buckets
--) {
287 struct crush_choose_arg
*arg
;
290 ceph_decode_32_safe(p
, end
, bucket_index
, e_inval
);
291 if (bucket_index
>= arg_map
->size
)
294 arg
= &arg_map
->args
[bucket_index
];
295 ret
= decode_choose_arg(p
, end
, arg
);
300 arg
->ids_size
!= c
->buckets
[bucket_index
]->size
)
304 insert_choose_arg_map(&c
->choose_args
, arg_map
);
312 free_choose_arg_map(arg_map
);
316 static void crush_finalize(struct crush_map
*c
)
320 /* Space for the array of pointers to per-bucket workspace */
321 c
->working_size
= sizeof(struct crush_work
) +
322 c
->max_buckets
* sizeof(struct crush_work_bucket
*);
324 for (b
= 0; b
< c
->max_buckets
; b
++) {
328 switch (c
->buckets
[b
]->alg
) {
331 * The base case, permutation variables and
332 * the pointer to the permutation array.
334 c
->working_size
+= sizeof(struct crush_work_bucket
);
337 /* Every bucket has a permutation array. */
338 c
->working_size
+= c
->buckets
[b
]->size
* sizeof(__u32
);
342 static struct crush_map
*crush_decode(void *pbyval
, void *end
)
348 void *start
= pbyval
;
351 dout("crush_decode %p to %p len %d\n", *p
, end
, (int)(end
- *p
));
353 c
= kzalloc(sizeof(*c
), GFP_NOFS
);
355 return ERR_PTR(-ENOMEM
);
357 c
->choose_args
= RB_ROOT
;
359 /* set tunables to default values */
360 c
->choose_local_tries
= 2;
361 c
->choose_local_fallback_tries
= 5;
362 c
->choose_total_tries
= 19;
363 c
->chooseleaf_descend_once
= 0;
365 ceph_decode_need(p
, end
, 4*sizeof(u32
), bad
);
366 magic
= ceph_decode_32(p
);
367 if (magic
!= CRUSH_MAGIC
) {
368 pr_err("crush_decode magic %x != current %x\n",
369 (unsigned int)magic
, (unsigned int)CRUSH_MAGIC
);
372 c
->max_buckets
= ceph_decode_32(p
);
373 c
->max_rules
= ceph_decode_32(p
);
374 c
->max_devices
= ceph_decode_32(p
);
376 c
->buckets
= kcalloc(c
->max_buckets
, sizeof(*c
->buckets
), GFP_NOFS
);
377 if (c
->buckets
== NULL
)
379 c
->rules
= kcalloc(c
->max_rules
, sizeof(*c
->rules
), GFP_NOFS
);
380 if (c
->rules
== NULL
)
384 for (i
= 0; i
< c
->max_buckets
; i
++) {
387 struct crush_bucket
*b
;
389 ceph_decode_32_safe(p
, end
, alg
, bad
);
391 c
->buckets
[i
] = NULL
;
394 dout("crush_decode bucket %d off %x %p to %p\n",
395 i
, (int)(*p
-start
), *p
, end
);
398 case CRUSH_BUCKET_UNIFORM
:
399 size
= sizeof(struct crush_bucket_uniform
);
401 case CRUSH_BUCKET_LIST
:
402 size
= sizeof(struct crush_bucket_list
);
404 case CRUSH_BUCKET_TREE
:
405 size
= sizeof(struct crush_bucket_tree
);
407 case CRUSH_BUCKET_STRAW
:
408 size
= sizeof(struct crush_bucket_straw
);
410 case CRUSH_BUCKET_STRAW2
:
411 size
= sizeof(struct crush_bucket_straw2
);
417 b
= c
->buckets
[i
] = kzalloc(size
, GFP_NOFS
);
421 ceph_decode_need(p
, end
, 4*sizeof(u32
), bad
);
422 b
->id
= ceph_decode_32(p
);
423 b
->type
= ceph_decode_16(p
);
424 b
->alg
= ceph_decode_8(p
);
425 b
->hash
= ceph_decode_8(p
);
426 b
->weight
= ceph_decode_32(p
);
427 b
->size
= ceph_decode_32(p
);
429 dout("crush_decode bucket size %d off %x %p to %p\n",
430 b
->size
, (int)(*p
-start
), *p
, end
);
432 b
->items
= kcalloc(b
->size
, sizeof(__s32
), GFP_NOFS
);
433 if (b
->items
== NULL
)
436 ceph_decode_need(p
, end
, b
->size
*sizeof(u32
), bad
);
437 for (j
= 0; j
< b
->size
; j
++)
438 b
->items
[j
] = ceph_decode_32(p
);
441 case CRUSH_BUCKET_UNIFORM
:
442 err
= crush_decode_uniform_bucket(p
, end
,
443 (struct crush_bucket_uniform
*)b
);
447 case CRUSH_BUCKET_LIST
:
448 err
= crush_decode_list_bucket(p
, end
,
449 (struct crush_bucket_list
*)b
);
453 case CRUSH_BUCKET_TREE
:
454 err
= crush_decode_tree_bucket(p
, end
,
455 (struct crush_bucket_tree
*)b
);
459 case CRUSH_BUCKET_STRAW
:
460 err
= crush_decode_straw_bucket(p
, end
,
461 (struct crush_bucket_straw
*)b
);
465 case CRUSH_BUCKET_STRAW2
:
466 err
= crush_decode_straw2_bucket(p
, end
,
467 (struct crush_bucket_straw2
*)b
);
475 dout("rule vec is %p\n", c
->rules
);
476 for (i
= 0; i
< c
->max_rules
; i
++) {
478 struct crush_rule
*r
;
480 ceph_decode_32_safe(p
, end
, yes
, bad
);
482 dout("crush_decode NO rule %d off %x %p to %p\n",
483 i
, (int)(*p
-start
), *p
, end
);
488 dout("crush_decode rule %d off %x %p to %p\n",
489 i
, (int)(*p
-start
), *p
, end
);
492 ceph_decode_32_safe(p
, end
, yes
, bad
);
493 #if BITS_PER_LONG == 32
494 if (yes
> (ULONG_MAX
- sizeof(*r
))
495 / sizeof(struct crush_rule_step
))
498 r
= c
->rules
[i
] = kmalloc(sizeof(*r
) +
499 yes
*sizeof(struct crush_rule_step
),
503 dout(" rule %d is at %p\n", i
, r
);
505 ceph_decode_copy_safe(p
, end
, &r
->mask
, 4, bad
); /* 4 u8's */
506 ceph_decode_need(p
, end
, r
->len
*3*sizeof(u32
), bad
);
507 for (j
= 0; j
< r
->len
; j
++) {
508 r
->steps
[j
].op
= ceph_decode_32(p
);
509 r
->steps
[j
].arg1
= ceph_decode_32(p
);
510 r
->steps
[j
].arg2
= ceph_decode_32(p
);
514 ceph_decode_skip_map(p
, end
, 32, string
, bad
); /* type_map */
515 ceph_decode_skip_map(p
, end
, 32, string
, bad
); /* name_map */
516 ceph_decode_skip_map(p
, end
, 32, string
, bad
); /* rule_name_map */
519 ceph_decode_need(p
, end
, 3*sizeof(u32
), done
);
520 c
->choose_local_tries
= ceph_decode_32(p
);
521 c
->choose_local_fallback_tries
= ceph_decode_32(p
);
522 c
->choose_total_tries
= ceph_decode_32(p
);
523 dout("crush decode tunable choose_local_tries = %d\n",
524 c
->choose_local_tries
);
525 dout("crush decode tunable choose_local_fallback_tries = %d\n",
526 c
->choose_local_fallback_tries
);
527 dout("crush decode tunable choose_total_tries = %d\n",
528 c
->choose_total_tries
);
530 ceph_decode_need(p
, end
, sizeof(u32
), done
);
531 c
->chooseleaf_descend_once
= ceph_decode_32(p
);
532 dout("crush decode tunable chooseleaf_descend_once = %d\n",
533 c
->chooseleaf_descend_once
);
535 ceph_decode_need(p
, end
, sizeof(u8
), done
);
536 c
->chooseleaf_vary_r
= ceph_decode_8(p
);
537 dout("crush decode tunable chooseleaf_vary_r = %d\n",
538 c
->chooseleaf_vary_r
);
540 /* skip straw_calc_version, allowed_bucket_algs */
541 ceph_decode_need(p
, end
, sizeof(u8
) + sizeof(u32
), done
);
542 *p
+= sizeof(u8
) + sizeof(u32
);
544 ceph_decode_need(p
, end
, sizeof(u8
), done
);
545 c
->chooseleaf_stable
= ceph_decode_8(p
);
546 dout("crush decode tunable chooseleaf_stable = %d\n",
547 c
->chooseleaf_stable
);
551 ceph_decode_skip_map(p
, end
, 32, 32, bad
);
553 ceph_decode_skip_map(p
, end
, 32, string
, bad
);
555 ceph_decode_skip_map_of_map(p
, end
, 32, 32, 32, bad
);
559 err
= decode_choose_args(p
, end
, c
);
566 dout("crush_decode success\n");
572 dout("crush_decode fail %d\n", err
);
581 int ceph_pg_compare(const struct ceph_pg
*lhs
, const struct ceph_pg
*rhs
)
583 if (lhs
->pool
< rhs
->pool
)
585 if (lhs
->pool
> rhs
->pool
)
587 if (lhs
->seed
< rhs
->seed
)
589 if (lhs
->seed
> rhs
->seed
)
595 int ceph_spg_compare(const struct ceph_spg
*lhs
, const struct ceph_spg
*rhs
)
599 ret
= ceph_pg_compare(&lhs
->pgid
, &rhs
->pgid
);
603 if (lhs
->shard
< rhs
->shard
)
605 if (lhs
->shard
> rhs
->shard
)
611 static struct ceph_pg_mapping
*alloc_pg_mapping(size_t payload_len
)
613 struct ceph_pg_mapping
*pg
;
615 pg
= kmalloc(sizeof(*pg
) + payload_len
, GFP_NOIO
);
619 RB_CLEAR_NODE(&pg
->node
);
623 static void free_pg_mapping(struct ceph_pg_mapping
*pg
)
625 WARN_ON(!RB_EMPTY_NODE(&pg
->node
));
631 * rbtree of pg_mapping for handling pg_temp (explicit mapping of pgid
632 * to a set of osds) and primary_temp (explicit primary setting)
634 DEFINE_RB_FUNCS2(pg_mapping
, struct ceph_pg_mapping
, pgid
, ceph_pg_compare
,
635 RB_BYPTR
, const struct ceph_pg
*, node
)
638 * rbtree of pg pool info
640 static int __insert_pg_pool(struct rb_root
*root
, struct ceph_pg_pool_info
*new)
642 struct rb_node
**p
= &root
->rb_node
;
643 struct rb_node
*parent
= NULL
;
644 struct ceph_pg_pool_info
*pi
= NULL
;
648 pi
= rb_entry(parent
, struct ceph_pg_pool_info
, node
);
649 if (new->id
< pi
->id
)
651 else if (new->id
> pi
->id
)
657 rb_link_node(&new->node
, parent
, p
);
658 rb_insert_color(&new->node
, root
);
662 static struct ceph_pg_pool_info
*__lookup_pg_pool(struct rb_root
*root
, u64 id
)
664 struct ceph_pg_pool_info
*pi
;
665 struct rb_node
*n
= root
->rb_node
;
668 pi
= rb_entry(n
, struct ceph_pg_pool_info
, node
);
671 else if (id
> pi
->id
)
679 struct ceph_pg_pool_info
*ceph_pg_pool_by_id(struct ceph_osdmap
*map
, u64 id
)
681 return __lookup_pg_pool(&map
->pg_pools
, id
);
684 const char *ceph_pg_pool_name_by_id(struct ceph_osdmap
*map
, u64 id
)
686 struct ceph_pg_pool_info
*pi
;
688 if (id
== CEPH_NOPOOL
)
691 if (WARN_ON_ONCE(id
> (u64
) INT_MAX
))
694 pi
= __lookup_pg_pool(&map
->pg_pools
, (int) id
);
696 return pi
? pi
->name
: NULL
;
698 EXPORT_SYMBOL(ceph_pg_pool_name_by_id
);
700 int ceph_pg_poolid_by_name(struct ceph_osdmap
*map
, const char *name
)
704 for (rbp
= rb_first(&map
->pg_pools
); rbp
; rbp
= rb_next(rbp
)) {
705 struct ceph_pg_pool_info
*pi
=
706 rb_entry(rbp
, struct ceph_pg_pool_info
, node
);
707 if (pi
->name
&& strcmp(pi
->name
, name
) == 0)
712 EXPORT_SYMBOL(ceph_pg_poolid_by_name
);
714 static void __remove_pg_pool(struct rb_root
*root
, struct ceph_pg_pool_info
*pi
)
716 rb_erase(&pi
->node
, root
);
721 static int decode_pool(void **p
, void *end
, struct ceph_pg_pool_info
*pi
)
727 ceph_decode_need(p
, end
, 2 + 4, bad
);
728 ev
= ceph_decode_8(p
); /* encoding version */
729 cv
= ceph_decode_8(p
); /* compat version */
731 pr_warn("got v %d < 5 cv %d of ceph_pg_pool\n", ev
, cv
);
735 pr_warn("got v %d cv %d > 9 of ceph_pg_pool\n", ev
, cv
);
738 len
= ceph_decode_32(p
);
739 ceph_decode_need(p
, end
, len
, bad
);
742 pi
->type
= ceph_decode_8(p
);
743 pi
->size
= ceph_decode_8(p
);
744 pi
->crush_ruleset
= ceph_decode_8(p
);
745 pi
->object_hash
= ceph_decode_8(p
);
747 pi
->pg_num
= ceph_decode_32(p
);
748 pi
->pgp_num
= ceph_decode_32(p
);
750 *p
+= 4 + 4; /* skip lpg* */
751 *p
+= 4; /* skip last_change */
752 *p
+= 8 + 4; /* skip snap_seq, snap_epoch */
755 num
= ceph_decode_32(p
);
757 *p
+= 8; /* snapid key */
758 *p
+= 1 + 1; /* versions */
759 len
= ceph_decode_32(p
);
763 /* skip removed_snaps */
764 num
= ceph_decode_32(p
);
767 *p
+= 8; /* skip auid */
768 pi
->flags
= ceph_decode_64(p
);
769 *p
+= 4; /* skip crash_replay_interval */
772 pi
->min_size
= ceph_decode_8(p
);
774 pi
->min_size
= pi
->size
- pi
->size
/ 2;
777 *p
+= 8 + 8; /* skip quota_max_* */
781 num
= ceph_decode_32(p
);
784 *p
+= 8; /* skip tier_of */
785 *p
+= 1; /* skip cache_mode */
787 pi
->read_tier
= ceph_decode_64(p
);
788 pi
->write_tier
= ceph_decode_64(p
);
795 /* skip properties */
796 num
= ceph_decode_32(p
);
798 len
= ceph_decode_32(p
);
800 len
= ceph_decode_32(p
);
806 /* skip hit_set_params */
807 *p
+= 1 + 1; /* versions */
808 len
= ceph_decode_32(p
);
811 *p
+= 4; /* skip hit_set_period */
812 *p
+= 4; /* skip hit_set_count */
816 *p
+= 4; /* skip stripe_width */
819 *p
+= 8; /* skip target_max_bytes */
820 *p
+= 8; /* skip target_max_objects */
821 *p
+= 4; /* skip cache_target_dirty_ratio_micro */
822 *p
+= 4; /* skip cache_target_full_ratio_micro */
823 *p
+= 4; /* skip cache_min_flush_age */
824 *p
+= 4; /* skip cache_min_evict_age */
828 /* skip erasure_code_profile */
829 len
= ceph_decode_32(p
);
834 * last_force_op_resend_preluminous, will be overridden if the
835 * map was encoded with RESEND_ON_SPLIT
838 pi
->last_force_request_resend
= ceph_decode_32(p
);
840 pi
->last_force_request_resend
= 0;
843 *p
+= 4; /* skip min_read_recency_for_promote */
846 *p
+= 8; /* skip expected_num_objects */
849 *p
+= 4; /* skip cache_target_dirty_high_ratio_micro */
852 *p
+= 4; /* skip min_write_recency_for_promote */
855 *p
+= 1; /* skip use_gmt_hitset */
858 *p
+= 1; /* skip fast_read */
861 *p
+= 4; /* skip hit_set_grade_decay_rate */
862 *p
+= 4; /* skip hit_set_search_last_n */
867 *p
+= 1 + 1; /* versions */
868 len
= ceph_decode_32(p
);
873 pi
->last_force_request_resend
= ceph_decode_32(p
);
875 /* ignore the rest */
885 static int decode_pool_names(void **p
, void *end
, struct ceph_osdmap
*map
)
887 struct ceph_pg_pool_info
*pi
;
891 ceph_decode_32_safe(p
, end
, num
, bad
);
892 dout(" %d pool names\n", num
);
894 ceph_decode_64_safe(p
, end
, pool
, bad
);
895 ceph_decode_32_safe(p
, end
, len
, bad
);
896 dout(" pool %llu len %d\n", pool
, len
);
897 ceph_decode_need(p
, end
, len
, bad
);
898 pi
= __lookup_pg_pool(&map
->pg_pools
, pool
);
900 char *name
= kstrndup(*p
, len
, GFP_NOFS
);
906 dout(" name is %s\n", pi
->name
);
919 struct ceph_osdmap
*ceph_osdmap_alloc(void)
921 struct ceph_osdmap
*map
;
923 map
= kzalloc(sizeof(*map
), GFP_NOIO
);
927 map
->pg_pools
= RB_ROOT
;
929 map
->pg_temp
= RB_ROOT
;
930 map
->primary_temp
= RB_ROOT
;
931 map
->pg_upmap
= RB_ROOT
;
932 map
->pg_upmap_items
= RB_ROOT
;
933 mutex_init(&map
->crush_workspace_mutex
);
938 void ceph_osdmap_destroy(struct ceph_osdmap
*map
)
940 dout("osdmap_destroy %p\n", map
);
942 crush_destroy(map
->crush
);
943 while (!RB_EMPTY_ROOT(&map
->pg_temp
)) {
944 struct ceph_pg_mapping
*pg
=
945 rb_entry(rb_first(&map
->pg_temp
),
946 struct ceph_pg_mapping
, node
);
947 erase_pg_mapping(&map
->pg_temp
, pg
);
950 while (!RB_EMPTY_ROOT(&map
->primary_temp
)) {
951 struct ceph_pg_mapping
*pg
=
952 rb_entry(rb_first(&map
->primary_temp
),
953 struct ceph_pg_mapping
, node
);
954 erase_pg_mapping(&map
->primary_temp
, pg
);
957 while (!RB_EMPTY_ROOT(&map
->pg_upmap
)) {
958 struct ceph_pg_mapping
*pg
=
959 rb_entry(rb_first(&map
->pg_upmap
),
960 struct ceph_pg_mapping
, node
);
961 rb_erase(&pg
->node
, &map
->pg_upmap
);
964 while (!RB_EMPTY_ROOT(&map
->pg_upmap_items
)) {
965 struct ceph_pg_mapping
*pg
=
966 rb_entry(rb_first(&map
->pg_upmap_items
),
967 struct ceph_pg_mapping
, node
);
968 rb_erase(&pg
->node
, &map
->pg_upmap_items
);
971 while (!RB_EMPTY_ROOT(&map
->pg_pools
)) {
972 struct ceph_pg_pool_info
*pi
=
973 rb_entry(rb_first(&map
->pg_pools
),
974 struct ceph_pg_pool_info
, node
);
975 __remove_pg_pool(&map
->pg_pools
, pi
);
977 kfree(map
->osd_state
);
978 kfree(map
->osd_weight
);
979 kfree(map
->osd_addr
);
980 kfree(map
->osd_primary_affinity
);
981 kfree(map
->crush_workspace
);
986 * Adjust max_osd value, (re)allocate arrays.
988 * The new elements are properly initialized.
990 static int osdmap_set_max_osd(struct ceph_osdmap
*map
, int max
)
994 struct ceph_entity_addr
*addr
;
997 state
= krealloc(map
->osd_state
, max
*sizeof(*state
), GFP_NOFS
);
1000 map
->osd_state
= state
;
1002 weight
= krealloc(map
->osd_weight
, max
*sizeof(*weight
), GFP_NOFS
);
1005 map
->osd_weight
= weight
;
1007 addr
= krealloc(map
->osd_addr
, max
*sizeof(*addr
), GFP_NOFS
);
1010 map
->osd_addr
= addr
;
1012 for (i
= map
->max_osd
; i
< max
; i
++) {
1013 map
->osd_state
[i
] = 0;
1014 map
->osd_weight
[i
] = CEPH_OSD_OUT
;
1015 memset(map
->osd_addr
+ i
, 0, sizeof(*map
->osd_addr
));
1018 if (map
->osd_primary_affinity
) {
1021 affinity
= krealloc(map
->osd_primary_affinity
,
1022 max
*sizeof(*affinity
), GFP_NOFS
);
1025 map
->osd_primary_affinity
= affinity
;
1027 for (i
= map
->max_osd
; i
< max
; i
++)
1028 map
->osd_primary_affinity
[i
] =
1029 CEPH_OSD_DEFAULT_PRIMARY_AFFINITY
;
1037 static int osdmap_set_crush(struct ceph_osdmap
*map
, struct crush_map
*crush
)
1043 return PTR_ERR(crush
);
1045 work_size
= crush_work_size(crush
, CEPH_PG_MAX_SIZE
);
1046 dout("%s work_size %zu bytes\n", __func__
, work_size
);
1047 workspace
= kmalloc(work_size
, GFP_NOIO
);
1049 crush_destroy(crush
);
1052 crush_init_workspace(crush
, workspace
);
1055 crush_destroy(map
->crush
);
1056 kfree(map
->crush_workspace
);
1058 map
->crush_workspace
= workspace
;
1062 #define OSDMAP_WRAPPER_COMPAT_VER 7
1063 #define OSDMAP_CLIENT_DATA_COMPAT_VER 1
1066 * Return 0 or error. On success, *v is set to 0 for old (v6) osdmaps,
1067 * to struct_v of the client_data section for new (v7 and above)
1070 static int get_osdmap_client_data_v(void **p
, void *end
,
1071 const char *prefix
, u8
*v
)
1075 ceph_decode_8_safe(p
, end
, struct_v
, e_inval
);
1076 if (struct_v
>= 7) {
1079 ceph_decode_8_safe(p
, end
, struct_compat
, e_inval
);
1080 if (struct_compat
> OSDMAP_WRAPPER_COMPAT_VER
) {
1081 pr_warn("got v %d cv %d > %d of %s ceph_osdmap\n",
1082 struct_v
, struct_compat
,
1083 OSDMAP_WRAPPER_COMPAT_VER
, prefix
);
1086 *p
+= 4; /* ignore wrapper struct_len */
1088 ceph_decode_8_safe(p
, end
, struct_v
, e_inval
);
1089 ceph_decode_8_safe(p
, end
, struct_compat
, e_inval
);
1090 if (struct_compat
> OSDMAP_CLIENT_DATA_COMPAT_VER
) {
1091 pr_warn("got v %d cv %d > %d of %s ceph_osdmap client data\n",
1092 struct_v
, struct_compat
,
1093 OSDMAP_CLIENT_DATA_COMPAT_VER
, prefix
);
1096 *p
+= 4; /* ignore client data struct_len */
1101 ceph_decode_16_safe(p
, end
, version
, e_inval
);
1103 pr_warn("got v %d < 6 of %s ceph_osdmap\n",
1108 /* old osdmap enconding */
1119 static int __decode_pools(void **p
, void *end
, struct ceph_osdmap
*map
,
1124 ceph_decode_32_safe(p
, end
, n
, e_inval
);
1126 struct ceph_pg_pool_info
*pi
;
1130 ceph_decode_64_safe(p
, end
, pool
, e_inval
);
1132 pi
= __lookup_pg_pool(&map
->pg_pools
, pool
);
1133 if (!incremental
|| !pi
) {
1134 pi
= kzalloc(sizeof(*pi
), GFP_NOFS
);
1140 ret
= __insert_pg_pool(&map
->pg_pools
, pi
);
1147 ret
= decode_pool(p
, end
, pi
);
1158 static int decode_pools(void **p
, void *end
, struct ceph_osdmap
*map
)
1160 return __decode_pools(p
, end
, map
, false);
1163 static int decode_new_pools(void **p
, void *end
, struct ceph_osdmap
*map
)
1165 return __decode_pools(p
, end
, map
, true);
1168 typedef struct ceph_pg_mapping
*(*decode_mapping_fn_t
)(void **, void *, bool);
1170 static int decode_pg_mapping(void **p
, void *end
, struct rb_root
*mapping_root
,
1171 decode_mapping_fn_t fn
, bool incremental
)
1175 WARN_ON(!incremental
&& !fn
);
1177 ceph_decode_32_safe(p
, end
, n
, e_inval
);
1179 struct ceph_pg_mapping
*pg
;
1180 struct ceph_pg pgid
;
1183 ret
= ceph_decode_pgid(p
, end
, &pgid
);
1187 pg
= lookup_pg_mapping(mapping_root
, &pgid
);
1189 WARN_ON(!incremental
);
1190 erase_pg_mapping(mapping_root
, pg
);
1191 free_pg_mapping(pg
);
1195 pg
= fn(p
, end
, incremental
);
1200 pg
->pgid
= pgid
; /* struct */
1201 insert_pg_mapping(mapping_root
, pg
);
1212 static struct ceph_pg_mapping
*__decode_pg_temp(void **p
, void *end
,
1215 struct ceph_pg_mapping
*pg
;
1218 ceph_decode_32_safe(p
, end
, len
, e_inval
);
1219 if (len
== 0 && incremental
)
1220 return NULL
; /* new_pg_temp: [] to remove */
1221 if (len
> (SIZE_MAX
- sizeof(*pg
)) / sizeof(u32
))
1222 return ERR_PTR(-EINVAL
);
1224 ceph_decode_need(p
, end
, len
* sizeof(u32
), e_inval
);
1225 pg
= alloc_pg_mapping(len
* sizeof(u32
));
1227 return ERR_PTR(-ENOMEM
);
1229 pg
->pg_temp
.len
= len
;
1230 for (i
= 0; i
< len
; i
++)
1231 pg
->pg_temp
.osds
[i
] = ceph_decode_32(p
);
1236 return ERR_PTR(-EINVAL
);
1239 static int decode_pg_temp(void **p
, void *end
, struct ceph_osdmap
*map
)
1241 return decode_pg_mapping(p
, end
, &map
->pg_temp
, __decode_pg_temp
,
1245 static int decode_new_pg_temp(void **p
, void *end
, struct ceph_osdmap
*map
)
1247 return decode_pg_mapping(p
, end
, &map
->pg_temp
, __decode_pg_temp
,
1251 static struct ceph_pg_mapping
*__decode_primary_temp(void **p
, void *end
,
1254 struct ceph_pg_mapping
*pg
;
1257 ceph_decode_32_safe(p
, end
, osd
, e_inval
);
1258 if (osd
== (u32
)-1 && incremental
)
1259 return NULL
; /* new_primary_temp: -1 to remove */
1261 pg
= alloc_pg_mapping(0);
1263 return ERR_PTR(-ENOMEM
);
1265 pg
->primary_temp
.osd
= osd
;
1269 return ERR_PTR(-EINVAL
);
1272 static int decode_primary_temp(void **p
, void *end
, struct ceph_osdmap
*map
)
1274 return decode_pg_mapping(p
, end
, &map
->primary_temp
,
1275 __decode_primary_temp
, false);
1278 static int decode_new_primary_temp(void **p
, void *end
,
1279 struct ceph_osdmap
*map
)
1281 return decode_pg_mapping(p
, end
, &map
->primary_temp
,
1282 __decode_primary_temp
, true);
1285 u32
ceph_get_primary_affinity(struct ceph_osdmap
*map
, int osd
)
1287 BUG_ON(osd
>= map
->max_osd
);
1289 if (!map
->osd_primary_affinity
)
1290 return CEPH_OSD_DEFAULT_PRIMARY_AFFINITY
;
1292 return map
->osd_primary_affinity
[osd
];
1295 static int set_primary_affinity(struct ceph_osdmap
*map
, int osd
, u32 aff
)
1297 BUG_ON(osd
>= map
->max_osd
);
1299 if (!map
->osd_primary_affinity
) {
1302 map
->osd_primary_affinity
= kmalloc(map
->max_osd
*sizeof(u32
),
1304 if (!map
->osd_primary_affinity
)
1307 for (i
= 0; i
< map
->max_osd
; i
++)
1308 map
->osd_primary_affinity
[i
] =
1309 CEPH_OSD_DEFAULT_PRIMARY_AFFINITY
;
1312 map
->osd_primary_affinity
[osd
] = aff
;
1317 static int decode_primary_affinity(void **p
, void *end
,
1318 struct ceph_osdmap
*map
)
1322 ceph_decode_32_safe(p
, end
, len
, e_inval
);
1324 kfree(map
->osd_primary_affinity
);
1325 map
->osd_primary_affinity
= NULL
;
1328 if (len
!= map
->max_osd
)
1331 ceph_decode_need(p
, end
, map
->max_osd
*sizeof(u32
), e_inval
);
1333 for (i
= 0; i
< map
->max_osd
; i
++) {
1336 ret
= set_primary_affinity(map
, i
, ceph_decode_32(p
));
1347 static int decode_new_primary_affinity(void **p
, void *end
,
1348 struct ceph_osdmap
*map
)
1352 ceph_decode_32_safe(p
, end
, n
, e_inval
);
1357 ceph_decode_32_safe(p
, end
, osd
, e_inval
);
1358 ceph_decode_32_safe(p
, end
, aff
, e_inval
);
1360 ret
= set_primary_affinity(map
, osd
, aff
);
1364 pr_info("osd%d primary-affinity 0x%x\n", osd
, aff
);
1373 static struct ceph_pg_mapping
*__decode_pg_upmap(void **p
, void *end
,
1376 return __decode_pg_temp(p
, end
, false);
1379 static int decode_pg_upmap(void **p
, void *end
, struct ceph_osdmap
*map
)
1381 return decode_pg_mapping(p
, end
, &map
->pg_upmap
, __decode_pg_upmap
,
1385 static int decode_new_pg_upmap(void **p
, void *end
, struct ceph_osdmap
*map
)
1387 return decode_pg_mapping(p
, end
, &map
->pg_upmap
, __decode_pg_upmap
,
1391 static int decode_old_pg_upmap(void **p
, void *end
, struct ceph_osdmap
*map
)
1393 return decode_pg_mapping(p
, end
, &map
->pg_upmap
, NULL
, true);
1396 static struct ceph_pg_mapping
*__decode_pg_upmap_items(void **p
, void *end
,
1399 struct ceph_pg_mapping
*pg
;
1402 ceph_decode_32_safe(p
, end
, len
, e_inval
);
1403 if (len
> (SIZE_MAX
- sizeof(*pg
)) / (2 * sizeof(u32
)))
1404 return ERR_PTR(-EINVAL
);
1406 ceph_decode_need(p
, end
, 2 * len
* sizeof(u32
), e_inval
);
1407 pg
= alloc_pg_mapping(2 * len
* sizeof(u32
));
1409 return ERR_PTR(-ENOMEM
);
1411 pg
->pg_upmap_items
.len
= len
;
1412 for (i
= 0; i
< len
; i
++) {
1413 pg
->pg_upmap_items
.from_to
[i
][0] = ceph_decode_32(p
);
1414 pg
->pg_upmap_items
.from_to
[i
][1] = ceph_decode_32(p
);
1420 return ERR_PTR(-EINVAL
);
1423 static int decode_pg_upmap_items(void **p
, void *end
, struct ceph_osdmap
*map
)
1425 return decode_pg_mapping(p
, end
, &map
->pg_upmap_items
,
1426 __decode_pg_upmap_items
, false);
1429 static int decode_new_pg_upmap_items(void **p
, void *end
,
1430 struct ceph_osdmap
*map
)
1432 return decode_pg_mapping(p
, end
, &map
->pg_upmap_items
,
1433 __decode_pg_upmap_items
, true);
1436 static int decode_old_pg_upmap_items(void **p
, void *end
,
1437 struct ceph_osdmap
*map
)
1439 return decode_pg_mapping(p
, end
, &map
->pg_upmap_items
, NULL
, true);
1443 * decode a full map.
1445 static int osdmap_decode(void **p
, void *end
, struct ceph_osdmap
*map
)
1454 dout("%s %p to %p len %d\n", __func__
, *p
, end
, (int)(end
- *p
));
1456 err
= get_osdmap_client_data_v(p
, end
, "full", &struct_v
);
1460 /* fsid, epoch, created, modified */
1461 ceph_decode_need(p
, end
, sizeof(map
->fsid
) + sizeof(u32
) +
1462 sizeof(map
->created
) + sizeof(map
->modified
), e_inval
);
1463 ceph_decode_copy(p
, &map
->fsid
, sizeof(map
->fsid
));
1464 epoch
= map
->epoch
= ceph_decode_32(p
);
1465 ceph_decode_copy(p
, &map
->created
, sizeof(map
->created
));
1466 ceph_decode_copy(p
, &map
->modified
, sizeof(map
->modified
));
1469 err
= decode_pools(p
, end
, map
);
1474 err
= decode_pool_names(p
, end
, map
);
1478 ceph_decode_32_safe(p
, end
, map
->pool_max
, e_inval
);
1480 ceph_decode_32_safe(p
, end
, map
->flags
, e_inval
);
1483 ceph_decode_32_safe(p
, end
, max
, e_inval
);
1485 /* (re)alloc osd arrays */
1486 err
= osdmap_set_max_osd(map
, max
);
1490 /* osd_state, osd_weight, osd_addrs->client_addr */
1491 ceph_decode_need(p
, end
, 3*sizeof(u32
) +
1492 map
->max_osd
*((struct_v
>= 5 ? sizeof(u32
) :
1494 sizeof(*map
->osd_weight
) +
1495 sizeof(*map
->osd_addr
)), e_inval
);
1497 if (ceph_decode_32(p
) != map
->max_osd
)
1500 if (struct_v
>= 5) {
1501 for (i
= 0; i
< map
->max_osd
; i
++)
1502 map
->osd_state
[i
] = ceph_decode_32(p
);
1504 for (i
= 0; i
< map
->max_osd
; i
++)
1505 map
->osd_state
[i
] = ceph_decode_8(p
);
1508 if (ceph_decode_32(p
) != map
->max_osd
)
1511 for (i
= 0; i
< map
->max_osd
; i
++)
1512 map
->osd_weight
[i
] = ceph_decode_32(p
);
1514 if (ceph_decode_32(p
) != map
->max_osd
)
1517 ceph_decode_copy(p
, map
->osd_addr
, map
->max_osd
*sizeof(*map
->osd_addr
));
1518 for (i
= 0; i
< map
->max_osd
; i
++)
1519 ceph_decode_addr(&map
->osd_addr
[i
]);
1522 err
= decode_pg_temp(p
, end
, map
);
1527 if (struct_v
>= 1) {
1528 err
= decode_primary_temp(p
, end
, map
);
1533 /* primary_affinity */
1534 if (struct_v
>= 2) {
1535 err
= decode_primary_affinity(p
, end
, map
);
1539 WARN_ON(map
->osd_primary_affinity
);
1543 ceph_decode_32_safe(p
, end
, len
, e_inval
);
1544 err
= osdmap_set_crush(map
, crush_decode(*p
, min(*p
+ len
, end
)));
1549 if (struct_v
>= 3) {
1550 /* erasure_code_profiles */
1551 ceph_decode_skip_map_of_map(p
, end
, string
, string
, string
,
1555 if (struct_v
>= 4) {
1556 err
= decode_pg_upmap(p
, end
, map
);
1560 err
= decode_pg_upmap_items(p
, end
, map
);
1564 WARN_ON(!RB_EMPTY_ROOT(&map
->pg_upmap
));
1565 WARN_ON(!RB_EMPTY_ROOT(&map
->pg_upmap_items
));
1568 /* ignore the rest */
1571 dout("full osdmap epoch %d max_osd %d\n", map
->epoch
, map
->max_osd
);
1577 pr_err("corrupt full osdmap (%d) epoch %d off %d (%p of %p-%p)\n",
1578 err
, epoch
, (int)(*p
- start
), *p
, start
, end
);
1579 print_hex_dump(KERN_DEBUG
, "osdmap: ",
1580 DUMP_PREFIX_OFFSET
, 16, 1,
1581 start
, end
- start
, true);
1586 * Allocate and decode a full map.
1588 struct ceph_osdmap
*ceph_osdmap_decode(void **p
, void *end
)
1590 struct ceph_osdmap
*map
;
1593 map
= ceph_osdmap_alloc();
1595 return ERR_PTR(-ENOMEM
);
1597 ret
= osdmap_decode(p
, end
, map
);
1599 ceph_osdmap_destroy(map
);
1600 return ERR_PTR(ret
);
1607 * Encoding order is (new_up_client, new_state, new_weight). Need to
1608 * apply in the (new_weight, new_state, new_up_client) order, because
1609 * an incremental map may look like e.g.
1611 * new_up_client: { osd=6, addr=... } # set osd_state and addr
1612 * new_state: { osd=6, xorstate=EXISTS } # clear osd_state
1614 static int decode_new_up_state_weight(void **p
, void *end
, u8 struct_v
,
1615 struct ceph_osdmap
*map
)
1617 void *new_up_client
;
1619 void *new_weight_end
;
1623 ceph_decode_32_safe(p
, end
, len
, e_inval
);
1624 len
*= sizeof(u32
) + sizeof(struct ceph_entity_addr
);
1625 ceph_decode_need(p
, end
, len
, e_inval
);
1629 ceph_decode_32_safe(p
, end
, len
, e_inval
);
1630 len
*= sizeof(u32
) + (struct_v
>= 5 ? sizeof(u32
) : sizeof(u8
));
1631 ceph_decode_need(p
, end
, len
, e_inval
);
1635 ceph_decode_32_safe(p
, end
, len
, e_inval
);
1640 ceph_decode_need(p
, end
, 2*sizeof(u32
), e_inval
);
1641 osd
= ceph_decode_32(p
);
1642 w
= ceph_decode_32(p
);
1643 BUG_ON(osd
>= map
->max_osd
);
1644 pr_info("osd%d weight 0x%x %s\n", osd
, w
,
1645 w
== CEPH_OSD_IN
? "(in)" :
1646 (w
== CEPH_OSD_OUT
? "(out)" : ""));
1647 map
->osd_weight
[osd
] = w
;
1650 * If we are marking in, set the EXISTS, and clear the
1651 * AUTOOUT and NEW bits.
1654 map
->osd_state
[osd
] |= CEPH_OSD_EXISTS
;
1655 map
->osd_state
[osd
] &= ~(CEPH_OSD_AUTOOUT
|
1659 new_weight_end
= *p
;
1661 /* new_state (up/down) */
1663 len
= ceph_decode_32(p
);
1669 osd
= ceph_decode_32(p
);
1671 xorstate
= ceph_decode_32(p
);
1673 xorstate
= ceph_decode_8(p
);
1675 xorstate
= CEPH_OSD_UP
;
1676 BUG_ON(osd
>= map
->max_osd
);
1677 if ((map
->osd_state
[osd
] & CEPH_OSD_UP
) &&
1678 (xorstate
& CEPH_OSD_UP
))
1679 pr_info("osd%d down\n", osd
);
1680 if ((map
->osd_state
[osd
] & CEPH_OSD_EXISTS
) &&
1681 (xorstate
& CEPH_OSD_EXISTS
)) {
1682 pr_info("osd%d does not exist\n", osd
);
1683 ret
= set_primary_affinity(map
, osd
,
1684 CEPH_OSD_DEFAULT_PRIMARY_AFFINITY
);
1687 memset(map
->osd_addr
+ osd
, 0, sizeof(*map
->osd_addr
));
1688 map
->osd_state
[osd
] = 0;
1690 map
->osd_state
[osd
] ^= xorstate
;
1696 len
= ceph_decode_32(p
);
1699 struct ceph_entity_addr addr
;
1701 osd
= ceph_decode_32(p
);
1702 ceph_decode_copy(p
, &addr
, sizeof(addr
));
1703 ceph_decode_addr(&addr
);
1704 BUG_ON(osd
>= map
->max_osd
);
1705 pr_info("osd%d up\n", osd
);
1706 map
->osd_state
[osd
] |= CEPH_OSD_EXISTS
| CEPH_OSD_UP
;
1707 map
->osd_addr
[osd
] = addr
;
1710 *p
= new_weight_end
;
1718 * decode and apply an incremental map update.
1720 struct ceph_osdmap
*osdmap_apply_incremental(void **p
, void *end
,
1721 struct ceph_osdmap
*map
)
1723 struct ceph_fsid fsid
;
1725 struct ceph_timespec modified
;
1729 __s32 new_flags
, max
;
1734 dout("%s %p to %p len %d\n", __func__
, *p
, end
, (int)(end
- *p
));
1736 err
= get_osdmap_client_data_v(p
, end
, "inc", &struct_v
);
1740 /* fsid, epoch, modified, new_pool_max, new_flags */
1741 ceph_decode_need(p
, end
, sizeof(fsid
) + sizeof(u32
) + sizeof(modified
) +
1742 sizeof(u64
) + sizeof(u32
), e_inval
);
1743 ceph_decode_copy(p
, &fsid
, sizeof(fsid
));
1744 epoch
= ceph_decode_32(p
);
1745 BUG_ON(epoch
!= map
->epoch
+1);
1746 ceph_decode_copy(p
, &modified
, sizeof(modified
));
1747 new_pool_max
= ceph_decode_64(p
);
1748 new_flags
= ceph_decode_32(p
);
1751 ceph_decode_32_safe(p
, end
, len
, e_inval
);
1753 dout("apply_incremental full map len %d, %p to %p\n",
1755 return ceph_osdmap_decode(p
, min(*p
+len
, end
));
1759 ceph_decode_32_safe(p
, end
, len
, e_inval
);
1761 err
= osdmap_set_crush(map
,
1762 crush_decode(*p
, min(*p
+ len
, end
)));
1770 map
->flags
= new_flags
;
1771 if (new_pool_max
>= 0)
1772 map
->pool_max
= new_pool_max
;
1775 ceph_decode_32_safe(p
, end
, max
, e_inval
);
1777 err
= osdmap_set_max_osd(map
, max
);
1783 map
->modified
= modified
;
1786 err
= decode_new_pools(p
, end
, map
);
1790 /* new_pool_names */
1791 err
= decode_pool_names(p
, end
, map
);
1796 ceph_decode_32_safe(p
, end
, len
, e_inval
);
1798 struct ceph_pg_pool_info
*pi
;
1800 ceph_decode_64_safe(p
, end
, pool
, e_inval
);
1801 pi
= __lookup_pg_pool(&map
->pg_pools
, pool
);
1803 __remove_pg_pool(&map
->pg_pools
, pi
);
1806 /* new_up_client, new_state, new_weight */
1807 err
= decode_new_up_state_weight(p
, end
, struct_v
, map
);
1812 err
= decode_new_pg_temp(p
, end
, map
);
1816 /* new_primary_temp */
1817 if (struct_v
>= 1) {
1818 err
= decode_new_primary_temp(p
, end
, map
);
1823 /* new_primary_affinity */
1824 if (struct_v
>= 2) {
1825 err
= decode_new_primary_affinity(p
, end
, map
);
1830 if (struct_v
>= 3) {
1831 /* new_erasure_code_profiles */
1832 ceph_decode_skip_map_of_map(p
, end
, string
, string
, string
,
1834 /* old_erasure_code_profiles */
1835 ceph_decode_skip_set(p
, end
, string
, e_inval
);
1838 if (struct_v
>= 4) {
1839 err
= decode_new_pg_upmap(p
, end
, map
);
1843 err
= decode_old_pg_upmap(p
, end
, map
);
1847 err
= decode_new_pg_upmap_items(p
, end
, map
);
1851 err
= decode_old_pg_upmap_items(p
, end
, map
);
1856 /* ignore the rest */
1859 dout("inc osdmap epoch %d max_osd %d\n", map
->epoch
, map
->max_osd
);
1865 pr_err("corrupt inc osdmap (%d) epoch %d off %d (%p of %p-%p)\n",
1866 err
, epoch
, (int)(*p
- start
), *p
, start
, end
);
1867 print_hex_dump(KERN_DEBUG
, "osdmap: ",
1868 DUMP_PREFIX_OFFSET
, 16, 1,
1869 start
, end
- start
, true);
1870 return ERR_PTR(err
);
1873 void ceph_oloc_copy(struct ceph_object_locator
*dest
,
1874 const struct ceph_object_locator
*src
)
1876 ceph_oloc_destroy(dest
);
1878 dest
->pool
= src
->pool
;
1880 dest
->pool_ns
= ceph_get_string(src
->pool_ns
);
1882 dest
->pool_ns
= NULL
;
1884 EXPORT_SYMBOL(ceph_oloc_copy
);
1886 void ceph_oloc_destroy(struct ceph_object_locator
*oloc
)
1888 ceph_put_string(oloc
->pool_ns
);
1890 EXPORT_SYMBOL(ceph_oloc_destroy
);
1892 void ceph_oid_copy(struct ceph_object_id
*dest
,
1893 const struct ceph_object_id
*src
)
1895 ceph_oid_destroy(dest
);
1897 if (src
->name
!= src
->inline_name
) {
1898 /* very rare, see ceph_object_id definition */
1899 dest
->name
= kmalloc(src
->name_len
+ 1,
1900 GFP_NOIO
| __GFP_NOFAIL
);
1902 dest
->name
= dest
->inline_name
;
1904 memcpy(dest
->name
, src
->name
, src
->name_len
+ 1);
1905 dest
->name_len
= src
->name_len
;
1907 EXPORT_SYMBOL(ceph_oid_copy
);
1909 static __printf(2, 0)
1910 int oid_printf_vargs(struct ceph_object_id
*oid
, const char *fmt
, va_list ap
)
1914 WARN_ON(!ceph_oid_empty(oid
));
1916 len
= vsnprintf(oid
->inline_name
, sizeof(oid
->inline_name
), fmt
, ap
);
1917 if (len
>= sizeof(oid
->inline_name
))
1920 oid
->name_len
= len
;
1925 * If oid doesn't fit into inline buffer, BUG.
1927 void ceph_oid_printf(struct ceph_object_id
*oid
, const char *fmt
, ...)
1932 BUG_ON(oid_printf_vargs(oid
, fmt
, ap
));
1935 EXPORT_SYMBOL(ceph_oid_printf
);
1937 static __printf(3, 0)
1938 int oid_aprintf_vargs(struct ceph_object_id
*oid
, gfp_t gfp
,
1939 const char *fmt
, va_list ap
)
1945 len
= oid_printf_vargs(oid
, fmt
, aq
);
1949 char *external_name
;
1951 external_name
= kmalloc(len
+ 1, gfp
);
1955 oid
->name
= external_name
;
1956 WARN_ON(vsnprintf(oid
->name
, len
+ 1, fmt
, ap
) != len
);
1957 oid
->name_len
= len
;
1964 * If oid doesn't fit into inline buffer, allocate.
1966 int ceph_oid_aprintf(struct ceph_object_id
*oid
, gfp_t gfp
,
1967 const char *fmt
, ...)
1973 ret
= oid_aprintf_vargs(oid
, gfp
, fmt
, ap
);
1978 EXPORT_SYMBOL(ceph_oid_aprintf
);
1980 void ceph_oid_destroy(struct ceph_object_id
*oid
)
1982 if (oid
->name
!= oid
->inline_name
)
1985 EXPORT_SYMBOL(ceph_oid_destroy
);
1990 static bool __osds_equal(const struct ceph_osds
*lhs
,
1991 const struct ceph_osds
*rhs
)
1993 if (lhs
->size
== rhs
->size
&&
1994 !memcmp(lhs
->osds
, rhs
->osds
, rhs
->size
* sizeof(rhs
->osds
[0])))
2003 static bool osds_equal(const struct ceph_osds
*lhs
,
2004 const struct ceph_osds
*rhs
)
2006 if (__osds_equal(lhs
, rhs
) &&
2007 lhs
->primary
== rhs
->primary
)
2013 static bool osds_valid(const struct ceph_osds
*set
)
2016 if (set
->size
> 0 && set
->primary
>= 0)
2019 /* empty can_shift_osds set */
2020 if (!set
->size
&& set
->primary
== -1)
2023 /* empty !can_shift_osds set - all NONE */
2024 if (set
->size
> 0 && set
->primary
== -1) {
2027 for (i
= 0; i
< set
->size
; i
++) {
2028 if (set
->osds
[i
] != CRUSH_ITEM_NONE
)
2038 void ceph_osds_copy(struct ceph_osds
*dest
, const struct ceph_osds
*src
)
2040 memcpy(dest
->osds
, src
->osds
, src
->size
* sizeof(src
->osds
[0]));
2041 dest
->size
= src
->size
;
2042 dest
->primary
= src
->primary
;
2045 bool ceph_pg_is_split(const struct ceph_pg
*pgid
, u32 old_pg_num
,
2048 int old_bits
= calc_bits_of(old_pg_num
);
2049 int old_mask
= (1 << old_bits
) - 1;
2052 WARN_ON(pgid
->seed
>= old_pg_num
);
2053 if (new_pg_num
<= old_pg_num
)
2056 for (n
= 1; ; n
++) {
2057 int next_bit
= n
<< (old_bits
- 1);
2058 u32 s
= next_bit
| pgid
->seed
;
2060 if (s
< old_pg_num
|| s
== pgid
->seed
)
2062 if (s
>= new_pg_num
)
2065 s
= ceph_stable_mod(s
, old_pg_num
, old_mask
);
2066 if (s
== pgid
->seed
)
2073 bool ceph_is_new_interval(const struct ceph_osds
*old_acting
,
2074 const struct ceph_osds
*new_acting
,
2075 const struct ceph_osds
*old_up
,
2076 const struct ceph_osds
*new_up
,
2083 bool old_sort_bitwise
,
2084 bool new_sort_bitwise
,
2085 bool old_recovery_deletes
,
2086 bool new_recovery_deletes
,
2087 const struct ceph_pg
*pgid
)
2089 return !osds_equal(old_acting
, new_acting
) ||
2090 !osds_equal(old_up
, new_up
) ||
2091 old_size
!= new_size
||
2092 old_min_size
!= new_min_size
||
2093 ceph_pg_is_split(pgid
, old_pg_num
, new_pg_num
) ||
2094 old_sort_bitwise
!= new_sort_bitwise
||
2095 old_recovery_deletes
!= new_recovery_deletes
;
2098 static int calc_pg_rank(int osd
, const struct ceph_osds
*acting
)
2102 for (i
= 0; i
< acting
->size
; i
++) {
2103 if (acting
->osds
[i
] == osd
)
2110 static bool primary_changed(const struct ceph_osds
*old_acting
,
2111 const struct ceph_osds
*new_acting
)
2113 if (!old_acting
->size
&& !new_acting
->size
)
2114 return false; /* both still empty */
2116 if (!old_acting
->size
^ !new_acting
->size
)
2117 return true; /* was empty, now not, or vice versa */
2119 if (old_acting
->primary
!= new_acting
->primary
)
2120 return true; /* primary changed */
2122 if (calc_pg_rank(old_acting
->primary
, old_acting
) !=
2123 calc_pg_rank(new_acting
->primary
, new_acting
))
2126 return false; /* same primary (tho replicas may have changed) */
2129 bool ceph_osds_changed(const struct ceph_osds
*old_acting
,
2130 const struct ceph_osds
*new_acting
,
2133 if (primary_changed(old_acting
, new_acting
))
2136 if (any_change
&& !__osds_equal(old_acting
, new_acting
))
2143 * calculate file layout from given offset, length.
2144 * fill in correct oid, logical length, and object extent
2147 * for now, we write only a single su, until we can
2148 * pass a stride back to the caller.
2150 int ceph_calc_file_object_mapping(struct ceph_file_layout
*layout
,
2153 u64
*oxoff
, u64
*oxlen
)
2155 u32 osize
= layout
->object_size
;
2156 u32 su
= layout
->stripe_unit
;
2157 u32 sc
= layout
->stripe_count
;
2158 u32 bl
, stripeno
, stripepos
, objsetno
;
2162 dout("mapping %llu~%llu osize %u fl_su %u\n", off
, len
,
2164 if (su
== 0 || sc
== 0)
2166 su_per_object
= osize
/ su
;
2167 if (su_per_object
== 0)
2169 dout("osize %u / su %u = su_per_object %u\n", osize
, su
,
2172 if ((su
& ~PAGE_MASK
) != 0)
2175 /* bl = *off / su; */
2179 dout("off %llu / su %u = bl %u\n", off
, su
, bl
);
2182 stripepos
= bl
% sc
;
2183 objsetno
= stripeno
/ su_per_object
;
2185 *ono
= objsetno
* sc
+ stripepos
;
2186 dout("objset %u * sc %u = ono %u\n", objsetno
, sc
, (unsigned int)*ono
);
2188 /* *oxoff = *off % layout->fl_stripe_unit; # offset in su */
2190 su_offset
= do_div(t
, su
);
2191 *oxoff
= su_offset
+ (stripeno
% su_per_object
) * su
;
2194 * Calculate the length of the extent being written to the selected
2195 * object. This is the minimum of the full length requested (len) or
2196 * the remainder of the current stripe being written to.
2198 *oxlen
= min_t(u64
, len
, su
- su_offset
);
2200 dout(" obj extent %llu~%llu\n", *oxoff
, *oxlen
);
2204 dout(" invalid layout\n");
2210 EXPORT_SYMBOL(ceph_calc_file_object_mapping
);
2213 * Map an object into a PG.
2215 * Should only be called with target_oid and target_oloc (as opposed to
2216 * base_oid and base_oloc), since tiering isn't taken into account.
2218 int __ceph_object_locator_to_pg(struct ceph_pg_pool_info
*pi
,
2219 const struct ceph_object_id
*oid
,
2220 const struct ceph_object_locator
*oloc
,
2221 struct ceph_pg
*raw_pgid
)
2223 WARN_ON(pi
->id
!= oloc
->pool
);
2225 if (!oloc
->pool_ns
) {
2226 raw_pgid
->pool
= oloc
->pool
;
2227 raw_pgid
->seed
= ceph_str_hash(pi
->object_hash
, oid
->name
,
2229 dout("%s %s -> raw_pgid %llu.%x\n", __func__
, oid
->name
,
2230 raw_pgid
->pool
, raw_pgid
->seed
);
2232 char stack_buf
[256];
2233 char *buf
= stack_buf
;
2234 int nsl
= oloc
->pool_ns
->len
;
2235 size_t total
= nsl
+ 1 + oid
->name_len
;
2237 if (total
> sizeof(stack_buf
)) {
2238 buf
= kmalloc(total
, GFP_NOIO
);
2242 memcpy(buf
, oloc
->pool_ns
->str
, nsl
);
2244 memcpy(buf
+ nsl
+ 1, oid
->name
, oid
->name_len
);
2245 raw_pgid
->pool
= oloc
->pool
;
2246 raw_pgid
->seed
= ceph_str_hash(pi
->object_hash
, buf
, total
);
2247 if (buf
!= stack_buf
)
2249 dout("%s %s ns %.*s -> raw_pgid %llu.%x\n", __func__
,
2250 oid
->name
, nsl
, oloc
->pool_ns
->str
,
2251 raw_pgid
->pool
, raw_pgid
->seed
);
2256 int ceph_object_locator_to_pg(struct ceph_osdmap
*osdmap
,
2257 const struct ceph_object_id
*oid
,
2258 const struct ceph_object_locator
*oloc
,
2259 struct ceph_pg
*raw_pgid
)
2261 struct ceph_pg_pool_info
*pi
;
2263 pi
= ceph_pg_pool_by_id(osdmap
, oloc
->pool
);
2267 return __ceph_object_locator_to_pg(pi
, oid
, oloc
, raw_pgid
);
2269 EXPORT_SYMBOL(ceph_object_locator_to_pg
);
2272 * Map a raw PG (full precision ps) into an actual PG.
2274 static void raw_pg_to_pg(struct ceph_pg_pool_info
*pi
,
2275 const struct ceph_pg
*raw_pgid
,
2276 struct ceph_pg
*pgid
)
2278 pgid
->pool
= raw_pgid
->pool
;
2279 pgid
->seed
= ceph_stable_mod(raw_pgid
->seed
, pi
->pg_num
,
2284 * Map a raw PG (full precision ps) into a placement ps (placement
2285 * seed). Include pool id in that value so that different pools don't
2286 * use the same seeds.
2288 static u32
raw_pg_to_pps(struct ceph_pg_pool_info
*pi
,
2289 const struct ceph_pg
*raw_pgid
)
2291 if (pi
->flags
& CEPH_POOL_FLAG_HASHPSPOOL
) {
2292 /* hash pool id and seed so that pool PGs do not overlap */
2293 return crush_hash32_2(CRUSH_HASH_RJENKINS1
,
2294 ceph_stable_mod(raw_pgid
->seed
,
2300 * legacy behavior: add ps and pool together. this is
2301 * not a great approach because the PGs from each pool
2302 * will overlap on top of each other: 0.5 == 1.4 ==
2305 return ceph_stable_mod(raw_pgid
->seed
, pi
->pgp_num
,
2307 (unsigned)raw_pgid
->pool
;
2312 * Magic value used for a "default" fallback choose_args, used if the
2313 * crush_choose_arg_map passed to do_crush() does not exist. If this
2314 * also doesn't exist, fall back to canonical weights.
2316 #define CEPH_DEFAULT_CHOOSE_ARGS -1
2318 static int do_crush(struct ceph_osdmap
*map
, int ruleno
, int x
,
2319 int *result
, int result_max
,
2320 const __u32
*weight
, int weight_max
,
2321 s64 choose_args_index
)
2323 struct crush_choose_arg_map
*arg_map
;
2326 BUG_ON(result_max
> CEPH_PG_MAX_SIZE
);
2328 arg_map
= lookup_choose_arg_map(&map
->crush
->choose_args
,
2331 arg_map
= lookup_choose_arg_map(&map
->crush
->choose_args
,
2332 CEPH_DEFAULT_CHOOSE_ARGS
);
2334 mutex_lock(&map
->crush_workspace_mutex
);
2335 r
= crush_do_rule(map
->crush
, ruleno
, x
, result
, result_max
,
2336 weight
, weight_max
, map
->crush_workspace
,
2337 arg_map
? arg_map
->args
: NULL
);
2338 mutex_unlock(&map
->crush_workspace_mutex
);
2343 static void remove_nonexistent_osds(struct ceph_osdmap
*osdmap
,
2344 struct ceph_pg_pool_info
*pi
,
2345 struct ceph_osds
*set
)
2349 if (ceph_can_shift_osds(pi
)) {
2353 for (i
= 0; i
< set
->size
; i
++) {
2354 if (!ceph_osd_exists(osdmap
, set
->osds
[i
])) {
2359 set
->osds
[i
- removed
] = set
->osds
[i
];
2361 set
->size
-= removed
;
2363 /* set dne devices to NONE */
2364 for (i
= 0; i
< set
->size
; i
++) {
2365 if (!ceph_osd_exists(osdmap
, set
->osds
[i
]))
2366 set
->osds
[i
] = CRUSH_ITEM_NONE
;
2372 * Calculate raw set (CRUSH output) for given PG and filter out
2373 * nonexistent OSDs. ->primary is undefined for a raw set.
2375 * Placement seed (CRUSH input) is returned through @ppps.
2377 static void pg_to_raw_osds(struct ceph_osdmap
*osdmap
,
2378 struct ceph_pg_pool_info
*pi
,
2379 const struct ceph_pg
*raw_pgid
,
2380 struct ceph_osds
*raw
,
2383 u32 pps
= raw_pg_to_pps(pi
, raw_pgid
);
2387 ceph_osds_init(raw
);
2391 ruleno
= crush_find_rule(osdmap
->crush
, pi
->crush_ruleset
, pi
->type
,
2394 pr_err("no crush rule: pool %lld ruleset %d type %d size %d\n",
2395 pi
->id
, pi
->crush_ruleset
, pi
->type
, pi
->size
);
2399 if (pi
->size
> ARRAY_SIZE(raw
->osds
)) {
2400 pr_err_ratelimited("pool %lld ruleset %d type %d too wide: size %d > %zu\n",
2401 pi
->id
, pi
->crush_ruleset
, pi
->type
, pi
->size
,
2402 ARRAY_SIZE(raw
->osds
));
2406 len
= do_crush(osdmap
, ruleno
, pps
, raw
->osds
, pi
->size
,
2407 osdmap
->osd_weight
, osdmap
->max_osd
, pi
->id
);
2409 pr_err("error %d from crush rule %d: pool %lld ruleset %d type %d size %d\n",
2410 len
, ruleno
, pi
->id
, pi
->crush_ruleset
, pi
->type
,
2416 remove_nonexistent_osds(osdmap
, pi
, raw
);
2419 /* apply pg_upmap[_items] mappings */
2420 static void apply_upmap(struct ceph_osdmap
*osdmap
,
2421 const struct ceph_pg
*pgid
,
2422 struct ceph_osds
*raw
)
2424 struct ceph_pg_mapping
*pg
;
2427 pg
= lookup_pg_mapping(&osdmap
->pg_upmap
, pgid
);
2429 /* make sure targets aren't marked out */
2430 for (i
= 0; i
< pg
->pg_upmap
.len
; i
++) {
2431 int osd
= pg
->pg_upmap
.osds
[i
];
2433 if (osd
!= CRUSH_ITEM_NONE
&&
2434 osd
< osdmap
->max_osd
&&
2435 osdmap
->osd_weight
[osd
] == 0) {
2436 /* reject/ignore explicit mapping */
2440 for (i
= 0; i
< pg
->pg_upmap
.len
; i
++)
2441 raw
->osds
[i
] = pg
->pg_upmap
.osds
[i
];
2442 raw
->size
= pg
->pg_upmap
.len
;
2443 /* check and apply pg_upmap_items, if any */
2446 pg
= lookup_pg_mapping(&osdmap
->pg_upmap_items
, pgid
);
2448 for (i
= 0; i
< raw
->size
; i
++) {
2449 for (j
= 0; j
< pg
->pg_upmap_items
.len
; j
++) {
2450 int from
= pg
->pg_upmap_items
.from_to
[j
][0];
2451 int to
= pg
->pg_upmap_items
.from_to
[j
][1];
2453 if (from
== raw
->osds
[i
]) {
2454 if (!(to
!= CRUSH_ITEM_NONE
&&
2455 to
< osdmap
->max_osd
&&
2456 osdmap
->osd_weight
[to
] == 0))
2466 * Given raw set, calculate up set and up primary. By definition of an
2467 * up set, the result won't contain nonexistent or down OSDs.
2469 * This is done in-place - on return @set is the up set. If it's
2470 * empty, ->primary will remain undefined.
2472 static void raw_to_up_osds(struct ceph_osdmap
*osdmap
,
2473 struct ceph_pg_pool_info
*pi
,
2474 struct ceph_osds
*set
)
2478 /* ->primary is undefined for a raw set */
2479 BUG_ON(set
->primary
!= -1);
2481 if (ceph_can_shift_osds(pi
)) {
2485 for (i
= 0; i
< set
->size
; i
++) {
2486 if (ceph_osd_is_down(osdmap
, set
->osds
[i
])) {
2491 set
->osds
[i
- removed
] = set
->osds
[i
];
2493 set
->size
-= removed
;
2495 set
->primary
= set
->osds
[0];
2497 /* set down/dne devices to NONE */
2498 for (i
= set
->size
- 1; i
>= 0; i
--) {
2499 if (ceph_osd_is_down(osdmap
, set
->osds
[i
]))
2500 set
->osds
[i
] = CRUSH_ITEM_NONE
;
2502 set
->primary
= set
->osds
[i
];
2507 static void apply_primary_affinity(struct ceph_osdmap
*osdmap
,
2508 struct ceph_pg_pool_info
*pi
,
2510 struct ceph_osds
*up
)
2516 * Do we have any non-default primary_affinity values for these
2519 if (!osdmap
->osd_primary_affinity
)
2522 for (i
= 0; i
< up
->size
; i
++) {
2523 int osd
= up
->osds
[i
];
2525 if (osd
!= CRUSH_ITEM_NONE
&&
2526 osdmap
->osd_primary_affinity
[osd
] !=
2527 CEPH_OSD_DEFAULT_PRIMARY_AFFINITY
) {
2535 * Pick the primary. Feed both the seed (for the pg) and the
2536 * osd into the hash/rng so that a proportional fraction of an
2537 * osd's pgs get rejected as primary.
2539 for (i
= 0; i
< up
->size
; i
++) {
2540 int osd
= up
->osds
[i
];
2543 if (osd
== CRUSH_ITEM_NONE
)
2546 aff
= osdmap
->osd_primary_affinity
[osd
];
2547 if (aff
< CEPH_OSD_MAX_PRIMARY_AFFINITY
&&
2548 (crush_hash32_2(CRUSH_HASH_RJENKINS1
,
2549 pps
, osd
) >> 16) >= aff
) {
2551 * We chose not to use this primary. Note it
2552 * anyway as a fallback in case we don't pick
2553 * anyone else, but keep looking.
2565 up
->primary
= up
->osds
[pos
];
2567 if (ceph_can_shift_osds(pi
) && pos
> 0) {
2568 /* move the new primary to the front */
2569 for (i
= pos
; i
> 0; i
--)
2570 up
->osds
[i
] = up
->osds
[i
- 1];
2571 up
->osds
[0] = up
->primary
;
2576 * Get pg_temp and primary_temp mappings for given PG.
2578 * Note that a PG may have none, only pg_temp, only primary_temp or
2579 * both pg_temp and primary_temp mappings. This means @temp isn't
2580 * always a valid OSD set on return: in the "only primary_temp" case,
2581 * @temp will have its ->primary >= 0 but ->size == 0.
2583 static void get_temp_osds(struct ceph_osdmap
*osdmap
,
2584 struct ceph_pg_pool_info
*pi
,
2585 const struct ceph_pg
*pgid
,
2586 struct ceph_osds
*temp
)
2588 struct ceph_pg_mapping
*pg
;
2591 ceph_osds_init(temp
);
2594 pg
= lookup_pg_mapping(&osdmap
->pg_temp
, pgid
);
2596 for (i
= 0; i
< pg
->pg_temp
.len
; i
++) {
2597 if (ceph_osd_is_down(osdmap
, pg
->pg_temp
.osds
[i
])) {
2598 if (ceph_can_shift_osds(pi
))
2601 temp
->osds
[temp
->size
++] = CRUSH_ITEM_NONE
;
2603 temp
->osds
[temp
->size
++] = pg
->pg_temp
.osds
[i
];
2607 /* apply pg_temp's primary */
2608 for (i
= 0; i
< temp
->size
; i
++) {
2609 if (temp
->osds
[i
] != CRUSH_ITEM_NONE
) {
2610 temp
->primary
= temp
->osds
[i
];
2617 pg
= lookup_pg_mapping(&osdmap
->primary_temp
, pgid
);
2619 temp
->primary
= pg
->primary_temp
.osd
;
2623 * Map a PG to its acting set as well as its up set.
2625 * Acting set is used for data mapping purposes, while up set can be
2626 * recorded for detecting interval changes and deciding whether to
2629 void ceph_pg_to_up_acting_osds(struct ceph_osdmap
*osdmap
,
2630 struct ceph_pg_pool_info
*pi
,
2631 const struct ceph_pg
*raw_pgid
,
2632 struct ceph_osds
*up
,
2633 struct ceph_osds
*acting
)
2635 struct ceph_pg pgid
;
2638 WARN_ON(pi
->id
!= raw_pgid
->pool
);
2639 raw_pg_to_pg(pi
, raw_pgid
, &pgid
);
2641 pg_to_raw_osds(osdmap
, pi
, raw_pgid
, up
, &pps
);
2642 apply_upmap(osdmap
, &pgid
, up
);
2643 raw_to_up_osds(osdmap
, pi
, up
);
2644 apply_primary_affinity(osdmap
, pi
, pps
, up
);
2645 get_temp_osds(osdmap
, pi
, &pgid
, acting
);
2646 if (!acting
->size
) {
2647 memcpy(acting
->osds
, up
->osds
, up
->size
* sizeof(up
->osds
[0]));
2648 acting
->size
= up
->size
;
2649 if (acting
->primary
== -1)
2650 acting
->primary
= up
->primary
;
2652 WARN_ON(!osds_valid(up
) || !osds_valid(acting
));
2655 bool ceph_pg_to_primary_shard(struct ceph_osdmap
*osdmap
,
2656 struct ceph_pg_pool_info
*pi
,
2657 const struct ceph_pg
*raw_pgid
,
2658 struct ceph_spg
*spgid
)
2660 struct ceph_pg pgid
;
2661 struct ceph_osds up
, acting
;
2664 WARN_ON(pi
->id
!= raw_pgid
->pool
);
2665 raw_pg_to_pg(pi
, raw_pgid
, &pgid
);
2667 if (ceph_can_shift_osds(pi
)) {
2668 spgid
->pgid
= pgid
; /* struct */
2669 spgid
->shard
= CEPH_SPG_NOSHARD
;
2673 ceph_pg_to_up_acting_osds(osdmap
, pi
, &pgid
, &up
, &acting
);
2674 for (i
= 0; i
< acting
.size
; i
++) {
2675 if (acting
.osds
[i
] == acting
.primary
) {
2676 spgid
->pgid
= pgid
; /* struct */
2686 * Return acting primary for given PG, or -1 if none.
2688 int ceph_pg_to_acting_primary(struct ceph_osdmap
*osdmap
,
2689 const struct ceph_pg
*raw_pgid
)
2691 struct ceph_pg_pool_info
*pi
;
2692 struct ceph_osds up
, acting
;
2694 pi
= ceph_pg_pool_by_id(osdmap
, raw_pgid
->pool
);
2698 ceph_pg_to_up_acting_osds(osdmap
, pi
, raw_pgid
, &up
, &acting
);
2699 return acting
.primary
;
2701 EXPORT_SYMBOL(ceph_pg_to_acting_primary
);