pd.if.org Git - nbds/blob - struct/skiplist.c

   1 /*
   2  * Written by Josh Dybnis and released to the public domain, as explained at
   3  * http://creativecommons.org/licenses/publicdomain
   4  *
   5  * lock-free skiplist
   6  */
   7 #include <stdio.h>
   8 #include <string.h>
   9
  10 #include "common.h"
  11 #include "struct.h"
  12 #include "mem.h"
  13
  14 #define MAX_LEVEL 3
  15
  16 typedef struct node {
  17     uint64_t key;
  18     uint64_t value;
  19     int top_level;
  20     struct node *next[];
  21 } node_t;
  22
  23 typedef struct skiplist {
  24     node_t *head;
  25     node_t *last;
  26     int top_level;
  27 } skiplist_t;
  28
  29 static int random_level (int r) {
  30     if (r&1)
  31         return 0;
  32     int n = __builtin_ctz(r);
  33     if (n < MAX_LEVEL)
  34         return n;
  35     return MAX_LEVEL;
  36 }
  37
  38 node_t *node_alloc (int top_level, uint64_t key, uint64_t value) {
  39     assert(top_level >= 0 && top_level <= MAX_LEVEL);
  40     size_t sz = sizeof(node_t) + (top_level + 1) * sizeof(node_t *);
  41     node_t *item = (node_t *)nbd_malloc(sz);
  42     memset(item, 0, sz);
  43     item->key   = key;
  44     item->value = value;
  45     item->top_level = top_level;
  46     return item;
  47 }
  48
  49 skiplist_t *skiplist_alloc (void) {
  50     skiplist_t *skiplist = (skiplist_t *)nbd_malloc(sizeof(skiplist_t));
  51     skiplist->head = node_alloc(MAX_LEVEL, 0, 0);
  52     skiplist->last = node_alloc(MAX_LEVEL, (uint64_t)-1, 0);
  53     for (int level = 0; level <= MAX_LEVEL; ++level) {
  54         skiplist->head->next[level] = skiplist->last;
  55     }
  56     return skiplist;
  57 }
  58
  59 static node_t *find_preds (node_t **preds, skiplist_t *skiplist, uint64_t key, int help_remove) {
  60     node_t *pred = skiplist->head;
  61     node_t *item = NULL;
  62     TRACE("s3", "find_preds: searching for key %p in skiplist (head is %p)", key, pred);
  63 #ifndef NDEBUG
  64     int count = 0;
  65 #endif
  66
  67     // Traverse the levels of the skiplist from the top level to the bottom
  68     for (int level = MAX_LEVEL; level >= 0; --level) {
  69         TRACE("s3", "find_preds: level %llu", level, 0);
  70         item = pred->next[level];
  71         if (IS_TAGGED(item)) {
  72             TRACE("s3", "find_preds: pred %p is marked for removal (item %p); retry", pred, item);
  73             return find_preds(preds, skiplist, key, help_remove); // retry
  74         }
  75         do {
  76             node_t *next = item->next[level];
  77             TRACE("s3", "find_preds: visiting item %p (next %p)", item, next);
  78             TRACE("s3", "find_preds: key %p", item->key, 0);
  79
  80             // Marked items are logically removed, but not fully unlinked yet.
  81             while (EXPECT_FALSE(IS_TAGGED(next))) {
  82
  83                 // Skip over partially removed items.
  84                 if (!help_remove) {
  85                     item = (node_t *)STRIP_TAG(item->next);
  86                     next = item->next[level];
  87                     continue;
  88                 }
  89
  90                 // Unlink partially removed items.
  91                 node_t *other;
  92                 if ((other = SYNC_CAS(&pred->next[level], item, STRIP_TAG(next))) == item) {
  93                     item = (node_t *)STRIP_TAG(next);
  94                     next = item->next[level];
  95                     TRACE("s3", "find_preds: unlinked item %p from pred %p", item, pred);
  96                     TRACE("s3", "find_preds: now item is %p next is %p", item, next);
  97
  98                     // The thread that completes the unlink should free the memory.
  99                     if (level == 0) { nbd_defer_free(other); }
 100                 } else {
 101                     TRACE("s3", "find_preds: lost race to unlink item from pred %p; its link changed to %p",pred,other);
 102                     if (IS_TAGGED(other))
 103                         return find_preds(preds, skiplist, key, help_remove); // retry
 104                     item = other;
 105                     next = item->next[level];
 106                 }
 107             }
 108
 109             // If we reached the key (or passed where it should be), we found a pred. Save it and continue down.
 110             if (item->key >= key) {
 111                 TRACE("s3", "find_preds: found pred %p item %p", pred, item);
 112                 if (preds != NULL) {
 113                     preds[level] = pred;
 114                 }
 115                 break;
 116             }
 117
 118             assert(count++ < 18);
 119             pred = item;
 120             item = next;
 121
 122         } while (1);
 123     }
 124     return item;
 125 }
 126
 127 // Fast find that does not help unlink partially removed nodes and does not return the node's predecessors.
 128 uint64_t skiplist_lookup (skiplist_t *skiplist, uint64_t key) {
 129     TRACE("s3", "skiplist_lookup: searching for key %p in skiplist %p", key, skiplist);
 130     node_t *item = find_preds(NULL, skiplist, key, FALSE);
 131
 132     // If we found an <item> matching the <key> return its value.
 133     return (item->key == key) ? item->value : DOES_NOT_EXIST;
 134 }
 135
 136 // Insert the <key> if it doesn't already exist in the <skiplist>
 137 uint64_t skiplist_add_i (int n, skiplist_t *skiplist, uint64_t key, uint64_t value) {
 138     TRACE("s3", "skiplist_add: inserting key %p value %p", key, value);
 139     node_t *preds[MAX_LEVEL+1];
 140     node_t *item = NULL;
 141     do {
 142         node_t *next = find_preds(preds, skiplist, key, TRUE);
 143
 144         // If a node matching <key> already exists in the skiplist, return its value.
 145         if (next->key == key) {
 146             TRACE("s3", "skiplist_add: there is already an item %p (value %p) with the same key", next, next->value);
 147             if (EXPECT_FALSE(item != NULL)) { nbd_free(item); }
 148             return next->value;
 149         }
 150
 151         // First insert <item> into the bottom level.
 152         if (EXPECT_TRUE(item == NULL)) { item = node_alloc(random_level(n), key, value); }
 153         TRACE("s3", "skiplist_add: attempting to insert item between %p and %p", preds[0], next);
 154         item->next[0] = next;
 155         for (int level = 1; level <= item->top_level; ++level) {
 156             item->next[level] = preds[level]->next[level];
 157         }
 158         node_t *other = SYNC_CAS(&preds[0]->next[0], next, item);
 159         if (other == next) {
 160             TRACE("s3", "skiplist_add: successfully inserted item %p at level 0", item, 0);
 161             break; // success
 162         }
 163         TRACE("s3", "skiplist_add: failed to change pred's link: expected %p found %p", next, other);
 164
 165     } while (1);
 166
 167     // Insert <item> into the skiplist from the bottom level up.
 168     for (int level = 1; level <= item->top_level; ++level) {
 169         do {
 170             node_t *pred = preds[level];
 171             node_t *next = pred->next[level];
 172             while (EXPECT_FALSE(IS_TAGGED(next) || next->key < key)) {
 173                 find_preds(preds, skiplist, key, TRUE);
 174                 pred = preds[level];
 175                 next = pred->next[level];
 176             }
 177
 178             do {
 179                 // There in no need to continue linking in the item if another thread removed it.
 180                 node_t *old_next = ((volatile node_t *)item)->next[level];
 181                 if (IS_TAGGED(old_next))
 182                     return DOES_NOT_EXIST; // success
 183
 184                 // Use a CAS so we to not inadvertantly remove a mark another thread placed on the item.
 185                 if (next == old_next || SYNC_CAS(&item->next[level], old_next, next) == old_next)
 186                     break;
 187             } while (1);
 188
 189             TRACE("s3", "skiplist_add: attempting to insert item between %p and %p", pred, next);
 190             node_t *other = SYNC_CAS(&pred->next[level], next, item);
 191             if (other == next) {
 192                 TRACE("s3", "skiplist_add: successfully inserted item %p at level %llu", item, level);
 193                 break; // success
 194             }
 195             TRACE("s3", "skiplist_add: failed to change pred's link: expected %p found %p", next, other);
 196
 197         } while (1);
 198     }
 199     return value;
 200 }
 201
 202 uint64_t skiplist_remove (skiplist_t *skiplist, uint64_t key) {
 203     TRACE("s3", "skiplist_remove: removing item with key %p from skiplist %p", key, skiplist);
 204     node_t *preds[MAX_LEVEL+1];
 205     node_t *item = find_preds(preds, skiplist, key, TRUE);
 206     if (item->key != key) {
 207         TRACE("s3", "skiplist_remove: remove failed, an item with a matching key does not exist in the skiplist", 0, 0);
 208         return DOES_NOT_EXIST;
 209     }
 210
 211     // Mark <item> removed at each level of the skiplist from the top down. This must be atomic. If multiple threads
 212     // try to remove the same item only one of them should succeed. Marking the bottom level establishes which of
 213     // them succeeds.
 214     for (int level = item->top_level; level >= 0; --level) {
 215         if (EXPECT_FALSE(IS_TAGGED(item->next[level]))) {
 216             TRACE("s3", "skiplist_remove: %p is already marked for removal by another thread", item, 0);
 217             if (level == 0)
 218                 return DOES_NOT_EXIST;
 219             continue;
 220         }
 221         node_t *next = SYNC_FETCH_AND_OR(&item->next[level], TAG);
 222         if (EXPECT_FALSE(IS_TAGGED(next))) {
 223             TRACE("s3", "skiplist_remove: lost race -- %p is already marked for removal by another thread", item, 0);
 224             if (level == 0)
 225                 return DOES_NOT_EXIST;
 226             continue;
 227         }
 228     }
 229
 230     uint64_t value = item->value;
 231
 232     // Unlink <item> from the top down.
 233     int level = item->top_level;
 234     while (level >= 0) {
 235         node_t *pred = preds[level];
 236         node_t *next = item->next[level];
 237         TRACE("s3", "skiplist_remove: link item's pred %p to it's successor %p", pred, STRIP_TAG(next));
 238         node_t *other = NULL;
 239         if ((other = SYNC_CAS(&pred->next[level], item, STRIP_TAG(next))) != item) {
 240             TRACE("s3", "skiplist_remove: unlink failed; pred's link changed from %p to %p", item, other);
 241             // By marking the item earlier, we logically removed it. It is safe to leave the item partially
 242             // unlinked. Another thread will finish physically removing it from the skiplist.
 243             return value;
 244         }
 245         --level;
 246     }
 247
 248     // The thread that completes the unlink should free the memory.
 249     nbd_defer_free(item);
 250     return value;
 251 }
 252
 253 void skiplist_print (skiplist_t *skiplist) {
 254     for (int level = MAX_LEVEL; level >= 0; --level) {
 255         printf("(%d) ", level);
 256         node_t *item = skiplist->head;
 257         while (item) {
 258             node_t *next = item->next[level];
 259             printf("%s%p:0x%llx ", IS_TAGGED(next) ? "*" : "", item, item->key);
 260             item = (node_t *)STRIP_TAG(next);
 261         }
 262         printf("\n");
 263         fflush(stdout);
 264     }
 265
 266     node_t *item = skiplist->head;
 267     while (item) {
 268         assert(item->top_level <= MAX_LEVEL);
 269         int is_marked = IS_TAGGED(item->next[0]);
 270         printf("%s%p:0x%llx (%d", is_marked ? "*" : "", item, item->key, item->top_level);
 271         for (int i = 1; i <= item->top_level; ++i) {
 272             node_t *next = (node_t *)STRIP_TAG(item->next[i]);
 273             printf(" %p:0x%llx", item->next[i], next ? next->key : 0);
 274         }
 275         printf(")\n");
 276         fflush(stdout);
 277         item = (node_t *)STRIP_TAG(item->next[0]);
 278     }
 279 }
 280
 281 #ifdef MAKE_skiplist_test
 282 #include <errno.h>
 283 #include <pthread.h>
 284 #include <sys/time.h>
 285
 286 #include "runtime.h"
 287
 288 #define NUM_ITERATIONS 10000000
 289
 290 static volatile int wait_;
 291 static long num_threads_;
 292 static skiplist_t *skiplist_;
 293
 294 void *worker (void *arg) {
 295     int id = (int)(size_t)arg;
 296
 297     unsigned int rand_seed = id+1;//rdtsc_l();
 298
 299     // Wait for all the worker threads to be ready.
 300     SYNC_ADD(&wait_, -1);
 301     do {} while (wait_);
 302
 303     for (int i = 0; i < NUM_ITERATIONS/num_threads_; ++i) {
 304         int n = rand_r(&rand_seed);
 305         int key = (n & 0xF) + 1;
 306         if (n & (1 << 8)) {
 307             skiplist_add_i(n, skiplist_, key, 1);
 308         } else {
 309             skiplist_remove(skiplist_, key);
 310         }
 311
 312         rcu_update();
 313     }
 314
 315     return NULL;
 316 }
 317
 318 int main (int argc, char **argv) {
 319     nbd_init();
 320     lwt_set_trace_level("s3");
 321
 322     char* program_name = argv[0];
 323     pthread_t thread[MAX_NUM_THREADS];
 324
 325     if (argc > 2) {
 326         fprintf(stderr, "Usage: %s num_threads\n", program_name);
 327         return -1;
 328     }
 329
 330     num_threads_ = 2;
 331     if (argc == 2)
 332     {
 333         errno = 0;
 334         num_threads_ = strtol(argv[1], NULL, 10);
 335         if (errno) {
 336             fprintf(stderr, "%s: Invalid argument for number of threads\n", program_name);
 337             return -1;
 338         }
 339         if (num_threads_ <= 0) {
 340             fprintf(stderr, "%s: Number of threads must be at least 1\n", program_name);
 341             return -1;
 342         }
 343         if (num_threads_ > MAX_NUM_THREADS) {
 344             fprintf(stderr, "%s: Number of threads cannot be more than %d\n", program_name, MAX_NUM_THREADS);
 345             return -1;
 346         }
 347     }
 348
 349     skiplist_ = skiplist_alloc();
 350
 351     struct timeval tv1, tv2;
 352     gettimeofday(&tv1, NULL);
 353
 354     wait_ = num_threads_;
 355
 356     for (int i = 0; i < num_threads_; ++i) {
 357         int rc = nbd_thread_create(thread + i, i, worker, (void*)(size_t)i);
 358         if (rc != 0) { perror("pthread_create"); return rc; }
 359     }
 360
 361     for (int i = 0; i < num_threads_; ++i) {
 362         pthread_join(thread[i], NULL);
 363     }
 364
 365     gettimeofday(&tv2, NULL);
 366     int ms = (int)(1000000*(tv2.tv_sec - tv1.tv_sec) + tv2.tv_usec - tv1.tv_usec) / 1000;
 367     printf("Th:%ld Time:%dms\n", num_threads_, ms);
 368     skiplist_print(skiplist_);
 369     lwt_dump("lwt.out");
 370
 371     return 0;
 372 }
 373 #endif//skiplist_test