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