* implementation:
*
* - RCU read-side critical section allows readers to perform hash
- * table lookups and use the returned objects safely by delaying
- * memory reclaim of a grace period.
+ * table lookups, as well as traversals, and use the returned objects
+ * safely by allowing memory reclaim to take place only after a grace
+ * period.
* - Add and remove operations are lock-free, and do not need to
* allocate memory. They need to be executed within RCU read-side
* critical section to ensure the objects they read are valid and to
* deal with the cmpxchg ABA problem.
* - add and add_unique operations are supported. add_unique checks if
- * the node key already exists in the hash table. It ensures no key
- * duplicata exists.
- * - The resize operation executes concurrently with add/remove/lookup.
+ * the node key already exists in the hash table. It ensures not to
+ * populate a duplicate key if the node key already exists in the hash
+ * table.
+ * - The resize operation executes concurrently with
+ * add/add_unique/add_replace/remove/lookup/traversal.
* - Hash table nodes are contained within a split-ordered list. This
* list is ordered by incrementing reversed-bits-hash value.
* - An index of bucket nodes is kept. These bucket nodes are the hash
- * table "buckets", and they are also chained together in the
- * split-ordered list, which allows recursive expansion.
- * - The resize operation for small tables only allows expanding the hash table.
- * It is triggered automatically by detecting long chains in the add
- * operation.
+ * table "buckets". These buckets are internal nodes that allow to
+ * perform a fast hash lookup, similarly to a skip list. These
+ * buckets are chained together in the split-ordered list, which
+ * allows recursive expansion by inserting new buckets between the
+ * existing buckets. The split-ordered list allows adding new buckets
+ * between existing buckets as the table needs to grow.
+ * - The resize operation for small tables only allows expanding the
+ * hash table. It is triggered automatically by detecting long chains
+ * in the add operation.
* - The resize operation for larger tables (and available through an
* API) allows both expanding and shrinking the hash table.
* - Split-counters are used to keep track of the number of
* (not visible to lookups anymore) before the RCU read-side critical
* section held across removal ends. Furthermore, this ensures that
* the node with "removed" flag set is removed from the linked-list
- * before its memory is reclaimed. Only the thread which removal
- * successfully set the "removed" flag (with a cmpxchg) into a node's
- * next pointer is considered to have succeeded its removal (and thus
- * owns the node to reclaim). Because we garbage-collect starting from
- * an invariant node (the start-of-bucket bucket node) up to the
- * "removed" node (or find a reverse-hash that is higher), we are sure
- * that a successful traversal of the chain leads to a chain that is
- * present in the linked-list (the start node is never removed) and
- * that is does not contain the "removed" node anymore, even if
- * concurrent delete/add operations are changing the structure of the
- * list concurrently.
- * - The add operation performs gargage collection of buckets if it
- * encounters nodes with removed flag set in the bucket where it wants
- * to add its new node. This ensures lock-freedom of add operation by
+ * before its memory is reclaimed. After setting the "removal" flag,
+ * only the thread which removal is the first to set the "removal
+ * owner" flag (with an xchg) into a node's next pointer is considered
+ * to have succeeded its removal (and thus owns the node to reclaim).
+ * Because we garbage-collect starting from an invariant node (the
+ * start-of-bucket bucket node) up to the "removed" node (or find a
+ * reverse-hash that is higher), we are sure that a successful
+ * traversal of the chain leads to a chain that is present in the
+ * linked-list (the start node is never removed) and that it does not
+ * contain the "removed" node anymore, even if concurrent delete/add
+ * operations are changing the structure of the list concurrently.
+ * - The add operations perform garbage collection of buckets if they
+ * encounter nodes with removed flag set in the bucket where they want
+ * to add their new node. This ensures lock-freedom of add operation by
* helping the remover unlink nodes from the list rather than to wait
* for it do to so.
- * - A RCU "order table" indexed by log2(hash index) is copied and
- * expanded by the resize operation. This order table allows finding
- * the "bucket node" tables.
- * - There is one bucket node table per hash index order. The size of
- * each bucket node table is half the number of hashes contained in
- * this order (except for order 0).
- * - synchronzie_rcu is used to garbage-collect the old bucket node table.
- * - The per-order bucket node tables contain a compact version of the
- * hash table nodes. These tables are invariant after they are
- * populated into the hash table.
+ * - There are three memory backends for the hash table buckets: the
+ * "order table", the "chunks", and the "mmap".
+ * - These bucket containers contain a compact version of the hash table
+ * nodes.
+ * - The RCU "order table":
+ * - has a first level table indexed by log2(hash index) which is
+ * copied and expanded by the resize operation. This order table
+ * allows finding the "bucket node" tables.
+ * - There is one bucket node table per hash index order. The size of
+ * each bucket node table is half the number of hashes contained in
+ * this order (except for order 0).
+ * - The RCU "chunks" is best suited for close interaction with a page
+ * allocator. It uses a linear array as index to "chunks" containing
+ * each the same number of buckets.
+ * - The RCU "mmap" memory backend uses a single memory map to hold
+ * all buckets.
+ * - synchronize_rcu is used to garbage-collect the old bucket node table.
+ *
+ * Ordering Guarantees:
+ *
+ * To discuss these guarantees, we first define "read" operation as any
+ * of the the basic cds_lfht_lookup, cds_lfht_next_duplicate,
+ * cds_lfht_first, cds_lfht_next operation, as well as
+ * cds_lfht_add_unique (failure).
+ *
+ * We define "read traversal" operation as any of the following
+ * group of operations
+ * - cds_lfht_lookup followed by iteration with cds_lfht_next_duplicate
+ * (and/or cds_lfht_next, although less common).
+ * - cds_lfht_add_unique (failure) followed by iteration with
+ * cds_lfht_next_duplicate (and/or cds_lfht_next, although less
+ * common).
+ * - cds_lfht_first followed iteration with cds_lfht_next (and/or
+ * cds_lfht_next_duplicate, although less common).
+ *
+ * We define "write" operations as any of cds_lfht_add, cds_lfht_replace,
+ * cds_lfht_add_unique (success), cds_lfht_add_replace, cds_lfht_del.
+ *
+ * When cds_lfht_add_unique succeeds (returns the node passed as
+ * parameter), it acts as a "write" operation. When cds_lfht_add_unique
+ * fails (returns a node different from the one passed as parameter), it
+ * acts as a "read" operation. A cds_lfht_add_unique failure is a
+ * cds_lfht_lookup "read" operation, therefore, any ordering guarantee
+ * referring to "lookup" imply any of "lookup" or cds_lfht_add_unique
+ * (failure).
+ *
+ * We define "prior" and "later" node as nodes observable by reads and
+ * read traversals respectively before and after a write or sequence of
+ * write operations.
+ *
+ * Hash-table operations are often cascaded, for example, the pointer
+ * returned by a cds_lfht_lookup() might be passed to a cds_lfht_next(),
+ * whose return value might in turn be passed to another hash-table
+ * operation. This entire cascaded series of operations must be enclosed
+ * by a pair of matching rcu_read_lock() and rcu_read_unlock()
+ * operations.
+ *
+ * The following ordering guarantees are offered by this hash table:
+ *
+ * A.1) "read" after "write": if there is ordering between a write and a
+ * later read, then the read is guaranteed to see the write or some
+ * later write.
+ * A.2) "read traversal" after "write": given that there is dependency
+ * ordering between reads in a "read traversal", if there is
+ * ordering between a write and the first read of the traversal,
+ * then the "read traversal" is guaranteed to see the write or
+ * some later write.
+ * B.1) "write" after "read": if there is ordering between a read and a
+ * later write, then the read will never see the write.
+ * B.2) "write" after "read traversal": given that there is dependency
+ * ordering between reads in a "read traversal", if there is
+ * ordering between the last read of the traversal and a later
+ * write, then the "read traversal" will never see the write.
+ * C) "write" while "read traversal": if a write occurs during a "read
+ * traversal", the traversal may, or may not, see the write.
+ * D.1) "write" after "write": if there is ordering between a write and
+ * a later write, then the later write is guaranteed to see the
+ * effects of the first write.
+ * D.2) Concurrent "write" pairs: The system will assign an arbitrary
+ * order to any pair of concurrent conflicting writes.
+ * Non-conflicting writes (for example, to different keys) are
+ * unordered.
+ * E) If a grace period separates a "del" or "replace" operation
+ * and a subsequent operation, then that subsequent operation is
+ * guaranteed not to see the removed item.
+ * F) Uniqueness guarantee: given a hash table that does not contain
+ * duplicate items for a given key, there will only be one item in
+ * the hash table after an arbitrary sequence of add_unique and/or
+ * add_replace operations. Note, however, that a pair of
+ * concurrent read operations might well access two different items
+ * with that key.
+ * G.1) If a pair of lookups for a given key are ordered (e.g. by a
+ * memory barrier), then the second lookup will return the same
+ * node as the previous lookup, or some later node.
+ * G.2) A "read traversal" that starts after the end of a prior "read
+ * traversal" (ordered by memory barriers) is guaranteed to see the
+ * same nodes as the previous traversal, or some later nodes.
+ * G.3) Concurrent "read" pairs: concurrent reads are unordered. For
+ * example, if a pair of reads to the same key run concurrently
+ * with an insertion of that same key, the reads remain unordered
+ * regardless of their return values. In other words, you cannot
+ * rely on the values returned by the reads to deduce ordering.
+ *
+ * Progress guarantees:
+ *
+ * * Reads are wait-free. These operations always move forward in the
+ * hash table linked list, and this list has no loop.
+ * * Writes are lock-free. Any retry loop performed by a write operation
+ * is triggered by progress made within another update operation.
*
* Bucket node tables:
*
* shrink hash table from order 6 to 5: fini the index=6 bucket node table
*
* A bit of ascii art explanation:
- *
- * Order index is the off-by-one compare to the actual power of 2 because
- * we use index 0 to deal with the 0 special-case.
- *
+ *
+ * The order index is the off-by-one compared to the actual power of 2
+ * because we use index 0 to deal with the 0 special-case.
+ *
* This shows the nodes for a small table ordered by reversed bits:
- *
+ *
* bits reverse
* 0 000 000
* 4 100 001
* 5 101 101
* 3 011 110
* 7 111 111
- *
- * This shows the nodes in order of non-reversed bits, linked by
+ *
+ * This shows the nodes in order of non-reversed bits, linked by
* reversed-bit order.
- *
+ *
* order bits reverse
* 0 0 000 000
* 1 | 1 001 100 <-
*/
#define _LGPL_SOURCE
+#define _GNU_SOURCE
#include <stdlib.h>
#include <errno.h>
#include <assert.h>
#include <stdio.h>
#include <stdint.h>
#include <string.h>
+#include <sched.h>
+#include <unistd.h>
#include "config.h"
-#include <urcu.h>
+#include "compat-getcpu.h"
+#include <urcu-pointer.h>
#include <urcu-call-rcu.h>
#include <urcu-flavor.h>
#include <urcu/arch.h>
* Split-counters lazily update the global counter each 1024
* addition/removal. It automatically keeps track of resize required.
* We use the bucket length as indicator for need to expand for small
- * tables and machines lacking per-cpu data suppport.
+ * tables and machines lacking per-cpu data support.
*/
#define COUNT_COMMIT_ORDER 10
#define DEFAULT_SPLIT_COUNT_MASK 0xFUL
* removal, and that node garbage collection must be performed.
* The bucket flag does not require to be updated atomically with the
* pointer, but it is added as a pointer low bit flag to save space.
+ * The "removal owner" flag is used to detect which of the "del"
+ * operation that has set the "removed flag" gets to return the removed
+ * node to its caller. Note that the replace operation does not need to
+ * iteract with the "removal owner" flag, because it validates that
+ * the "removed" flag is not set before performing its cmpxchg.
*/
#define REMOVED_FLAG (1UL << 0)
#define BUCKET_FLAG (1UL << 1)
* Originally from Public Domain.
*/
-static const uint8_t BitReverseTable256[256] =
+static const uint8_t BitReverseTable256[256] =
{
#define R2(n) (n), (n) + 2*64, (n) + 1*64, (n) + 3*64
#define R4(n) R2(n), R2((n) + 2*16), R2((n) + 1*16), R2((n) + 3*16)
return BitReverseTable256[v];
}
-static __attribute__((unused))
+#if (CAA_BITS_PER_LONG == 32)
+static
uint32_t bit_reverse_u32(uint32_t v)
{
- return ((uint32_t) bit_reverse_u8(v) << 24) |
- ((uint32_t) bit_reverse_u8(v >> 8) << 16) |
- ((uint32_t) bit_reverse_u8(v >> 16) << 8) |
+ return ((uint32_t) bit_reverse_u8(v) << 24) |
+ ((uint32_t) bit_reverse_u8(v >> 8) << 16) |
+ ((uint32_t) bit_reverse_u8(v >> 16) << 8) |
((uint32_t) bit_reverse_u8(v >> 24));
}
-
-static __attribute__((unused))
+#else
+static
uint64_t bit_reverse_u64(uint64_t v)
{
- return ((uint64_t) bit_reverse_u8(v) << 56) |
- ((uint64_t) bit_reverse_u8(v >> 8) << 48) |
+ return ((uint64_t) bit_reverse_u8(v) << 56) |
+ ((uint64_t) bit_reverse_u8(v >> 8) << 48) |
((uint64_t) bit_reverse_u8(v >> 16) << 40) |
((uint64_t) bit_reverse_u8(v >> 24) << 32) |
- ((uint64_t) bit_reverse_u8(v >> 32) << 24) |
- ((uint64_t) bit_reverse_u8(v >> 40) << 16) |
+ ((uint64_t) bit_reverse_u8(v >> 32) << 24) |
+ ((uint64_t) bit_reverse_u8(v >> 40) << 16) |
((uint64_t) bit_reverse_u8(v >> 48) << 8) |
((uint64_t) bit_reverse_u8(v >> 56));
}
+#endif
static
unsigned long bit_reverse_ulong(unsigned long v)
{
int r;
- asm("bsrl %1,%0\n\t"
+ __asm__ ("bsrl %1,%0\n\t"
"jnz 1f\n\t"
"movl $-1,%0\n\t"
"1:\n\t"
{
long r;
- asm("bsrq %1,%0\n\t"
+ __asm__ ("bsrq %1,%0\n\t"
"jnz 1f\n\t"
"movq $-1,%0\n\t"
"1:\n\t"
static long nr_cpus_mask = -1;
static long split_count_mask = -1;
+static int split_count_order = -1;
#if defined(HAVE_SYSCONF)
static void ht_init_nr_cpus_mask(void)
static
void alloc_split_items_count(struct cds_lfht *ht)
{
- struct ht_items_count *count;
-
if (nr_cpus_mask == -1) {
ht_init_nr_cpus_mask();
if (nr_cpus_mask < 0)
split_count_mask = DEFAULT_SPLIT_COUNT_MASK;
else
split_count_mask = nr_cpus_mask;
+ split_count_order =
+ cds_lfht_get_count_order_ulong(split_count_mask + 1);
}
assert(split_count_mask >= 0);
if (ht->flags & CDS_LFHT_ACCOUNTING) {
- ht->split_count = calloc(split_count_mask + 1, sizeof(*count));
+ ht->split_count = calloc(split_count_mask + 1,
+ sizeof(struct ht_items_count));
assert(ht->split_count);
} else {
ht->split_count = NULL;
poison_free(ht->split_count);
}
-#if defined(HAVE_SCHED_GETCPU)
static
int ht_get_split_count_index(unsigned long hash)
{
int cpu;
assert(split_count_mask >= 0);
- cpu = sched_getcpu();
+ cpu = urcu_sched_getcpu();
if (caa_unlikely(cpu < 0))
return hash & split_count_mask;
else
return cpu & split_count_mask;
}
-#else /* #if defined(HAVE_SCHED_GETCPU) */
-static
-int ht_get_split_count_index(unsigned long hash)
-{
- return hash & split_count_mask;
-}
-#endif /* #else #if defined(HAVE_SCHED_GETCPU) */
static
void ht_count_add(struct cds_lfht *ht, unsigned long size, unsigned long hash)
* Use bucket-local length for small table expand and for
* environments lacking per-cpu data support.
*/
- if (count >= (1UL << COUNT_COMMIT_ORDER))
+ if (count >= (1UL << (COUNT_COMMIT_ORDER + split_count_order)))
return;
if (chain_len > 100)
dbg_printf("WARNING: large chain length: %u.\n",
chain_len);
- if (chain_len >= CHAIN_LEN_RESIZE_THRESHOLD)
- cds_lfht_resize_lazy_grow(ht, size,
- cds_lfht_get_count_order_u32(chain_len - (CHAIN_LEN_TARGET - 1)));
+ if (chain_len >= CHAIN_LEN_RESIZE_THRESHOLD) {
+ int growth;
+
+ /*
+ * Ideal growth calculated based on chain length.
+ */
+ growth = cds_lfht_get_count_order_u32(chain_len
+ - (CHAIN_LEN_TARGET - 1));
+ if ((ht->flags & CDS_LFHT_ACCOUNTING)
+ && (size << growth)
+ >= (1UL << (COUNT_COMMIT_ORDER
+ + split_count_order))) {
+ /*
+ * If ideal growth expands the hash table size
+ * beyond the "small hash table" sizes, use the
+ * maximum small hash table size to attempt
+ * expanding the hash table. This only applies
+ * when node accounting is available, otherwise
+ * the chain length is used to expand the hash
+ * table in every case.
+ */
+ growth = COUNT_COMMIT_ORDER + split_count_order
+ - cds_lfht_get_count_order_ulong(size);
+ if (growth <= 0)
+ return;
+ }
+ cds_lfht_resize_lazy_grow(ht, size, growth);
+ }
}
static
return ((unsigned long) node) & REMOVED_FLAG;
}
-static
-struct cds_lfht_node *flag_removed(struct cds_lfht_node *node)
-{
- return (struct cds_lfht_node *) (((unsigned long) node) | REMOVED_FLAG);
-}
-
static
int is_bucket(struct cds_lfht_node *node)
{
return (struct cds_lfht_node *) (((unsigned long) node) | REMOVAL_OWNER_FLAG);
}
+static
+struct cds_lfht_node *flag_removed_or_removal_owner(struct cds_lfht_node *node)
+{
+ return (struct cds_lfht_node *) (((unsigned long) node) | REMOVED_FLAG | REMOVAL_OWNER_FLAG);
+}
+
static
struct cds_lfht_node *get_end(void)
{
assert(!is_bucket(bucket));
assert(!is_removed(bucket));
+ assert(!is_removal_owner(bucket));
assert(!is_bucket(node));
assert(!is_removed(node));
+ assert(!is_removal_owner(node));
for (;;) {
iter_prev = bucket;
/* We can always skip the bucket node initially */
iter = rcu_dereference(iter_prev->next);
assert(!is_removed(iter));
+ assert(!is_removal_owner(iter));
assert(iter_prev->reverse_hash <= node->reverse_hash);
/*
* We should never be called with bucket (start of chain)
iter = next;
}
assert(!is_removed(iter));
+ assert(!is_removal_owner(iter));
if (is_bucket(iter))
new_next = flag_bucket(clear_flag(next));
else
return -ENOENT;
assert(!is_removed(old_node));
+ assert(!is_removal_owner(old_node));
assert(!is_bucket(old_node));
assert(!is_removed(new_node));
+ assert(!is_removal_owner(new_node));
assert(!is_bucket(new_node));
assert(new_node != old_node);
for (;;) {
*/
return -ENOENT;
}
- assert(!is_bucket(old_next));
- assert(new_node != clear_flag(old_next));
- new_node->next = clear_flag(old_next);
+ assert(old_next == clear_flag(old_next));
+ assert(new_node != old_next);
+ /*
+ * REMOVAL_OWNER flag is _NEVER_ set before the REMOVED
+ * flag. It is either set atomically at the same time
+ * (replace) or after (del).
+ */
+ assert(!is_removal_owner(old_next));
+ new_node->next = old_next;
/*
* Here is the whole trick for lock-free replace: we add
* the replacement node _after_ the node we want to
* the old node, but will not see the new one.
* This is a replacement of a node with another node
* that has the same value: we are therefore not
- * removing a value from the hash table.
+ * removing a value from the hash table. We set both the
+ * REMOVED and REMOVAL_OWNER flags atomically so we own
+ * the node after successful cmpxchg.
*/
ret_next = uatomic_cmpxchg(&old_node->next,
- old_next, flag_removed(new_node));
+ old_next, flag_removed_or_removal_owner(new_node));
if (ret_next == old_next)
break; /* We performed the replacement. */
old_next = ret_next;
bucket = lookup_bucket(ht, size, bit_reverse_ulong(old_node->reverse_hash));
_cds_lfht_gc_bucket(bucket, new_node);
- assert(is_removed(rcu_dereference(old_node->next)));
+ assert(is_removed(CMM_LOAD_SHARED(old_node->next)));
return 0;
}
*/
static
void _cds_lfht_add(struct cds_lfht *ht,
+ unsigned long hash,
cds_lfht_match_fct match,
const void *key,
unsigned long size,
assert(!is_bucket(node));
assert(!is_removed(node));
- bucket = lookup_bucket(ht, size, bit_reverse_ulong(node->reverse_hash));
+ assert(!is_removal_owner(node));
+ bucket = lookup_bucket(ht, size, hash);
for (;;) {
uint32_t chain_len = 0;
*
* This semantic ensures no duplicated keys
* should ever be observable in the table
- * (including observe one node by one node
- * by forward iterations)
+ * (including traversing the table node by
+ * node by forward iterations)
*/
cds_lfht_next_duplicate(ht, match, key, &d_iter);
if (!d_iter.node)
insert:
assert(node != clear_flag(iter));
assert(!is_removed(iter_prev));
+ assert(!is_removal_owner(iter_prev));
assert(!is_removed(iter));
+ assert(!is_removal_owner(iter));
assert(iter_prev != node);
if (!bucket_flag)
node->next = clear_flag(iter);
gc_node:
assert(!is_removed(iter));
+ assert(!is_removal_owner(iter));
if (is_bucket(iter))
new_next = flag_bucket(clear_flag(next));
else
* logical removal flag). Return -ENOENT if the node had
* previously been removed.
*/
- next = rcu_dereference(node->next);
+ next = CMM_LOAD_SHARED(node->next); /* next is not dereferenced */
if (caa_unlikely(is_removed(next)))
return -ENOENT;
assert(!is_bucket(next));
+ /*
+ * The del operation semantic guarantees a full memory barrier
+ * before the uatomic_or atomic commit of the deletion flag.
+ */
+ cmm_smp_mb__before_uatomic_or();
/*
* We set the REMOVED_FLAG unconditionally. Note that there may
* be more than one concurrent thread setting this flag.
bucket = lookup_bucket(ht, size, bit_reverse_ulong(node->reverse_hash));
_cds_lfht_gc_bucket(bucket, node);
- assert(is_removed(rcu_dereference(node->next)));
+ assert(is_removed(CMM_LOAD_SHARED(node->next)));
/*
* Last phase: atomically exchange node->next with a version
* having "REMOVAL_OWNER_FLAG" set. If the returned node->next
void (*fct)(struct cds_lfht *ht, unsigned long i,
unsigned long start, unsigned long len))
{
- unsigned long partition_len;
+ unsigned long partition_len, start = 0;
struct partition_resize_work *work;
int thread, ret;
unsigned long nr_threads;
+ assert(nr_cpus_mask != -1);
+ if (nr_cpus_mask < 0 || len < 2 * MIN_PARTITION_PER_THREAD)
+ goto fallback;
+
/*
* Note: nr_cpus_mask + 1 is always power of 2.
* We spawn just the number of threads we need to satisfy the minimum
}
partition_len = len >> cds_lfht_get_count_order_ulong(nr_threads);
work = calloc(nr_threads, sizeof(*work));
- assert(work);
+ if (!work) {
+ dbg_printf("error allocating for resize, single-threading\n");
+ goto fallback;
+ }
for (thread = 0; thread < nr_threads; thread++) {
work[thread].ht = ht;
work[thread].i = i;
work[thread].fct = fct;
ret = pthread_create(&(work[thread].thread_id), ht->resize_attr,
partition_resize_thread, &work[thread]);
+ if (ret == EAGAIN) {
+ /*
+ * Out of resources: wait and join the threads
+ * we've created, then handle leftovers.
+ */
+ dbg_printf("error spawning for resize, single-threading\n");
+ start = work[thread].start;
+ len -= start;
+ nr_threads = thread;
+ break;
+ }
assert(!ret);
}
for (thread = 0; thread < nr_threads; thread++) {
assert(!ret);
}
free(work);
+
+ /*
+ * A pthread_create failure above will either lead in us having
+ * no threads to join or starting at a non-zero offset,
+ * fallback to single thread processing of leftovers.
+ */
+ if (start == 0 && nr_threads > 0)
+ return;
+fallback:
+ ht->flavor->thread_online();
+ fct(ht, i, start, len);
+ ht->flavor->thread_offline();
}
/*
dbg_printf("init populate: order %lu index %lu hash %lu\n",
i, j, j);
new_node->reverse_hash = bit_reverse_ulong(j);
- _cds_lfht_add(ht, NULL, NULL, size, new_node, NULL, 1);
+ _cds_lfht_add(ht, j, NULL, NULL, size, new_node, NULL, 1);
}
ht->flavor->read_unlock();
}
void init_table_populate(struct cds_lfht *ht, unsigned long i,
unsigned long len)
{
- assert(nr_cpus_mask != -1);
- if (nr_cpus_mask < 0 || len < 2 * MIN_PARTITION_PER_THREAD) {
- ht->flavor->thread_online();
- init_table_populate_partition(ht, i, 0, len);
- ht->flavor->thread_offline();
- return;
- }
partition_resize_helper(ht, i, len, init_table_populate_partition);
}
* removed nodes have been garbage-collected (unlinked) before call_rcu is
* invoked to free a hole level of bucket nodes (after a grace period).
*
- * Logical removal and garbage collection can therefore be done in batch or on a
- * node-per-node basis, as long as the guarantee above holds.
+ * Logical removal and garbage collection can therefore be done in batch
+ * or on a node-per-node basis, as long as the guarantee above holds.
*
* When we reach a certain length, we can split this removal over many worker
* threads, based on the number of CPUs available in the system. This should
static
void remove_table(struct cds_lfht *ht, unsigned long i, unsigned long len)
{
-
- assert(nr_cpus_mask != -1);
- if (nr_cpus_mask < 0 || len < 2 * MIN_PARTITION_PER_THREAD) {
- ht->flavor->thread_online();
- remove_table_partition(ht, i, 0, len);
- ht->flavor->thread_offline();
- return;
- }
partition_resize_helper(ht, i, len, remove_table_partition);
}
unsigned long len;
len = 1UL << (i - 1);
- dbg_printf("fini order %lu len: %lu\n", i, len);
+ dbg_printf("fini order %ld len: %lu\n", i, len);
/* Stop shrink if the resize target changes under us */
if (CMM_LOAD_SHARED(ht->resize_target) > (1UL << (i - 1)))
}
node = clear_flag(next);
}
- assert(!node || !is_bucket(rcu_dereference(node->next)));
+ assert(!node || !is_bucket(CMM_LOAD_SHARED(node->next)));
iter->node = node;
iter->next = next;
}
}
node = clear_flag(next);
}
- assert(!node || !is_bucket(rcu_dereference(node->next)));
+ assert(!node || !is_bucket(CMM_LOAD_SHARED(node->next)));
iter->node = node;
iter->next = next;
}
}
node = clear_flag(next);
}
- assert(!node || !is_bucket(rcu_dereference(node->next)));
+ assert(!node || !is_bucket(CMM_LOAD_SHARED(node->next)));
iter->node = node;
iter->next = next;
}
{
unsigned long size;
- node->reverse_hash = bit_reverse_ulong((unsigned long) hash);
+ node->reverse_hash = bit_reverse_ulong(hash);
size = rcu_dereference(ht->size);
- _cds_lfht_add(ht, NULL, NULL, size, node, NULL, 0);
+ _cds_lfht_add(ht, hash, NULL, NULL, size, node, NULL, 0);
ht_count_add(ht, size, hash);
}
unsigned long size;
struct cds_lfht_iter iter;
- node->reverse_hash = bit_reverse_ulong((unsigned long) hash);
+ node->reverse_hash = bit_reverse_ulong(hash);
size = rcu_dereference(ht->size);
- _cds_lfht_add(ht, match, key, size, node, &iter, 0);
+ _cds_lfht_add(ht, hash, match, key, size, node, &iter, 0);
if (iter.node == node)
ht_count_add(ht, size, hash);
return iter.node;
unsigned long size;
struct cds_lfht_iter iter;
- node->reverse_hash = bit_reverse_ulong((unsigned long) hash);
+ node->reverse_hash = bit_reverse_ulong(hash);
size = rcu_dereference(ht->size);
for (;;) {
- _cds_lfht_add(ht, match, key, size, node, &iter, 0);
+ _cds_lfht_add(ht, hash, match, key, size, node, &iter, 0);
if (iter.node == node) {
ht_count_add(ht, size, hash);
return NULL;
}
}
-int cds_lfht_replace(struct cds_lfht *ht, struct cds_lfht_iter *old_iter,
+int cds_lfht_replace(struct cds_lfht *ht,
+ struct cds_lfht_iter *old_iter,
+ unsigned long hash,
+ cds_lfht_match_fct match,
+ const void *key,
struct cds_lfht_node *new_node)
{
unsigned long size;
+ new_node->reverse_hash = bit_reverse_ulong(hash);
+ if (!old_iter->node)
+ return -ENOENT;
+ if (caa_unlikely(old_iter->node->reverse_hash != new_node->reverse_hash))
+ return -EINVAL;
+ if (caa_unlikely(!match(old_iter->node, key)))
+ return -EINVAL;
size = rcu_dereference(ht->size);
return _cds_lfht_replace(ht, size, old_iter->node, old_iter->next,
new_node);
}
-int cds_lfht_del(struct cds_lfht *ht, struct cds_lfht_iter *iter)
+int cds_lfht_del(struct cds_lfht *ht, struct cds_lfht_node *node)
{
- unsigned long size, hash;
+ unsigned long size;
int ret;
size = rcu_dereference(ht->size);
- ret = _cds_lfht_del(ht, size, iter->node);
+ ret = _cds_lfht_del(ht, size, node);
if (!ret) {
- hash = bit_reverse_ulong(iter->node->reverse_hash);
+ unsigned long hash;
+
+ hash = bit_reverse_ulong(node->reverse_hash);
ht_count_del(ht, size, hash);
}
return ret;
}
+int cds_lfht_is_node_deleted(struct cds_lfht_node *node)
+{
+ return is_removed(CMM_LOAD_SHARED(node->next));
+}
+
static
int cds_lfht_delete_bucket(struct cds_lfht *ht)
{
if (!is_bucket(node))
return -EPERM;
assert(!is_removed(node));
+ assert(!is_removal_owner(node));
} while (!is_end(node));
/*
* size accessed without rcu_dereference because hash table is
* being destroyed.
*/
size = ht->size;
- /* Internal sanity check: all nodes left should be bucket */
+ /* Internal sanity check: all nodes left should be buckets */
for (i = 0; i < size; i++) {
node = bucket_at(ht, i);
dbg_printf("delete bucket: index %lu expected hash %lu hash %lu\n",
*/
int cds_lfht_destroy(struct cds_lfht *ht, pthread_attr_t **attr)
{
- int ret;
+ int ret, was_online;
/* Wait for in-flight resize operations to complete */
_CMM_STORE_SHARED(ht->in_progress_destroy, 1);
cmm_smp_mb(); /* Store destroy before load resize */
+ was_online = ht->flavor->read_ongoing();
+ if (was_online)
+ ht->flavor->thread_offline();
+ /* Calling with RCU read-side held is an error. */
+ if (ht->flavor->read_ongoing()) {
+ ret = -EINVAL;
+ if (was_online)
+ ht->flavor->thread_online();
+ goto end;
+ }
while (uatomic_read(&ht->in_progress_resize))
poll(NULL, 0, 100); /* wait for 100ms */
+ if (was_online)
+ ht->flavor->thread_online();
ret = cds_lfht_delete_bucket(ht);
if (ret)
return ret;
free_split_items_count(ht);
if (attr)
*attr = ht->resize_attr;
+ ret = pthread_mutex_destroy(&ht->resize_mutex);
+ if (ret)
+ ret = -EBUSY;
poison_free(ht);
+end:
return ret;
}
void cds_lfht_resize(struct cds_lfht *ht, unsigned long new_size)
{
+ int was_online;
+
+ was_online = ht->flavor->read_ongoing();
+ if (was_online)
+ ht->flavor->thread_offline();
+ /* Calling with RCU read-side held is an error. */
+ if (ht->flavor->read_ongoing()) {
+ static int print_once;
+
+ if (!CMM_LOAD_SHARED(print_once))
+ fprintf(stderr, "[error] rculfhash: cds_lfht_resize "
+ "called with RCU read-side lock held.\n");
+ CMM_STORE_SHARED(print_once, 1);
+ assert(0);
+ goto end;
+ }
resize_target_update_count(ht, new_size);
CMM_STORE_SHARED(ht->resize_initiated, 1);
- ht->flavor->thread_offline();
pthread_mutex_lock(&ht->resize_mutex);
_do_cds_lfht_resize(ht);
pthread_mutex_unlock(&ht->resize_mutex);
- ht->flavor->thread_online();
+end:
+ if (was_online)
+ ht->flavor->thread_online();
}
static
return;
}
work = malloc(sizeof(*work));
+ if (work == NULL) {
+ dbg_printf("error allocating resize work, bailing out\n");
+ uatomic_dec(&ht->in_progress_resize);
+ return;
+ }
work->ht = ht;
ht->flavor->update_call_rcu(&work->head, do_resize_cb);
CMM_STORE_SHARED(ht->resize_initiated, 1);
projects / urcu.git / blobdiff