rculfhash: make add/removal help expand

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>

diff --git a/rculfhash.c b/rculfhash.c
index 8dd5a7327ad81be4da1b3dddace433b4fd8aadc4..ce09a0f93261d9a5c2a76c0a530e7cca6118fe5f 100644 (file)
--- a/rculfhash.c
+++ b/rculfhash.c
@@ -167,12 +167,7 @@
 #define CHAIN_LEN_RESIZE_THRESHOLD     3
 
 /*
- * Define the minimum table size. Protects against hash table resize overload
- * when too many entries are added quickly before the resize can complete.
- * This is especially the case if the table could be shrinked to a size of 1.
- * TODO: we might want to make the add/remove operations help the resize to
- * add or remove dummy nodes when a resize is ongoing to ensure upper-bound on
- * chain length.
+ * Define the minimum table size.
  */
 #define MIN_TABLE_SIZE                 128
 
@@ -246,6 +241,12 @@ struct rcu_resize_work {
        struct cds_lfht *ht;
 };
 
+static
+struct cds_lfht_node *_cds_lfht_add(struct cds_lfht *ht,
+                               unsigned long size,
+                               struct cds_lfht_node *node,
+                               int unique, int dummy);
+
 /*
  * Algorithm to reverse bits in a word by lookup table, extended to
  * 64-bit words.
@@ -678,7 +679,7 @@ void cds_lfht_free_level(struct rcu_head *head)
  * Remove all logically deleted nodes from a bucket up to a certain node key.
  */
 static
-void _cds_lfht_gc_bucket(struct cds_lfht_node *dummy, struct cds_lfht_node *node)
+int _cds_lfht_gc_bucket(struct cds_lfht_node *dummy, struct cds_lfht_node *node)
 {
        struct cds_lfht_node *iter_prev, *iter, *next, *new_next;
 
@@ -690,6 +691,15 @@ void _cds_lfht_gc_bucket(struct cds_lfht_node *dummy, struct cds_lfht_node *node
                iter_prev = dummy;
                /* We can always skip the dummy node initially */
                iter = rcu_dereference(iter_prev->p.next);
+               if (unlikely(iter == NULL)) {
+                       /*
+                        * We are executing concurrently with a hash table
+                        * expand, so we see a dummy node with NULL next value.
+                        * Help expand by linking this node into the list and
+                        * retry.
+                        */
+                       return 1;
+               }
                assert(iter_prev->p.reverse_hash <= node->p.reverse_hash);
                /*
                 * We should never be called with dummy (start of chain)
@@ -700,9 +710,9 @@ void _cds_lfht_gc_bucket(struct cds_lfht_node *dummy, struct cds_lfht_node *node
                assert(dummy != node);
                for (;;) {
                        if (unlikely(!clear_flag(iter)))
-                               return;
+                               return 0;
                        if (likely(clear_flag(iter)->p.reverse_hash > node->p.reverse_hash))
-                               return;
+                               return 0;
                        next = rcu_dereference(clear_flag(iter)->p.next);
                        if (likely(is_removed(next)))
                                break;
@@ -716,6 +726,7 @@ void _cds_lfht_gc_bucket(struct cds_lfht_node *dummy, struct cds_lfht_node *node
                        new_next = clear_flag(next);
                (void) uatomic_cmpxchg(&iter_prev->p.next, iter, new_next);
        }
+       return 0;
 }
 
 static
@@ -750,8 +761,25 @@ struct cds_lfht_node *_cds_lfht_add(struct cds_lfht *ht,
                iter_prev = (struct cds_lfht_node *) lookup;
                /* We can always skip the dummy node initially */
                iter = rcu_dereference(iter_prev->p.next);
+               if (unlikely(iter == NULL)) {
+                       /*
+                        * We are executing concurrently with a hash table
+                        * expand, so we see a dummy node with NULL next value.
+                        * Help expand by linking this node into the list and
+                        * retry.
+                        */
+                       (void) _cds_lfht_add(ht, size >> 1, iter_prev, 0, 1);
+                       continue;       /* retry */
+               }
                assert(iter_prev->p.reverse_hash <= node->p.reverse_hash);
                for (;;) {
+                       /*
+                        * When adding a dummy node, we allow concurrent
+                        * add/removal to help. If we find the dummy node in
+                        * place, skip its insertion.
+                        */
+                       if (unlikely(dummy && clear_flag(iter) == node))
+                               return node;
                        if (unlikely(!clear_flag(iter)))
                                goto insert;
                        if (likely(clear_flag(iter)->p.reverse_hash > node->p.reverse_hash))
@@ -805,7 +833,11 @@ gc_end:
        order = get_count_order_ulong(index + 1);
        lookup = &ht->t.tbl[order]->nodes[index & (!order ? 0 : ((1UL << (order - 1)) - 1))];
        dummy_node = (struct cds_lfht_node *) lookup;
-       _cds_lfht_gc_bucket(dummy_node, node);
+       if (_cds_lfht_gc_bucket(dummy_node, node)) {
+               /* Help expand */
+               (void) _cds_lfht_add(ht, size >> 1, dummy_node, 0, 1);
+               goto gc_end;    /* retry */
+       }
        return node;
 }
 
@@ -843,13 +875,18 @@ int _cds_lfht_remove(struct cds_lfht *ht, unsigned long size,
         * the node, and remove it (along with any other logically removed node)
         * if found.
         */
+gc_retry:
        hash = bit_reverse_ulong(node->p.reverse_hash);
        assert(size > 0);
        index = hash & (size - 1);
        order = get_count_order_ulong(index + 1);
        lookup = &ht->t.tbl[order]->nodes[index & (!order ? 0 : ((1UL << (order - 1)) - 1))];
        dummy = (struct cds_lfht_node *) lookup;
-       _cds_lfht_gc_bucket(dummy, node);
+       if (_cds_lfht_gc_bucket(dummy, node)) {
+               /* Help expand */
+               (void) _cds_lfht_add(ht, size >> 1, dummy, 0, 1);
+               goto gc_retry;  /* retry */
+       }
 end:
        /*
         * Only the flagging action indicated that we (and no other)
@@ -928,18 +965,21 @@ void init_table(struct cds_lfht *ht,
                /* Set all dummy nodes reverse hash values for a level */
                init_table_hash(ht, i, len);
 
+               /*
+                * Update table size. At this point, concurrent add/remove see
+                * dummy nodes with correctly initialized reverse hash value,
+                * but with NULL next pointers. If they do, they can help us
+                * link the dummy nodes into the list and retry.
+                */
+               cmm_smp_wmb();  /* populate data before RCU size */
+               CMM_STORE_SHARED(ht->t.size, !i ? 1 : (1UL << i));
+
                /*
                 * Link all dummy nodes into the table. Concurrent
                 * add/remove are helping us.
                 */
                init_table_link(ht, i, len);
 
-               /*
-                * Update table size (after init for now, because no
-                * concurrent updater help (TODO)).
-                */
-               cmm_smp_wmb();  /* populate data before RCU size */
-               CMM_STORE_SHARED(ht->t.size, !i ? 1 : (1UL << i));
                dbg_printf("init new size: %lu\n", !i ? 1 : (1UL << i));
                if (CMM_LOAD_SHARED(ht->in_progress_destroy))
                        break;