rculfhash: wait for call_rcu quiescence before destroy

[urcu.git] / rculfhash.c

diff --git a/rculfhash.c b/rculfhash.c
index 720fd0151e20ce70b31b4d5a4012f4ae2d123592..a3223f37f046021cb6c3c0119e107e2e4dde4fd0 100644 (file)
--- a/rculfhash.c
+++ b/rculfhash.c
@@ -67,6 +67,7 @@ struct rcu_ht {
        ht_compare_fct compare_fct;
        unsigned long hash_seed;
        pthread_mutex_t resize_mutex;   /* resize mutex: add/del mutex */
+       unsigned int in_progress_resize;
        void (*ht_call_rcu)(struct rcu_head *head,
                      void (*func)(struct rcu_head *head));
 };
@@ -217,16 +218,14 @@ void _ht_gc_bucket(struct rcu_ht_node *dummy, struct rcu_ht_node *node)
                /* We can always skip the dummy node initially */
                iter = rcu_dereference(iter_prev->next);
                assert(iter_prev->reverse_hash <= node->reverse_hash);
-               if (unlikely(!iter))
-                       return;
                for (;;) {
+                       if (unlikely(!iter))
+                               return;
                        if (clear_flag(iter)->reverse_hash > node->reverse_hash)
                                return;
                        next = rcu_dereference(clear_flag(iter)->next);
                        if (is_removed(next))
                                break;
-                       if (unlikely(!next))
-                               return;
                        iter_prev = iter;
                        iter = next;
                }
@@ -236,13 +235,17 @@ void _ht_gc_bucket(struct rcu_ht_node *dummy, struct rcu_ht_node *node)
 }
 
 static
-int _ht_add(struct rcu_ht *ht, struct rcu_table *t, struct rcu_ht_node *node,
-            int unique)
+struct rcu_ht_node *_ht_add(struct rcu_ht *ht, struct rcu_table *t,
+                           struct rcu_ht_node *node, int unique)
 {
        struct rcu_ht_node *iter_prev, *dummy, *iter, *next;
+       unsigned long hash;
 
-       if (!t->size)
-               return 0;
+       if (!t->size) {
+               assert(node->dummy);
+               return node;    /* Initial first add (head) */
+       }
+       hash = bit_reverse_ulong(node->reverse_hash);
        for (;;) {
                uint32_t chain_len = 0;
 
@@ -250,7 +253,7 @@ int _ht_add(struct rcu_ht *ht, struct rcu_table *t, struct rcu_ht_node *node,
                 * iter_prev points to the non-removed node prior to the
                 * insert location.
                 */
-               iter_prev = rcu_dereference(t->tbl[node->hash & (t->size - 1)]);
+               iter_prev = rcu_dereference(t->tbl[hash & (t->size - 1)]);
                /* We can always skip the dummy node initially */
                iter = rcu_dereference(iter_prev->next);
                assert(iter_prev->reverse_hash <= node->reverse_hash);
@@ -261,7 +264,13 @@ int _ht_add(struct rcu_ht *ht, struct rcu_table *t, struct rcu_ht_node *node,
                                goto insert;
                        next = rcu_dereference(clear_flag(iter)->next);
                        if (is_removed(next))
-                               goto gc;
+                               goto gc_node;
+                       if (unique
+                           && !clear_flag(iter)->dummy
+                           && !ht->compare_fct(node->key, node->key_len,
+                                               clear_flag(iter)->key,
+                                               clear_flag(iter)->key_len))
+                               return clear_flag(iter);
                        /* Only account for identical reverse hash once */
                        if (iter_prev->reverse_hash != clear_flag(iter)->reverse_hash)
                                check_resize(ht, t, ++chain_len);
@@ -278,17 +287,16 @@ int _ht_add(struct rcu_ht *ht, struct rcu_table *t, struct rcu_ht_node *node,
                        continue;       /* retry */
                else
                        goto gc_end;
-       gc:
-               /* Garbage collect logically removed nodes in the bucket */
-               dummy = rcu_dereference(t->tbl[node->hash & (t->size - 1)]);
-               _ht_gc_bucket(dummy, node);
+       gc_node:
+               assert(!is_removed(iter));
+               (void) uatomic_cmpxchg(&iter_prev->next, iter, clear_flag(next));
                /* retry */
        }
 gc_end:
        /* Garbage collect logically removed nodes in the bucket */
-       dummy = rcu_dereference(t->tbl[node->hash & (t->size - 1)]);
+       dummy = rcu_dereference(t->tbl[hash & (t->size - 1)]);
        _ht_gc_bucket(dummy, node);
-       return 0;
+       return node;
 }
 
 static
@@ -296,6 +304,7 @@ int _ht_remove(struct rcu_ht *ht, struct rcu_table *t, struct rcu_ht_node *node)
 {
        struct rcu_ht_node *dummy, *next, *old;
        int flagged = 0;
+       unsigned long hash;
 
        /* logically delete the node */
        old = rcu_dereference(node->next);
@@ -316,7 +325,8 @@ int _ht_remove(struct rcu_ht *ht, struct rcu_table *t, struct rcu_ht_node *node)
         * the node, and remove it (along with any other logically removed node)
         * if found.
         */
-       dummy = rcu_dereference(t->tbl[node->hash & (t->size - 1)]);
+       hash = bit_reverse_ulong(node->reverse_hash);
+       dummy = rcu_dereference(t->tbl[hash & (t->size - 1)]);
        _ht_gc_bucket(dummy, node);
 end:
        /*
@@ -343,7 +353,6 @@ void init_table(struct rcu_ht *ht, struct rcu_table *t,
                        t->size = i;
                t->tbl[i] = calloc(1, sizeof(struct rcu_ht_node));
                t->tbl[i]->dummy = 1;
-               t->tbl[i]->hash = i;
                t->tbl[i]->reverse_hash = bit_reverse_ulong(i);
                (void) _ht_add(ht, t, t->tbl[i], 0);
        }
@@ -365,6 +374,7 @@ struct rcu_ht *ht_new(ht_hash_fct hash_fct,
        ht->compare_fct = compare_fct;
        ht->hash_seed = hash_seed;
        ht->ht_call_rcu = ht_call_rcu;
+       ht->in_progress_resize = 0;
        /* this mutex should not nest in read-side C.S. */
        pthread_mutex_init(&ht->resize_mutex, NULL);
        ht->t = calloc(1, sizeof(struct rcu_table)
@@ -394,9 +404,9 @@ struct rcu_ht_node *ht_lookup(struct rcu_ht *ht, void *key, size_t key_len)
                        node = NULL;
                        break;
                }
-               if (!ht->compare_fct(node->key, node->key_len, key, key_len)) {
-                       if (likely(!is_removed(rcu_dereference(node->next)))
-                           && likely(!node->dummy))
+               if (likely(!is_removed(rcu_dereference(node->next)))
+                   && !node->dummy
+                   && likely(!ht->compare_fct(node->key, node->key_len, key, key_len))) {
                                break;
                }
                node = clear_flag(rcu_dereference(node->next));
@@ -408,20 +418,22 @@ struct rcu_ht_node *ht_lookup(struct rcu_ht *ht, void *key, size_t key_len)
 void ht_add(struct rcu_ht *ht, struct rcu_ht_node *node)
 {
        struct rcu_table *t;
+       unsigned long hash;
 
-       node->hash = ht->hash_fct(node->key, node->key_len, ht->hash_seed);
-       node->reverse_hash = bit_reverse_ulong((unsigned long) node->hash);
+       hash = ht->hash_fct(node->key, node->key_len, ht->hash_seed);
+       node->reverse_hash = bit_reverse_ulong((unsigned long) hash);
 
        t = rcu_dereference(ht->t);
        (void) _ht_add(ht, t, node, 0);
 }
 
-int ht_add_unique(struct rcu_ht *ht, struct rcu_ht_node *node)
+struct rcu_ht_node *ht_add_unique(struct rcu_ht *ht, struct rcu_ht_node *node)
 {
        struct rcu_table *t;
+       unsigned long hash;
 
-       node->hash = ht->hash_fct(node->key, node->key_len, ht->hash_seed);
-       node->reverse_hash = bit_reverse_ulong((unsigned long) node->hash);
+       hash = ht->hash_fct(node->key, node->key_len, ht->hash_seed);
+       node->reverse_hash = bit_reverse_ulong((unsigned long) hash);
 
        t = rcu_dereference(ht->t);
        return _ht_add(ht, t, node, 1);
@@ -467,6 +479,9 @@ int ht_destroy(struct rcu_ht *ht)
 {
        int ret;
 
+       /* Wait for in-flight resize operations to complete */
+       while (uatomic_read(&ht->in_progress_resize))
+               poll(NULL, 0, 100);     /* wait for 100ms */
        ret = ht_delete_dummy(ht);
        if (ret)
                return ret;
@@ -566,6 +581,8 @@ void do_resize_cb(struct rcu_head *head)
        _do_ht_resize(ht);
        pthread_mutex_unlock(&ht->resize_mutex);
        free(work);
+       cmm_smp_mb();   /* finish resize before decrement */
+       uatomic_dec(&ht->in_progress_resize);
 }
 
 static
@@ -576,6 +593,8 @@ void ht_resize_lazy(struct rcu_ht *ht, struct rcu_table *t, int growth)
 
        target_size = resize_target_update(t, growth);
        if (!CMM_LOAD_SHARED(t->resize_initiated) && t->size < target_size) {
+               uatomic_inc(&ht->in_progress_resize);
+               cmm_smp_mb();   /* increment resize count before calling it */
                work = malloc(sizeof(*work));
                work->ht = ht;
                ht->ht_call_rcu(&work->head, do_resize_cb);