urcu-ht.c

   1
   2 /*
   3  * TODO: keys are currently assumed <= sizeof(void *). Key target never freed.
   4  */
   5
   6 #define _LGPL_SOURCE
   7 #include <stdlib.h>
   8 #include <errno.h>
   9 #include <assert.h>
  10 #include <stdio.h>
  11
  12 #include <urcu.h>
  13 #include <urcu-defer.h>
  14 #include <arch.h>
  15 #include <arch_atomic.h>
  16 #include <compiler.h>
  17 #include <urcu/jhash.h>
  18 #include <urcu-ht.h>
  19
  20 struct rcu_ht_node;
  21
  22 struct rcu_ht_node {
  23         struct rcu_ht_node *next;
  24         void *key;
  25         void *data;
  26         int stolen;
  27 };
  28
  29 struct rcu_ht {
  30         struct rcu_ht_node **tbl;
  31         ht_hash_fct hash_fct;
  32         void (*free_fct)(void *data);   /* fct to free data */
  33         uint32_t keylen;
  34         uint32_t hashseed;
  35         struct ht_size {
  36                 unsigned long add;
  37                 unsigned long lookup;
  38         } size;
  39 };
  40
  41 struct rcu_ht *ht_new(ht_hash_fct hash_fct, void (*free_fct)(void *data),
  42                       unsigned long init_size, uint32_t keylen,
  43                       uint32_t hashseed)
  44 {
  45         struct rcu_ht *ht;
  46
  47         ht = calloc(1, sizeof(struct rcu_ht));
  48         ht->hash_fct = hash_fct;
  49         ht->free_fct = free_fct;
  50         ht->size.add = init_size;
  51         ht->size.lookup = init_size;
  52         ht->keylen = keylen;
  53         ht->hashseed = hashseed;
  54         ht->tbl = calloc(init_size, sizeof(struct rcu_ht_node *));
  55         return ht;
  56 }
  57
  58 void *ht_lookup(struct rcu_ht *ht, void *key)
  59 {
  60         unsigned long hash;
  61         struct rcu_ht_node *node;
  62         void *ret;
  63
  64         hash = ht->hash_fct(key, ht->keylen, ht->hashseed) % ht->size.lookup;
  65
  66         rcu_read_lock();
  67         node = rcu_dereference(ht->tbl[hash]);
  68         for (;;) {
  69                 if (likely(!node)) {
  70                         ret = NULL;
  71                         break;
  72                 }
  73                 if (node->key == key) {
  74                         ret = node->data;
  75                         break;
  76                 }
  77                 node = rcu_dereference(node->next);
  78         }
  79         rcu_read_unlock();
  80
  81         return ret;
  82 }
  83
  84 /*
  85  * Will re-try until either:
  86  * - The key is already there (-EEXIST)
  87  * - We successfully add the key at the head of a table bucket.
  88  */
  89 int ht_add(struct rcu_ht *ht, void *key, void *data)
  90 {
  91         struct rcu_ht_node *node, *old_head, *new_head;
  92         unsigned long hash;
  93         int ret = 0;
  94
  95         new_head = calloc(1, sizeof(struct rcu_ht_node));
  96         new_head->key = key;
  97         new_head->data = data;
  98         new_head->stolen = 0;
  99         /* here comes the fun and tricky part.
 100          * Add at the beginning with a cmpxchg.
 101          * Hold a read lock between the moment the first element is read
 102          * and the nodes traversal (to find duplicates). This ensures
 103          * the head pointer has not been reclaimed when cmpxchg is done.
 104          * Always adding at the head ensures that we would have to
 105          * re-try if a new item has been added concurrently. So we ensure that
 106          * we never add duplicates. */
 107 retry:
 108         rcu_read_lock();
 109
 110         hash = ht->hash_fct(key, ht->keylen, ht->hashseed) % ht->size.add;
 111
 112         old_head = node = rcu_dereference(ht->tbl[hash]);
 113         for (;;) {
 114                 if (likely(!node)) {
 115                         break;
 116                 }
 117                 if (node->key == key) {
 118                         ret = -EEXIST;
 119                         goto end;
 120                 }
 121                 node = rcu_dereference(node->next);
 122         }
 123         new_head->next = old_head;
 124         if (rcu_cmpxchg_pointer(&ht->tbl[hash], old_head, new_head) != old_head)
 125                 goto restart;
 126 end:
 127         rcu_read_unlock();
 128
 129         return ret;
 130
 131         /* restart loop, release and re-take the read lock to be kind to GP */
 132 restart:
 133         rcu_read_unlock();
 134         goto retry;
 135 }
 136
 137 /*
 138  * Restart until we successfully remove the entry, or no entry is left
 139  * ((void *)(unsigned long)-ENOENT).
 140  * Deal with concurrent stealers by doing an extra verification pass to check
 141  * that no element in the list are still pointing to the element stolen.
 142  * This could happen if two concurrent steal for consecutive objects are
 143  * executed. A pointer to an object being stolen could be saved by the
 144  * concurrent stealer for the previous object.
 145  * Also, given that in this precise scenario, another stealer can also want to
 146  * delete the doubly-referenced object; use a "stolen" flag to let only one
 147  * stealer delete the object.
 148  */
 149 void *ht_steal(struct rcu_ht *ht, void *key)
 150 {
 151         struct rcu_ht_node **prev, *node, *del_node = NULL;
 152         unsigned long hash;
 153         void *data;
 154
 155 retry:
 156         rcu_read_lock();
 157
 158         hash = ht->hash_fct(key, ht->keylen, ht->hashseed) % ht->size.lookup;
 159
 160         prev = &ht->tbl[hash];
 161         node = rcu_dereference(*prev);
 162         for (;;) {
 163                 if (likely(!node)) {
 164                         if (del_node) {
 165                                 goto end;
 166                         } else {
 167                                 goto error;
 168                         }
 169                 }
 170                 if (node->key == key) {
 171                         break;
 172                 }
 173                 prev = &node->next;
 174                 node = rcu_dereference(*prev);
 175         }
 176
 177         if (!del_node) {
 178                 /*
 179                  * Another concurrent thread stole it ? If so, let it deal with
 180                  * this.
 181                  */
 182                 if (cmpxchg(&node->stolen, 0, 1) != 0)
 183                         goto error;
 184         }
 185
 186         /* Found it ! pointer to object is in "prev" */
 187         if (rcu_cmpxchg_pointer(prev, node, node->next) == node)
 188                 del_node = node;
 189         goto restart;
 190
 191 end:
 192         /*
 193          * From that point, we own node. Note that there can still be concurrent
 194          * RCU readers using it. We can free it outside of read lock after a GP.
 195          */
 196         rcu_read_unlock();
 197
 198         data = del_node->data;
 199         call_rcu(free, del_node);
 200         return data;
 201
 202 error:
 203         data = (void *)(unsigned long)-ENOENT;
 204         rcu_read_unlock();
 205         return data;
 206
 207         /* restart loop, release and re-take the read lock to be kind to GP */
 208 restart:
 209         rcu_read_unlock();
 210         goto retry;
 211 }
 212
 213 int ht_delete(struct rcu_ht *ht, void *key)
 214 {
 215         void *data;
 216
 217         data = ht_steal(ht, key);
 218         if (data && data != (void *)(unsigned long)-ENOENT) {
 219                 if (ht->free_fct)
 220                         call_rcu(ht->free_fct, data);
 221                 return 0;
 222         } else {
 223                 return -ENOENT;
 224         }
 225 }
 226
 227 /* Delete all old elements. Allow concurrent writer accesses. */
 228 int ht_delete_all(struct rcu_ht *ht)
 229 {
 230         unsigned long i;
 231         struct rcu_ht_node **prev, *node, *inext;
 232         int cnt = 0;
 233
 234         for (i = 0; i < ht->size.lookup; i++) {
 235                 rcu_read_lock();
 236                 prev = &ht->tbl[i];
 237                 /*
 238                  * Cut the head. After that, we own the first element.
 239                  */
 240                 node = rcu_xchg_pointer(prev, NULL);
 241                 if (!node) {
 242                         rcu_read_unlock();
 243                         continue;
 244                 }
 245                 /*
 246                  * We manage a list shared with concurrent writers and readers.
 247                  * Note that a concurrent add may or may not be deleted by us,
 248                  * depending if it arrives before or after the head is cut.
 249                  * "node" points to our first node. Remove first elements
 250                  * iteratively.
 251                  */
 252                 for (;;) {
 253                         inext = NULL;
 254                         prev = &node->next;
 255                         if (prev)
 256                                 inext = rcu_xchg_pointer(prev, NULL);
 257                         /*
 258                          * "node" is the first element of the list we have cut.
 259                          * We therefore own it, no concurrent writer may delete
 260                          * it. There can only be concurrent lookups. Concurrent
 261                          * add can only be done on a bucket head, but we've cut
 262                          * it already. inext is also owned by us, because we
 263                          * have exchanged it for "NULL". It will therefore be
 264                          * safe to use it after a G.P.
 265                          */
 266                         rcu_read_unlock();
 267                         if (node->data)
 268                                 call_rcu(ht->free_fct, node->data);
 269                         call_rcu(free, node);
 270                         cnt++;
 271                         if (likely(!inext))
 272                                 break;
 273                         rcu_read_lock();
 274                         node = inext;
 275                 }
 276         }
 277         return cnt;
 278 }
 279
 280 /*
 281  * Should only be called when no more concurrent readers nor writers can
 282  * possibly access the table.
 283  */
 284 int ht_destroy(struct rcu_ht *ht)
 285 {
 286         int ret;
 287
 288         ret = ht_delete_all(ht);
 289         free(ht->tbl);
 290         free(ht);
 291         return ret;
 292 }
 293
 294 /*
 295  * Expects keys <= than pointer size to be encoded in the pointer itself.
 296  */
 297 uint32_t ht_jhash(void *key, uint32_t length, uint32_t initval)
 298 {
 299         uint32_t ret;
 300         void *vkey;
 301
 302         if (length <= sizeof(void *))
 303                 vkey = &key;
 304         else
 305                 vkey = key;
 306         ret = jhash(vkey, length, initval);
 307         return ret;
 308 }