X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=rcuja%2Frcuja.c;h=9398a222bcd526fc4d3dcd1f86e851a02a17ced4;hb=refs%2Fheads%2Furcu%2Frcuja-range;hp=e6c6d5b51493e8b5bbc58d3044c04ae2a84bb183;hpb=3c52f0f98170ba2fdc22c879fd99053292aade44;p=userspace-rcu.git diff --git a/rcuja/rcuja.c b/rcuja/rcuja.c index e6c6d5b..9398a22 100644 --- a/rcuja/rcuja.c +++ b/rcuja/rcuja.c @@ -3,7 +3,8 @@ * * Userspace RCU library - RCU Judy Array * - * Copyright 2012 - Mathieu Desnoyers + * Copyright (C) 2000 - 2002 Hewlett-Packard Company + * Copyright 2012-2013 - Mathieu Desnoyers * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public @@ -25,13 +26,12 @@ #include #include #include +#include #include #include #include -#include #include #include -#include #include "rcuja-internal.h" @@ -120,7 +120,9 @@ const struct cds_ja_type ja_types[] = { { .type_class = RCU_JA_LINEAR, .min_child = 10, .max_child = ja_type_4_max_child, .max_linear_child = ja_type_4_max_linear_child, .order = 7, }, /* Pools may fill sooner than max_child */ + /* This pool is hardcoded at index 5. See ja_node_ptr(). */ { .type_class = RCU_JA_POOL, .min_child = 20, .max_child = ja_type_5_max_child, .max_linear_child = ja_type_5_max_linear_child, .order = 8, .nr_pool_order = ja_type_5_nr_pool_order, .pool_size_order = 7, }, + /* This pool is hardcoded at index 6. See ja_node_ptr(). */ { .type_class = RCU_JA_POOL, .min_child = 45, .max_child = ja_type_6_max_child, .max_linear_child = ja_type_6_max_linear_child, .order = 9, .nr_pool_order = ja_type_6_nr_pool_order, .pool_size_order = 7, }, /* @@ -168,7 +170,9 @@ const struct cds_ja_type ja_types[] = { { .type_class = RCU_JA_LINEAR, .min_child = 10, .max_child = ja_type_4_max_child, .max_linear_child = ja_type_4_max_linear_child, .order = 8, }, /* Pools may fill sooner than max_child. */ + /* This pool is hardcoded at index 5. See ja_node_ptr(). */ { .type_class = RCU_JA_POOL, .min_child = 22, .max_child = ja_type_5_max_child, .max_linear_child = ja_type_5_max_linear_child, .order = 9, .nr_pool_order = ja_type_5_nr_pool_order, .pool_size_order = 8, }, + /* This pool is hardcoded at index 6. See ja_node_ptr(). */ { .type_class = RCU_JA_POOL, .min_child = 51, .max_child = ja_type_6_max_child, .max_linear_child = ja_type_6_max_linear_child, .order = 10, .nr_pool_order = ja_type_6_nr_pool_order, .pool_size_order = 8, }, /* @@ -243,6 +247,18 @@ enum ja_recompact { JA_RECOMPACT_DEL, }; +enum ja_lookup_inequality { + JA_LOOKUP_BE, + JA_LOOKUP_AE, +}; + +enum ja_direction { + JA_LEFT, + JA_RIGHT, + JA_LEFTMOST, + JA_RIGHTMOST, +}; + static struct cds_ja_inode *_ja_node_mask_ptr(struct cds_ja_inode_flag *node) { @@ -261,30 +277,6 @@ unsigned long ja_node_type(struct cds_ja_inode_flag *node) return type; } -struct cds_ja_inode *ja_node_ptr(struct cds_ja_inode_flag *node) -{ - unsigned long type_index = ja_node_type(node); - const struct cds_ja_type *type; - - type = &ja_types[type_index]; - switch (type->type_class) { - case RCU_JA_LINEAR: - case RCU_JA_PIGEON: /* fall-through */ - case RCU_JA_NULL: /* fall-through */ - default: /* fall-through */ - return _ja_node_mask_ptr(node); - case RCU_JA_POOL: - switch (type->nr_pool_order) { - case 1: - return (struct cds_ja_inode *) (((unsigned long) node) & ~(JA_POOL_1D_MASK | JA_TYPE_MASK)); - case 2: - return (struct cds_ja_inode *) (((unsigned long) node) & ~(JA_POOL_2D_MASK | JA_POOL_1D_MASK | JA_TYPE_MASK)); - default: - assert(0); - } - } -} - static struct cds_ja_inode *alloc_cds_ja_node(struct cds_ja *ja, const struct cds_ja_type *ja_type) @@ -298,14 +290,15 @@ struct cds_ja_inode *alloc_cds_ja_node(struct cds_ja *ja, return NULL; } memset(p, 0, len); - uatomic_inc(&ja->nr_nodes_allocated); + if (ja_debug_counters()) + uatomic_inc(&ja->nr_nodes_allocated); return p; } void free_cds_ja_node(struct cds_ja *ja, struct cds_ja_inode *node) { free(node); - if (node) + if (ja_debug_counters() && node) uatomic_inc(&ja->nr_nodes_freed); } @@ -370,18 +363,26 @@ struct cds_ja_inode_flag *ja_linear_node_get_nth(const struct cds_ja_type *type, } static -struct cds_ja_inode_flag *ja_linear_node_get_left(const struct cds_ja_type *type, +struct cds_ja_inode_flag *ja_linear_node_get_direction(const struct cds_ja_type *type, struct cds_ja_inode *node, - unsigned int n) + int n, uint8_t *result_key, + enum ja_direction dir) { uint8_t nr_child; uint8_t *values; struct cds_ja_inode_flag **pointers; - struct cds_ja_inode_flag *ptr; - unsigned int i, match_idx; - int match_v = -1; + struct cds_ja_inode_flag *ptr, *match_ptr = NULL; + unsigned int i; + int match_v; assert(type->type_class == RCU_JA_LINEAR || type->type_class == RCU_JA_POOL); + assert(dir == JA_LEFT || dir == JA_RIGHT); + + if (dir == JA_LEFT) { + match_v = -1; + } else { + match_v = JA_ENTRY_PER_NODE; + } nr_child = ja_linear_node_get_nr_child(type, node); cmm_smp_rmb(); /* read nr_child before values and pointers */ @@ -389,21 +390,34 @@ struct cds_ja_inode_flag *ja_linear_node_get_left(const struct cds_ja_type *type assert(type->type_class != RCU_JA_LINEAR || nr_child >= type->min_child); values = &node->u.data[1]; + pointers = (struct cds_ja_inode_flag **) align_ptr_size(&values[type->max_linear_child]); for (i = 0; i < nr_child; i++) { unsigned int v; v = CMM_LOAD_SHARED(values[i]); - if (v < n && (int) v > match_v) { - match_v = v; - match_idx = i; + ptr = CMM_LOAD_SHARED(pointers[i]); + if (!ptr) + continue; + if (dir == JA_LEFT) { + if ((int) v < n && (int) v > match_v) { + match_v = v; + match_ptr = ptr; + } + } else { + if ((int) v > n && (int) v < match_v) { + match_v = v; + match_ptr = ptr; + } } } - if (match_v < 0) { + + if (!match_ptr) { return NULL; } - pointers = (struct cds_ja_inode_flag **) align_ptr_size(&values[type->max_linear_child]); - ptr = rcu_dereference(pointers[match_idx]); - return ptr; + assert(match_v >= 0 && match_v < JA_ENTRY_PER_NODE); + + *result_key = (uint8_t) match_v; + return rcu_dereference(match_ptr); } static @@ -479,15 +493,23 @@ struct cds_ja_inode *ja_pool_node_get_ith_pool(const struct cds_ja_type *type, } static -struct cds_ja_inode_flag *ja_pool_node_get_left(const struct cds_ja_type *type, +struct cds_ja_inode_flag *ja_pool_node_get_direction(const struct cds_ja_type *type, struct cds_ja_inode *node, - unsigned int n) + int n, uint8_t *result_key, + enum ja_direction dir) { unsigned int pool_nr; - int match_v = -1; + int match_v; struct cds_ja_inode_flag *match_node_flag = NULL; assert(type->type_class == RCU_JA_POOL); + assert(dir == JA_LEFT || dir == JA_RIGHT); + + if (dir == JA_LEFT) { + match_v = -1; + } else { + match_v = JA_ENTRY_PER_NODE; + } for (pool_nr = 0; pool_nr < (1U << type->nr_pool_order); pool_nr++) { struct cds_ja_inode *pool = @@ -505,12 +527,21 @@ struct cds_ja_inode_flag *ja_pool_node_get_left(const struct cds_ja_type *type, j, &v, &iter); if (!iter) continue; - if (v < n && (int) v > match_v) { - match_v = v; - match_node_flag = iter; + if (dir == JA_LEFT) { + if ((int) v < n && (int) v > match_v) { + match_v = v; + match_node_flag = iter; + } + } else { + if ((int) v > n && (int) v < match_v) { + match_v = v; + match_node_flag = iter; + } } } } + if (match_node_flag) + *result_key = (uint8_t) match_v; return match_node_flag; } @@ -534,24 +565,41 @@ struct cds_ja_inode_flag *ja_pigeon_node_get_nth(const struct cds_ja_type *type, } static -struct cds_ja_inode_flag *ja_pigeon_node_get_left(const struct cds_ja_type *type, +struct cds_ja_inode_flag *ja_pigeon_node_get_direction(const struct cds_ja_type *type, struct cds_ja_inode *node, - unsigned int n) + int n, uint8_t *result_key, + enum ja_direction dir) { struct cds_ja_inode_flag **child_node_flag_ptr; struct cds_ja_inode_flag *child_node_flag; int i; assert(type->type_class == RCU_JA_PIGEON); - - /* n - 1 is first value left of n */ - for (i = n - 1; i >= 0; i--) { - child_node_flag_ptr = &((struct cds_ja_inode_flag **) node->u.data)[i]; - child_node_flag = rcu_dereference(*child_node_flag_ptr); - if (child_node_flag) { - dbg_printf("ja_pigeon_node_get_left child_node_flag %p\n", - child_node_flag); - return child_node_flag; + assert(dir == JA_LEFT || dir == JA_RIGHT); + + if (dir == JA_LEFT) { + /* n - 1 is first value left of n */ + for (i = n - 1; i >= 0; i--) { + child_node_flag_ptr = &((struct cds_ja_inode_flag **) node->u.data)[i]; + child_node_flag = rcu_dereference(*child_node_flag_ptr); + if (child_node_flag) { + dbg_printf("ja_pigeon_node_get_left child_node_flag %p\n", + child_node_flag); + *result_key = (uint8_t) i; + return child_node_flag; + } + } + } else { + /* n + 1 is first value right of n */ + for (i = n + 1; i < JA_ENTRY_PER_NODE; i++) { + child_node_flag_ptr = &((struct cds_ja_inode_flag **) node->u.data)[i]; + child_node_flag = rcu_dereference(*child_node_flag_ptr); + if (child_node_flag) { + dbg_printf("ja_pigeon_node_get_right child_node_flag %p\n", + child_node_flag); + *result_key = (uint8_t) i; + return child_node_flag; + } } } return NULL; @@ -600,8 +648,9 @@ struct cds_ja_inode_flag *ja_node_get_nth(struct cds_ja_inode_flag *node_flag, } static -struct cds_ja_inode_flag *ja_node_get_left(struct cds_ja_inode_flag *node_flag, - unsigned int n) +struct cds_ja_inode_flag *ja_node_get_direction(struct cds_ja_inode_flag *node_flag, + int n, uint8_t *result_key, + enum ja_direction dir) { unsigned int type_index; struct cds_ja_inode *node; @@ -614,11 +663,11 @@ struct cds_ja_inode_flag *ja_node_get_left(struct cds_ja_inode_flag *node_flag, switch (type->type_class) { case RCU_JA_LINEAR: - return ja_linear_node_get_left(type, node, n); + return ja_linear_node_get_direction(type, node, n, result_key, dir); case RCU_JA_POOL: - return ja_pool_node_get_left(type, node, n); + return ja_pool_node_get_direction(type, node, n, result_key, dir); case RCU_JA_PIGEON: - return ja_pigeon_node_get_left(type, node, n); + return ja_pigeon_node_get_direction(type, node, n, result_key, dir); default: assert(0); return (void *) -1UL; @@ -626,9 +675,28 @@ struct cds_ja_inode_flag *ja_node_get_left(struct cds_ja_inode_flag *node_flag, } static -struct cds_ja_inode_flag *ja_node_get_rightmost(struct cds_ja_inode_flag *node_flag) +struct cds_ja_inode_flag *ja_node_get_leftright(struct cds_ja_inode_flag *node_flag, + unsigned int n, uint8_t *result_key, + enum ja_direction dir) { - return ja_node_get_left(node_flag, JA_ENTRY_PER_NODE); + return ja_node_get_direction(node_flag, n, result_key, dir); +} + +static +struct cds_ja_inode_flag *ja_node_get_minmax(struct cds_ja_inode_flag *node_flag, + uint8_t *result_key, + enum ja_direction dir) +{ + switch (dir) { + case JA_LEFTMOST: + return ja_node_get_direction(node_flag, + -1, result_key, JA_RIGHT); + case JA_RIGHTMOST: + return ja_node_get_direction(node_flag, + JA_ENTRY_PER_NODE, result_key, JA_LEFT); + default: + assert(0); + } } static @@ -1052,7 +1120,7 @@ unsigned int ja_node_sum_distribution_1d(enum ja_recompact mode, for (bit_i = 0; bit_i < JA_BITS_PER_BYTE; bit_i++) { unsigned int distance_to_best; - distance_to_best = abs_int((nr_one[bit_i] << 1U) - distrib_nr_child); + distance_to_best = abs_int(((unsigned int) nr_one[bit_i] << 1U) - distrib_nr_child); if (distance_to_best < overall_best_distance) { overall_best_distance = distance_to_best; bitsel = bit_i; @@ -1111,17 +1179,18 @@ void ja_node_sum_distribution_2d(enum ja_recompact mode, continue; for (bit_i = 0; bit_i < JA_BITS_PER_BYTE; bit_i++) { for (bit_j = 0; bit_j < bit_i; bit_j++) { - if ((v & (1U << bit_i)) && (v & (1U << bit_j))) { - nr_2d_11[bit_i][bit_j]++; - } - if ((v & (1U << bit_i)) && !(v & (1U << bit_j))) { - nr_2d_10[bit_i][bit_j]++; - } - if (!(v & (1U << bit_i)) && (v & (1U << bit_j))) { - nr_2d_01[bit_i][bit_j]++; - } - if (!(v & (1U << bit_i)) && !(v & (1U << bit_j))) { - nr_2d_00[bit_i][bit_j]++; + if (v & (1U << bit_i)) { + if (v & (1U << bit_j)) { + nr_2d_11[bit_i][bit_j]++; + } else { + nr_2d_10[bit_i][bit_j]++; + } + } else { + if (v & (1U << bit_j)) { + nr_2d_01[bit_i][bit_j]++; + } else { + nr_2d_00[bit_i][bit_j]++; + } } } } @@ -1153,17 +1222,18 @@ void ja_node_sum_distribution_2d(enum ja_recompact mode, continue; for (bit_i = 0; bit_i < JA_BITS_PER_BYTE; bit_i++) { for (bit_j = 0; bit_j < bit_i; bit_j++) { - if ((v & (1U << bit_i)) && (v & (1U << bit_j))) { - nr_2d_11[bit_i][bit_j]++; - } - if ((v & (1U << bit_i)) && !(v & (1U << bit_j))) { - nr_2d_10[bit_i][bit_j]++; - } - if (!(v & (1U << bit_i)) && (v & (1U << bit_j))) { - nr_2d_01[bit_i][bit_j]++; - } - if (!(v & (1U << bit_i)) && !(v & (1U << bit_j))) { - nr_2d_00[bit_i][bit_j]++; + if (v & (1U << bit_i)) { + if (v & (1U << bit_j)) { + nr_2d_11[bit_i][bit_j]++; + } else { + nr_2d_10[bit_i][bit_j]++; + } + } else { + if (v & (1U << bit_j)) { + nr_2d_01[bit_i][bit_j]++; + } else { + nr_2d_00[bit_i][bit_j]++; + } } } } @@ -1187,17 +1257,18 @@ void ja_node_sum_distribution_2d(enum ja_recompact mode, continue; for (bit_i = 0; bit_i < JA_BITS_PER_BYTE; bit_i++) { for (bit_j = 0; bit_j < bit_i; bit_j++) { - if ((i & (1U << bit_i)) && (i & (1U << bit_j))) { - nr_2d_11[bit_i][bit_j]++; - } - if ((i & (1U << bit_i)) && !(i & (1U << bit_j))) { - nr_2d_10[bit_i][bit_j]++; - } - if (!(i & (1U << bit_i)) && (i & (1U << bit_j))) { - nr_2d_01[bit_i][bit_j]++; - } - if (!(i & (1U << bit_i)) && !(i & (1U << bit_j))) { - nr_2d_00[bit_i][bit_j]++; + if (i & (1U << bit_i)) { + if (i & (1U << bit_j)) { + nr_2d_11[bit_i][bit_j]++; + } else { + nr_2d_10[bit_i][bit_j]++; + } + } else { + if (i & (1U << bit_j)) { + nr_2d_01[bit_i][bit_j]++; + } else { + nr_2d_00[bit_i][bit_j]++; + } } } } @@ -1216,17 +1287,18 @@ void ja_node_sum_distribution_2d(enum ja_recompact mode, if (mode == JA_RECOMPACT_ADD_NEXT || mode == JA_RECOMPACT_ADD_SAME) { for (bit_i = 0; bit_i < JA_BITS_PER_BYTE; bit_i++) { for (bit_j = 0; bit_j < bit_i; bit_j++) { - if ((n & (1U << bit_i)) && (n & (1U << bit_j))) { - nr_2d_11[bit_i][bit_j]++; - } - if ((n & (1U << bit_i)) && !(n & (1U << bit_j))) { - nr_2d_10[bit_i][bit_j]++; - } - if (!(n & (1U << bit_i)) && (n & (1U << bit_j))) { - nr_2d_01[bit_i][bit_j]++; - } - if (!(n & (1U << bit_i)) && !(n & (1U << bit_j))) { - nr_2d_00[bit_i][bit_j]++; + if (n & (1U << bit_i)) { + if (n & (1U << bit_j)) { + nr_2d_11[bit_i][bit_j]++; + } else { + nr_2d_10[bit_i][bit_j]++; + } + } else { + if (n & (1U << bit_j)) { + nr_2d_01[bit_i][bit_j]++; + } else { + nr_2d_00[bit_i][bit_j]++; + } } } } @@ -1244,10 +1316,10 @@ void ja_node_sum_distribution_2d(enum ja_recompact mode, for (bit_j = 0; bit_j < bit_i; bit_j++) { int distance_to_best[4]; - distance_to_best[0] = (nr_2d_11[bit_i][bit_j] << 2U) - distrib_nr_child; - distance_to_best[1] = (nr_2d_10[bit_i][bit_j] << 2U) - distrib_nr_child; - distance_to_best[2] = (nr_2d_01[bit_i][bit_j] << 2U) - distrib_nr_child; - distance_to_best[3] = (nr_2d_00[bit_i][bit_j] << 2U) - distrib_nr_child; + distance_to_best[0] = ((unsigned int) nr_2d_11[bit_i][bit_j] << 2U) - distrib_nr_child; + distance_to_best[1] = ((unsigned int) nr_2d_10[bit_i][bit_j] << 2U) - distrib_nr_child; + distance_to_best[2] = ((unsigned int) nr_2d_01[bit_i][bit_j] << 2U) - distrib_nr_child; + distance_to_best[3] = ((unsigned int) nr_2d_00[bit_i][bit_j] << 2U) - distrib_nr_child; /* Consider worse distance above best */ if (distance_to_best[1] > 0 && distance_to_best[1] > distance_to_best[0]) @@ -1534,7 +1606,8 @@ skip_copy: dbg_printf("Using fallback for %u children, node type index: %u, mode %s\n", new_shadow_node->nr_child, old_type_index, mode == JA_RECOMPACT_ADD_NEXT ? "add_next" : (mode == JA_RECOMPACT_DEL ? "del" : "add_same")); - uatomic_inc(&ja->node_fallback_count_distribution[new_shadow_node->nr_child]); + if (ja_debug_counters()) + uatomic_inc(&ja->node_fallback_count_distribution[new_shadow_node->nr_child]); } /* Return pointer to new recompacted node through old_node_flag_ptr */ @@ -1600,7 +1673,8 @@ fallback_toosmall: } else { new_type_index++; dbg_printf("Add fallback to type %d\n", new_type_index); - uatomic_inc(&ja->nr_fallback); + if (ja_debug_counters()) + uatomic_inc(&ja->nr_fallback); fallback = 1; goto retry; } @@ -1707,20 +1781,19 @@ int ja_node_clear_ptr(struct cds_ja *ja, return ret; } -struct cds_hlist_head cds_ja_lookup(struct cds_ja *ja, uint64_t key) +struct cds_ja_node *cds_ja_lookup(struct cds_ja *ja, uint64_t key) { unsigned int tree_depth, i; struct cds_ja_inode_flag *node_flag; - struct cds_hlist_head head = { NULL }; - if (caa_unlikely(key > ja->key_max)) - return head; + if (caa_unlikely(key > ja->key_max || key == UINT64_MAX)) + return NULL; tree_depth = ja->tree_depth; node_flag = rcu_dereference(ja->root); /* level 0: root node */ if (!ja_node_ptr(node_flag)) - return head; + return NULL; for (i = 1; i < tree_depth; i++) { uint8_t iter_key; @@ -1730,36 +1803,42 @@ struct cds_hlist_head cds_ja_lookup(struct cds_ja *ja, uint64_t key) dbg_printf("cds_ja_lookup iter key lookup %u finds node_flag %p\n", (unsigned int) iter_key, node_flag); if (!ja_node_ptr(node_flag)) - return head; + return NULL; } /* Last level lookup succeded. We got an actual match. */ - head.next = (struct cds_hlist_node *) node_flag; - return head; + return (struct cds_ja_node *) node_flag; } -/* - * cds_ja_lookup_lower_equal() may need to retry if a concurrent removal - * causes failure to find the previous node. - */ -struct cds_hlist_head cds_ja_lookup_lower_equal(struct cds_ja *ja, uint64_t key) +static +struct cds_ja_node *cds_ja_lookup_inequality(struct cds_ja *ja, uint64_t key, + uint64_t *result_key, enum ja_lookup_inequality mode) { int tree_depth, level; struct cds_ja_inode_flag *node_flag, *cur_node_depth[JA_MAX_DEPTH]; - struct cds_hlist_head head = { NULL }; + uint8_t cur_key[JA_MAX_DEPTH]; + uint64_t _result_key = 0; + enum ja_direction dir; - if (caa_unlikely(key > ja->key_max || !key)) - return head; + switch (mode) { + case JA_LOOKUP_BE: + case JA_LOOKUP_AE: + if (caa_unlikely(key > ja->key_max || key == UINT64_MAX)) + return NULL; + break; + default: + return NULL; + } -retry: memset(cur_node_depth, 0, sizeof(cur_node_depth)); + memset(cur_key, 0, sizeof(cur_key)); tree_depth = ja->tree_depth; node_flag = rcu_dereference(ja->root); cur_node_depth[0] = node_flag; /* level 0: root node */ if (!ja_node_ptr(node_flag)) - return head; + return NULL; for (level = 1; level < tree_depth; level++) { uint8_t iter_key; @@ -1768,73 +1847,113 @@ retry: node_flag = ja_node_get_nth(node_flag, NULL, iter_key); if (!ja_node_ptr(node_flag)) break; + cur_key[level - 1] = iter_key; cur_node_depth[level] = node_flag; - dbg_printf("cds_ja_lookup iter key lookup %u finds node_flag %p\n", + dbg_printf("cds_ja_lookup_inequality iter key lookup %u finds node_flag %p\n", (unsigned int) iter_key, node_flag); } if (level == tree_depth) { /* Last level lookup succeded. We got an equal match. */ - head.next = (struct cds_hlist_node *) node_flag; - return head; + if (result_key) + *result_key = key; + return (struct cds_ja_node *) node_flag; } /* - * Find highest value left of current node. + * Find highest value left/right of current node. * Current node is cur_node_depth[level]. - * Start at current level. If we cannot find any key left of - * ours, go one level up, seek highest value left of current - * (recursively), and when we find one, get the rightmost child - * of its rightmost child (recursively). + * Start at current level. If we cannot find any key left/right + * of ours, go one level up, seek highest value left/right of + * current (recursively), and when we find one, get the + * rightmost/leftmost child of its rightmost/leftmost child + * (recursively). */ + switch (mode) { + case JA_LOOKUP_BE: + dir = JA_LEFT; + break; + case JA_LOOKUP_AE: + dir = JA_RIGHT; + break; + default: + assert(0); + } for (; level > 0; level--) { uint8_t iter_key; iter_key = (uint8_t) (key >> (JA_BITS_PER_BYTE * (tree_depth - level - 1))); - node_flag = ja_node_get_left(cur_node_depth[level - 1], - iter_key); - /* If found left sibling, find rightmost child. */ + node_flag = ja_node_get_leftright(cur_node_depth[level - 1], + iter_key, &cur_key[level - 1], dir); + dbg_printf("cds_ja_lookup_inequality find sibling from %u at %u finds node_flag %p\n", + (unsigned int) iter_key, (unsigned int) cur_key[level - 1], + node_flag); + /* If found left/right sibling, find rightmost/leftmost child. */ if (ja_node_ptr(node_flag)) break; } if (!level) { - /* Reached the root and could not find a left sibling. */ - return head; + /* Reached the root and could not find a left/right sibling. */ + return NULL; } level++; /* - * From this point, we should be able to find a "lower than" - * match. The only reason why we could fail to find such a match - * would be due to a concurrent removal of the branch that - * contains the match. If this happens, we have no choice but to - * retry the entire lookup. Indeed, just because we reached a - * dead-end due to concurrent removal of the branch does not - * mean that other match don't exist. However, this requires - * going up into the tree, hence the retry. + * From this point, we are guaranteed to be able to find a + * "below than"/"above than" match. ja_attach_node() and + * ja_detach_node() both guarantee that it is not possible for a + * lookup to reach a dead-end. */ - /* Find rightmost child of rightmost child (recursively). */ + /* + * Find rightmost/leftmost child of rightmost/leftmost child + * (recursively). + */ + switch (mode) { + case JA_LOOKUP_BE: + dir = JA_RIGHTMOST; + break; + case JA_LOOKUP_AE: + dir = JA_LEFTMOST; + break; + default: + assert(0); + } for (; level < tree_depth; level++) { - node_flag = ja_node_get_rightmost(node_flag); - /* If found left sibling, find rightmost child. */ + node_flag = ja_node_get_minmax(node_flag, &cur_key[level - 1], dir); + dbg_printf("cds_ja_lookup_inequality find minmax at %u finds node_flag %p\n", + (unsigned int) cur_key[level - 1], + node_flag); if (!ja_node_ptr(node_flag)) break; } - if (level != tree_depth) { - /* - * We did not get a match. Caused by concurrent removal. - * We need to retry the entire lookup. - */ - goto retry; + assert(level == tree_depth); + + if (result_key) { + for (level = 1; level < tree_depth; level++) { + _result_key |= ((uint64_t) cur_key[level - 1]) + << (JA_BITS_PER_BYTE * (tree_depth - level - 1)); + } + *result_key = _result_key; } + return (struct cds_ja_node *) node_flag; +} - /* Last level lookup succeded. We got a "lower than" match. */ - head.next = (struct cds_hlist_node *) node_flag; - return head; +struct cds_ja_node *cds_ja_lookup_below_equal(struct cds_ja *ja, + uint64_t key, uint64_t *result_key) +{ + dbg_printf("cds_ja_lookup_below_equal key %" PRIu64 "\n", key); + return cds_ja_lookup_inequality(ja, key, result_key, JA_LOOKUP_BE); +} + +struct cds_ja_node *cds_ja_lookup_above_equal(struct cds_ja *ja, + uint64_t key, uint64_t *result_key) +{ + dbg_printf("cds_ja_lookup_above_equal key %" PRIu64 "\n", key); + return cds_ja_lookup_inequality(ja, key, result_key, JA_LOOKUP_AE); } /* @@ -1848,6 +1967,12 @@ retry: * parent lock (if needed). Then we can proceed to create the new * branch. Publish the new branch, and release locks. * TODO: we currently always take the parent lock even when not needed. + * + * ja_attach_node() ensures that a lookup will _never_ see a branch that + * leads to a dead-end: before attaching a branch, the entire content of + * the new branch is populated, thus creating a cluster, before + * attaching the cluster to the rest of the tree, thus making it visible + * to lookups. */ static int ja_attach_node(struct cds_ja *ja, @@ -1862,7 +1987,6 @@ int ja_attach_node(struct cds_ja *ja, { struct cds_ja_shadow_node *shadow_node = NULL, *parent_shadow_node = NULL; - struct cds_hlist_head head; struct cds_ja_inode_flag *iter_node_flag, *iter_dest_node_flag; int ret, i; struct cds_ja_inode_flag *created_nodes[JA_MAX_DEPTH]; @@ -1931,9 +2055,7 @@ int ja_attach_node(struct cds_ja *ja, } /* Create new branch, starting from bottom */ - CDS_INIT_HLIST_HEAD(&head); - cds_hlist_add_head_rcu(&child_node->list, &head); - iter_node_flag = (struct cds_ja_inode_flag *) head.next; + iter_node_flag = (struct cds_ja_inode_flag *) child_node; for (i = ja->tree_depth - 1; i >= (int) level; i--) { uint8_t iter_key; @@ -2013,9 +2135,9 @@ end: } /* - * Lock the parent containing the hlist head pointer, and add node to list of - * duplicates. Failure can happen if concurrent update changes the - * parent before we get the lock. We return -EAGAIN in that case. + * Lock the parent containing the pointer to list of duplicates, and add + * node to this list. Failure can happen if concurrent update changes + * the parent before we get the lock. We return -EAGAIN in that case. * Return 0 on success, negative error value on failure. */ static @@ -2023,20 +2145,41 @@ int ja_chain_node(struct cds_ja *ja, struct cds_ja_inode_flag *parent_node_flag, struct cds_ja_inode_flag **node_flag_ptr, struct cds_ja_inode_flag *node_flag, + struct cds_ja_node *last_node, struct cds_ja_node *node) { struct cds_ja_shadow_node *shadow_node; - int ret = 0; + struct cds_ja_node *iter_node; + int ret = 0, found = 0; shadow_node = rcuja_shadow_lookup_lock(ja->ht, parent_node_flag); if (!shadow_node) { return -EAGAIN; } - if (ja_node_ptr(*node_flag_ptr) != ja_node_ptr(node_flag)) { + /* + * Ensure that previous node is still there at end of list. + */ + iter_node = (struct cds_ja_node *) ja_node_ptr(node_flag); + if ((struct cds_ja_node *) ja_node_ptr(*node_flag_ptr) != iter_node) { + ret = -EAGAIN; + goto end; + } + cds_ja_for_each_duplicate(iter_node) { + if (iter_node == last_node) + found = 1; + } + if (!found) { ret = -EAGAIN; goto end; } - cds_hlist_add_head_rcu(&node->list, (struct cds_hlist_head *) node_flag_ptr); + /* + * Add node to tail of list to ensure that RCU traversals will + * always see either the prior node or the newly added if + * executed concurrently with a sequence of add followed by del + * on the same key. Safe against concurrent RCU read traversals. + */ + node->next = NULL; + rcu_assign_pointer(last_node->next, node); end: rcuja_shadow_unlock(shadow_node); return ret; @@ -2044,7 +2187,7 @@ end: static int _cds_ja_add(struct cds_ja *ja, uint64_t key, - struct cds_ja_node *new_node, + struct cds_ja_node *node, struct cds_ja_node **unique_node_ret) { unsigned int tree_depth, i; @@ -2058,14 +2201,14 @@ int _cds_ja_add(struct cds_ja *ja, uint64_t key, **node_flag_ptr; int ret; - if (caa_unlikely(key > ja->key_max)) { + if (caa_unlikely(key > ja->key_max || key == UINT64_MAX)) { return -EINVAL; } tree_depth = ja->tree_depth; retry: dbg_printf("cds_ja_add attempt: key %" PRIu64 ", node %p\n", - key, new_node); + key, node); parent2_node_flag = NULL; parent_node_flag = (struct cds_ja_inode_flag *) &ja->root; /* Use root ptr address as key for mutex */ @@ -2108,13 +2251,20 @@ retry: parent_attach_node_flag, node_flag_ptr, node_flag, - key, i, new_node); + key, i, node); } else { + struct cds_ja_node *iter_node, *last_node = NULL; + if (unique_node_ret) { *unique_node_ret = (struct cds_ja_node *) ja_node_ptr(node_flag); return -EEXIST; } + /* Find last duplicate */ + iter_node = (struct cds_ja_node *) ja_node_ptr(node_flag); + cds_ja_for_each_duplicate_rcu(iter_node) + last_node = iter_node; + dbg_printf("cds_ja_add duplicate parent2_node_flag %p parent_node_flag %p node_flag_ptr %p node_flag %p\n", parent2_node_flag, parent_node_flag, node_flag_ptr, node_flag); @@ -2126,7 +2276,8 @@ retry: parent_attach_node_flag, attach_node_flag_ptr, attach_node_flag, - new_node); + last_node, + node); } if (ret == -EAGAIN || ret == -EEXIST) goto retry; @@ -2135,22 +2286,22 @@ retry: } int cds_ja_add(struct cds_ja *ja, uint64_t key, - struct cds_ja_node *new_node) + struct cds_ja_node *node) { - return _cds_ja_add(ja, key, new_node, NULL); + return _cds_ja_add(ja, key, node, NULL); } struct cds_ja_node *cds_ja_add_unique(struct cds_ja *ja, uint64_t key, - struct cds_ja_node *new_node) + struct cds_ja_node *node) { int ret; struct cds_ja_node *ret_node; - ret = _cds_ja_add(ja, key, new_node, &ret_node); + ret = _cds_ja_add(ja, key, node, &ret_node); if (ret == -EEXIST) return ret_node; else - return new_node; + return node; } /* @@ -2163,6 +2314,11 @@ struct cds_ja_node *cds_ja_add_unique(struct cds_ja *ja, uint64_t key, * However, when a child is removed from "linear" nodes, its pointer * is set to NULL. We therefore check, while holding the locks, if this * pointer is NULL, and return -ENOENT to the caller if it is the case. + * + * ja_detach_node() ensures that a lookup will _never_ see a branch that + * leads to a dead-end: when removing branch, it makes sure to perform + * the "cut" at the highest node that has only one child, effectively + * replacing it with a NULL pointer. */ static int ja_detach_node(struct cds_ja *ja, @@ -2317,8 +2473,7 @@ int ja_unchain_node(struct cds_ja *ja, struct cds_ja_node *node) { struct cds_ja_shadow_node *shadow_node; - struct cds_hlist_node *hlist_node; - struct cds_hlist_head hlist_head; + struct cds_ja_node *iter_node, **iter_node_ptr, **prev_node_ptr = NULL; int ret = 0, count = 0, found = 0; shadow_node = rcuja_shadow_lookup_lock(ja->ht, parent_node_flag); @@ -2328,28 +2483,30 @@ int ja_unchain_node(struct cds_ja *ja, ret = -EAGAIN; goto end; } - hlist_head.next = (struct cds_hlist_node *) ja_node_ptr(node_flag); /* - * Retry if another thread removed all but one of duplicates - * since check (this check was performed without lock). - * Ensure that the node we are about to remove is still in the - * list (while holding lock). + * Find the previous node's next pointer pointing to our node, + * so we can update it. Retry if another thread removed all but + * one of duplicates since check (this check was performed + * without lock). Ensure that the node we are about to remove is + * still in the list (while holding lock). No need for RCU + * traversal here since we hold the lock on the parent. */ - cds_hlist_for_each_rcu(hlist_node, &hlist_head) { - if (count == 0) { - /* FIXME: currently a work-around */ - hlist_node->prev = (struct cds_hlist_node *) node_flag_ptr; - } + iter_node_ptr = (struct cds_ja_node **) node_flag_ptr; + iter_node = (struct cds_ja_node *) ja_node_ptr(node_flag); + cds_ja_for_each_duplicate(iter_node) { count++; - if (hlist_node == &node->list) + if (iter_node == node) { + prev_node_ptr = iter_node_ptr; found++; + } + iter_node_ptr = &iter_node->next; } assert(found <= 1); if (!found || count == 1) { ret = -EAGAIN; goto end; } - cds_hlist_del_rcu(&node->list); + CMM_STORE_SHARED(*prev_node_ptr, node->next); /* * Validate that we indeed removed the node from linked list. */ @@ -2375,7 +2532,7 @@ int cds_ja_del(struct cds_ja *ja, uint64_t key, int nr_snapshot; int ret; - if (caa_unlikely(key > ja->key_max)) + if (caa_unlikely(key > ja->key_max || key == UINT64_MAX)) return -EINVAL; tree_depth = ja->tree_depth; @@ -2424,22 +2581,17 @@ retry: key); return -ENOENT; } else { - struct cds_hlist_head hlist_head; - struct cds_hlist_node *hlist_node; - struct cds_ja_node *entry, *match = NULL; + struct cds_ja_node *iter_node, *match = NULL; int count = 0; - hlist_head.next = - (struct cds_hlist_node *) ja_node_ptr(node_flag); - cds_hlist_for_each_entry_rcu(entry, - hlist_node, - &hlist_head, - list) { - dbg_printf("cds_ja_del: compare %p with entry %p\n", node, entry); - if (entry == node) - match = entry; + iter_node = (struct cds_ja_node *) ja_node_ptr(node_flag); + cds_ja_for_each_duplicate_rcu(iter_node) { + dbg_printf("cds_ja_del: compare %p with iter_node %p\n", node, iter_node); + if (iter_node == node) + match = iter_node; count++; } + if (!match) { dbg_printf("cds_ja_del: no node match for node %p key %" PRIu64 "\n", node, key); return -ENOENT; @@ -2528,105 +2680,6 @@ ja_error: return NULL; } -/* - * Called from RCU read-side CS. - */ -__attribute__((visibility("protected"))) -void rcuja_free_all_children(struct cds_ja_shadow_node *shadow_node, - struct cds_ja_inode_flag *node_flag, - void (*free_node_cb)(struct rcu_head *head)) -{ - const struct rcu_flavor_struct *flavor; - unsigned int type_index; - struct cds_ja_inode *node; - const struct cds_ja_type *type; - - flavor = cds_lfht_rcu_flavor(shadow_node->ja->ht); - node = ja_node_ptr(node_flag); - assert(node != NULL); - type_index = ja_node_type(node_flag); - type = &ja_types[type_index]; - - switch (type->type_class) { - case RCU_JA_LINEAR: - { - uint8_t nr_child = - ja_linear_node_get_nr_child(type, node); - unsigned int i; - - for (i = 0; i < nr_child; i++) { - struct cds_ja_inode_flag *iter; - struct cds_hlist_head head; - struct cds_ja_node *entry; - struct cds_hlist_node *pos, *tmp; - uint8_t v; - - ja_linear_node_get_ith_pos(type, node, i, &v, &iter); - if (!iter) - continue; - head.next = (struct cds_hlist_node *) iter; - cds_hlist_for_each_entry_safe(entry, pos, tmp, &head, list) { - flavor->update_call_rcu(&entry->head, free_node_cb); - } - } - break; - } - case RCU_JA_POOL: - { - unsigned int pool_nr; - - for (pool_nr = 0; pool_nr < (1U << type->nr_pool_order); pool_nr++) { - struct cds_ja_inode *pool = - ja_pool_node_get_ith_pool(type, node, pool_nr); - uint8_t nr_child = - ja_linear_node_get_nr_child(type, pool); - unsigned int j; - - for (j = 0; j < nr_child; j++) { - struct cds_ja_inode_flag *iter; - struct cds_hlist_head head; - struct cds_ja_node *entry; - struct cds_hlist_node *pos, *tmp; - uint8_t v; - - ja_linear_node_get_ith_pos(type, pool, j, &v, &iter); - if (!iter) - continue; - head.next = (struct cds_hlist_node *) iter; - cds_hlist_for_each_entry_safe(entry, pos, tmp, &head, list) { - flavor->update_call_rcu(&entry->head, free_node_cb); - } - } - } - break; - } - case RCU_JA_NULL: - break; - case RCU_JA_PIGEON: - { - unsigned int i; - - for (i = 0; i < JA_ENTRY_PER_NODE; i++) { - struct cds_ja_inode_flag *iter; - struct cds_hlist_head head; - struct cds_ja_node *entry; - struct cds_hlist_node *pos, *tmp; - - iter = ja_pigeon_node_get_ith_pos(type, node, i); - if (!iter) - continue; - head.next = (struct cds_hlist_node *) iter; - cds_hlist_for_each_entry_safe(entry, pos, tmp, &head, list) { - flavor->update_call_rcu(&entry->head, free_node_cb); - } - } - break; - } - default: - assert(0); - } -} - static void print_debug_fallback_distribution(struct cds_ja *ja) { @@ -2648,6 +2701,9 @@ int ja_final_checks(struct cds_ja *ja) unsigned long na, nf, nr_fallback; int ret = 0; + if (!ja_debug_counters()) + return 0; + fallback_ratio = (double) uatomic_read(&ja->nr_fallback); fallback_ratio /= (double) uatomic_read(&ja->nr_nodes_allocated); nr_fallback = uatomic_read(&ja->nr_fallback); @@ -2672,19 +2728,18 @@ int ja_final_checks(struct cds_ja *ja) } /* - * There should be no more concurrent add to the judy array while it is - * being destroyed (ensured by the caller). + * There should be no more concurrent add, delete, nor look-up performed + * on the Judy array while it is being destroyed (ensured by the + * caller). */ -int cds_ja_destroy(struct cds_ja *ja, - void (*free_node_cb)(struct rcu_head *head)) +int cds_ja_destroy(struct cds_ja *ja) { const struct rcu_flavor_struct *flavor; int ret; flavor = cds_lfht_rcu_flavor(ja->ht); rcuja_shadow_prune(ja->ht, - RCUJA_SHADOW_CLEAR_FREE_NODE | RCUJA_SHADOW_CLEAR_FREE_LOCK, - free_node_cb); + RCUJA_SHADOW_CLEAR_FREE_NODE | RCUJA_SHADOW_CLEAR_FREE_LOCK); flavor->thread_offline(); ret = rcuja_delete_ht(ja->ht); if (ret)