From 7899dce86342c50be9a52d148fa27375bdb5d218 Mon Sep 17 00:00:00 2001 From: jdybnis Date: Sun, 23 Nov 2008 08:36:33 +0000 Subject: [PATCH] lock-free skiplist --- include/lwt.h | 2 + makefile | 2 +- runtime/lwt.c | 8 +- struct/list.c | 19 +-- struct/skiplist.c | 373 ++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 392 insertions(+), 12 deletions(-) create mode 100644 struct/skiplist.c diff --git a/include/lwt.h b/include/lwt.h index 9f07e7a..20c867e 100644 --- a/include/lwt.h +++ b/include/lwt.h @@ -37,4 +37,6 @@ static inline void lwt_trace (const char *flag, const char *format, size_t value } } +void lwt_halt (void); + #endif//LWT_H diff --git a/makefile b/makefile index 0a14c0e..bc31f32 100644 --- a/makefile +++ b/makefile @@ -15,7 +15,7 @@ RUNTIME_SRCS := runtime/runtime.c runtime/rcu.c runtime/lwt.c runtime/mem.c TEST_SRCS := $(RUNTIME_SRCS) rcu_test_SRCS := $(TEST_SRCS) list_test_SRCS := $(TEST_SRCS) struct/list.c -#skiplist_test_SRCS := $(TEST_SRCS) struct/skiplist.c +skiplist_test_SRCS := $(TEST_SRCS) struct/skiplist.c ht_test_SRCS := $(TEST_SRCS) struct/ht.c test/ht_test.c test/CuTest.c txn_test_SRCS := $(TEST_SRCS) struct/ht.c txn/txn.c diff --git a/runtime/lwt.c b/runtime/lwt.c index 8212c5c..83e8ee1 100644 --- a/runtime/lwt.c +++ b/runtime/lwt.c @@ -58,7 +58,7 @@ static inline void dump_record (FILE *file, int thread_id, lwt_record_t *r, uint int level = ((size_t)r->format >> 48) & 0xFF; if (flag_state_[(unsigned)flag] >= level) { char s[3] = {flag, level, '\0'}; - fprintf(file, "%09llu %d %s ", ((uint64_t)r->timestamp - offset) >> 6, thread_id, s); + fprintf(file, "%09llu %d %s ", ((uint64_t)r->timestamp - offset) >> 5, thread_id, s); const char *format = (const char *)((size_t)r->format & MASK(48)); // strip out the embedded flags fprintf(file, format, r->value1, r->value2); fprintf(file, "\n"); @@ -80,6 +80,10 @@ static void dump_buffer (FILE *file, int thread_id, uint64_t offset) } } +void lwt_halt (void) { + halt_ = 1; +} + void lwt_dump (const char *file_name) { halt_ = 1; @@ -115,7 +119,7 @@ void lwt_dump (const char *file_name) } void lwt_trace_i (const char *format, size_t value1, size_t value2) { - if (*(volatile int *)&halt_) + if (halt_) return; LOCALIZE_THREAD_LOCAL(tid_, int); lwt_buffer_t *tb = lwt_buf_[tid_]; diff --git a/struct/list.c b/struct/list.c index aa28b85..40a9735 100644 --- a/struct/list.c +++ b/struct/list.c @@ -50,9 +50,9 @@ static node_t *find_pred (node_t **pred_ptr, list_t *list, uint64_t key, int hel do { node_t *next = item->next; TRACE("l3", "find_pred: visiting item %p (next %p)", item, next); - TRACE("l3", "find_pred: key %p value %p", item->key, item->value); + TRACE("l3", "find_pred: key %p", item->key, item->value); - // Marked items are partially removed. + // Marked items are logically removed, but not unlinked yet. while (EXPECT_FALSE(IS_TAGGED(next))) { // Skip over partially removed items. @@ -67,7 +67,10 @@ static node_t *find_pred (node_t **pred_ptr, list_t *list, uint64_t key, int hel if ((other = SYNC_CAS(&pred->next, item, STRIP_TAG(next))) == item) { item = (node_t *)STRIP_TAG(next); next = item->next; - TRACE("l3", "find_pred: unlinked item; %p is the new item (next is %p)", item, next); + TRACE("l3", "find_pred: unlinked item %p from pred %p", item, pred); + TRACE("l3", "find_pred: now item is %p next is %p", item, next); + + // The thread that completes the unlink should free the memory. nbd_defer_free(other); } else { TRACE("l3", "find_pred: lost race to unlink item from pred %p; its link changed to %p", pred, other); @@ -80,7 +83,7 @@ static node_t *find_pred (node_t **pred_ptr, list_t *list, uint64_t key, int hel // If we reached the key (or passed where it should be), we found the right predesssor if (item->key >= key) { - TRACE("l3", "find_pred: returning pred %p and item %p", pred, item); + TRACE("l3", "find_pred: found pred %p item %p", pred, item); if (pred_ptr != NULL) { *pred_ptr = pred; } @@ -123,7 +126,7 @@ uint64_t list_add (list_t *list, uint64_t key, uint64_t value) { item->next = next; node_t *other = SYNC_CAS(&pred->next, next, item); if (other == next) { - TRACE("l3", "list_add: insert was successful", 0, 0); + TRACE("l3", "list_add: successfully inserted item %p", item, 0); return DOES_NOT_EXIST; // success } TRACE("l3", "list_add: failed to change pred's link: expected %p found %p", next, other); @@ -159,8 +162,8 @@ uint64_t list_remove (list_t *list, uint64_t key) { node_t *other; if ((other = SYNC_CAS(&pred->next, item, next)) != item) { TRACE("l3", "list_remove: unlink failed; pred's link changed from %p to %p", item, other); - // By being marked, the item was logically removed. It is safe to leave it for - // another thread to finish physically removing it from the skiplist. + // By marking the item earlier, we logically removed it. It is safe to leave the item. + // Another thread will finish physically removing it from the list. return value; } @@ -201,7 +204,6 @@ void *worker (void *arg) { // Wait for all the worker threads to be ready. SYNC_ADD(&wait_, -1); do {} while (wait_); - __asm__ __volatile__("lfence"); for (int i = 0; i < NUM_ITERATIONS/num_threads_; ++i) { int n = rand_r(&rand_seed); @@ -254,7 +256,6 @@ int main (int argc, char **argv) { struct timeval tv1, tv2; gettimeofday(&tv1, NULL); - __asm__ __volatile__("sfence"); wait_ = num_threads_; for (int i = 0; i < num_threads_; ++i) { diff --git a/struct/skiplist.c b/struct/skiplist.c new file mode 100644 index 0000000..7686b76 --- /dev/null +++ b/struct/skiplist.c @@ -0,0 +1,373 @@ +/* + * Written by Josh Dybnis and released to the public domain, as explained at + * http://creativecommons.org/licenses/publicdomain + * + * lock-free skiplist + */ +#include +#include + +#include "common.h" +#include "struct.h" +#include "mem.h" + +#define MAX_LEVEL 3 + +typedef struct node { + uint64_t key; + uint64_t value; + int top_level; + struct node *next[]; +} node_t; + +typedef struct skiplist { + node_t *head; + node_t *last; + int top_level; +} skiplist_t; + +static int random_level (int r) { + if (r&1) + return 0; + int n = __builtin_ctz(r); + if (n < MAX_LEVEL) + return n; + return MAX_LEVEL; +} + +node_t *node_alloc (int top_level, uint64_t key, uint64_t value) { + assert(top_level >= 0 && top_level <= MAX_LEVEL); + size_t sz = sizeof(node_t) + (top_level + 1) * sizeof(node_t *); + node_t *item = (node_t *)nbd_malloc(sz); + memset(item, 0, sz); + item->key = key; + item->value = value; + item->top_level = top_level; + return item; +} + +skiplist_t *skiplist_alloc (void) { + skiplist_t *skiplist = (skiplist_t *)nbd_malloc(sizeof(skiplist_t)); + skiplist->head = node_alloc(MAX_LEVEL, 0, 0); + skiplist->last = node_alloc(MAX_LEVEL, (uint64_t)-1, 0); + for (int level = 0; level <= MAX_LEVEL; ++level) { + skiplist->head->next[level] = skiplist->last; + } + return skiplist; +} + +static node_t *find_preds (node_t **preds, skiplist_t *skiplist, uint64_t key, int help_remove) { + node_t *pred = skiplist->head; + node_t *item = NULL; + TRACE("s3", "find_preds: searching for key %p in skiplist (head is %p)", key, pred); +#ifndef NDEBUG + int count = 0; +#endif + + // Traverse the levels of the skiplist from the top level to the bottom + for (int level = MAX_LEVEL; level >= 0; --level) { + TRACE("s3", "find_preds: level %llu", level, 0); + item = pred->next[level]; + if (IS_TAGGED(item)) { + TRACE("s3", "find_preds: pred %p is marked for removal (item %p); retry", pred, item); + return find_preds(preds, skiplist, key, help_remove); // retry + } + do { + node_t *next = item->next[level]; + TRACE("s3", "find_preds: visiting item %p (next %p)", item, next); + TRACE("s3", "find_preds: key %p", item->key, 0); + + // Marked items are logically removed, but not fully unlinked yet. + while (EXPECT_FALSE(IS_TAGGED(next))) { + + // Skip over partially removed items. + if (!help_remove) { + item = (node_t *)STRIP_TAG(item->next); + next = item->next[level]; + continue; + } + + // Unlink partially removed items. + node_t *other; + if ((other = SYNC_CAS(&pred->next[level], item, STRIP_TAG(next))) == item) { + item = (node_t *)STRIP_TAG(next); + next = item->next[level]; + TRACE("s3", "find_preds: unlinked item %p from pred %p", item, pred); + TRACE("s3", "find_preds: now item is %p next is %p", item, next); + + // The thread that completes the unlink should free the memory. + if (level == 0) { nbd_defer_free(other); } + } else { + TRACE("s3", "find_preds: lost race to unlink item from pred %p; its link changed to %p",pred,other); + if (IS_TAGGED(other)) + return find_preds(preds, skiplist, key, help_remove); // retry + item = other; + next = item->next[level]; + } + } + + // If we reached the key (or passed where it should be), we found a pred. Save it and continue down. + if (item->key >= key) { + TRACE("s3", "find_preds: found pred %p item %p", pred, item); + if (preds != NULL) { + preds[level] = pred; + } + break; + } + + assert(count++ < 18); + pred = item; + item = next; + + } while (1); + } + return item; +} + +// Fast find that does not help unlink partially removed nodes and does not return the node's predecessors. +uint64_t skiplist_lookup (skiplist_t *skiplist, uint64_t key) { + TRACE("s3", "skiplist_lookup: searching for key %p in skiplist %p", key, skiplist); + node_t *item = find_preds(NULL, skiplist, key, FALSE); + + // If we found an matching the return its value. + return (item->key == key) ? item->value : DOES_NOT_EXIST; +} + +// Insert the if it doesn't already exist in the +uint64_t skiplist_add_i (int n, skiplist_t *skiplist, uint64_t key, uint64_t value) { + TRACE("s3", "skiplist_add: inserting key %p value %p", key, value); + node_t *preds[MAX_LEVEL+1]; + node_t *item = NULL; + do { + node_t *next = find_preds(preds, skiplist, key, TRUE); + + // If a node matching already exists in the skiplist, return its value. + if (next->key == key) { + TRACE("s3", "skiplist_add: there is already an item %p (value %p) with the same key", next, next->value); + if (EXPECT_FALSE(item != NULL)) { nbd_free(item); } + return next->value; + } + + // First insert into the bottom level. + if (EXPECT_TRUE(item == NULL)) { item = node_alloc(random_level(n), key, value); } + TRACE("s3", "skiplist_add: attempting to insert item between %p and %p", preds[0], next); + item->next[0] = next; + for (int level = 1; level <= item->top_level; ++level) { + item->next[level] = preds[level]->next[level]; + } + node_t *other = SYNC_CAS(&preds[0]->next[0], next, item); + if (other == next) { + TRACE("s3", "skiplist_add: successfully inserted item %p at level 0", item, 0); + break; // success + } + TRACE("s3", "skiplist_add: failed to change pred's link: expected %p found %p", next, other); + + } while (1); + + // Insert into the skiplist from the bottom level up. + for (int level = 1; level <= item->top_level; ++level) { + do { + node_t *pred = preds[level]; + node_t *next = pred->next[level]; + while (EXPECT_FALSE(IS_TAGGED(next) || next->key < key)) { + find_preds(preds, skiplist, key, TRUE); + pred = preds[level]; + next = pred->next[level]; + } + + do { + // There in no need to continue linking in the item if another thread removed it. + node_t *old_next = ((volatile node_t *)item)->next[level]; + if (IS_TAGGED(old_next)) + return DOES_NOT_EXIST; // success + + // Use a CAS so we to not inadvertantly remove a mark another thread placed on the item. + if (next == old_next || SYNC_CAS(&item->next[level], old_next, next) == old_next) + break; + } while (1); + + TRACE("s3", "skiplist_add: attempting to insert item between %p and %p", pred, next); + node_t *other = SYNC_CAS(&pred->next[level], next, item); + if (other == next) { + TRACE("s3", "skiplist_add: successfully inserted item %p at level %llu", item, level); + break; // success + } + TRACE("s3", "skiplist_add: failed to change pred's link: expected %p found %p", next, other); + + } while (1); + } + return value; +} + +uint64_t skiplist_remove (skiplist_t *skiplist, uint64_t key) { + TRACE("s3", "skiplist_remove: removing item with key %p from skiplist %p", key, skiplist); + node_t *preds[MAX_LEVEL+1]; + node_t *item = find_preds(preds, skiplist, key, TRUE); + if (item->key != key) { + TRACE("s3", "skiplist_remove: remove failed, an item with a matching key does not exist in the skiplist", 0, 0); + return DOES_NOT_EXIST; + } + + // Mark removed at each level of the skiplist from the top down. This must be atomic. If multiple threads + // try to remove the same item only one of them should succeed. Marking the bottom level establishes which of + // them succeeds. + for (int level = item->top_level; level >= 0; --level) { + if (EXPECT_FALSE(IS_TAGGED(item->next[level]))) { + TRACE("s3", "skiplist_remove: %p is already marked for removal by another thread", item, 0); + if (level == 0) + return DOES_NOT_EXIST; + continue; + } + node_t *next = SYNC_FETCH_AND_OR(&item->next[level], TAG); + if (EXPECT_FALSE(IS_TAGGED(next))) { + TRACE("s3", "skiplist_remove: lost race -- %p is already marked for removal by another thread", item, 0); + if (level == 0) + return DOES_NOT_EXIST; + continue; + } + } + + uint64_t value = item->value; + + // Unlink from the top down. + int level = item->top_level; + while (level >= 0) { + node_t *pred = preds[level]; + node_t *next = item->next[level]; + TRACE("s3", "skiplist_remove: link item's pred %p to it's successor %p", pred, STRIP_TAG(next)); + node_t *other = NULL; + if ((other = SYNC_CAS(&pred->next[level], item, STRIP_TAG(next))) != item) { + TRACE("s3", "skiplist_remove: unlink failed; pred's link changed from %p to %p", item, other); + // By marking the item earlier, we logically removed it. It is safe to leave the item partially + // unlinked. Another thread will finish physically removing it from the skiplist. + return value; + } + --level; + } + + // The thread that completes the unlink should free the memory. + nbd_defer_free(item); + return value; +} + +void skiplist_print (skiplist_t *skiplist) { + for (int level = MAX_LEVEL; level >= 0; --level) { + printf("(%d) ", level); + node_t *item = skiplist->head; + while (item) { + node_t *next = item->next[level]; + printf("%s%p:0x%llx ", IS_TAGGED(next) ? "*" : "", item, item->key); + item = (node_t *)STRIP_TAG(next); + } + printf("\n"); + fflush(stdout); + } + + node_t *item = skiplist->head; + while (item) { + assert(item->top_level <= MAX_LEVEL); + int is_marked = IS_TAGGED(item->next[0]); + printf("%s%p:0x%llx (%d", is_marked ? "*" : "", item, item->key, item->top_level); + for (int i = 1; i <= item->top_level; ++i) { + node_t *next = (node_t *)STRIP_TAG(item->next[i]); + printf(" %p:0x%llx", item->next[i], next ? next->key : 0); + } + printf(")\n"); + fflush(stdout); + item = (node_t *)STRIP_TAG(item->next[0]); + } +} + +#ifdef MAKE_skiplist_test +#include +#include +#include + +#include "runtime.h" + +#define NUM_ITERATIONS 10000000 + +static volatile int wait_; +static long num_threads_; +static skiplist_t *skiplist_; + +void *worker (void *arg) { + int id = (int)(size_t)arg; + + unsigned int rand_seed = id+1;//rdtsc_l(); + + // Wait for all the worker threads to be ready. + SYNC_ADD(&wait_, -1); + do {} while (wait_); + + for (int i = 0; i < NUM_ITERATIONS/num_threads_; ++i) { + int n = rand_r(&rand_seed); + int key = (n & 0xF) + 1; + if (n & (1 << 8)) { + skiplist_add_i(n, skiplist_, key, 1); + } else { + skiplist_remove(skiplist_, key); + } + + rcu_update(); + } + + return NULL; +} + +int main (int argc, char **argv) { + nbd_init(); + lwt_set_trace_level("s3"); + + char* program_name = argv[0]; + pthread_t thread[MAX_NUM_THREADS]; + + if (argc > 2) { + fprintf(stderr, "Usage: %s num_threads\n", program_name); + return -1; + } + + num_threads_ = 2; + if (argc == 2) + { + errno = 0; + num_threads_ = strtol(argv[1], NULL, 10); + if (errno) { + fprintf(stderr, "%s: Invalid argument for number of threads\n", program_name); + return -1; + } + if (num_threads_ <= 0) { + fprintf(stderr, "%s: Number of threads must be at least 1\n", program_name); + return -1; + } + if (num_threads_ > MAX_NUM_THREADS) { + fprintf(stderr, "%s: Number of threads cannot be more than %d\n", program_name, MAX_NUM_THREADS); + return -1; + } + } + + skiplist_ = skiplist_alloc(); + + struct timeval tv1, tv2; + gettimeofday(&tv1, NULL); + + wait_ = num_threads_; + + for (int i = 0; i < num_threads_; ++i) { + int rc = nbd_thread_create(thread + i, i, worker, (void*)(size_t)i); + if (rc != 0) { perror("pthread_create"); return rc; } + } + + for (int i = 0; i < num_threads_; ++i) { + pthread_join(thread[i], NULL); + } + + gettimeofday(&tv2, NULL); + int ms = (int)(1000000*(tv2.tv_sec - tv1.tv_sec) + tv2.tv_usec - tv1.tv_usec) / 1000; + printf("Th:%ld Time:%dms\n", num_threads_, ms); + skiplist_print(skiplist_); + lwt_dump("lwt.out"); + + return 0; +} +#endif//skiplist_test -- 2.40.0