X-Git-Url: https://pd.if.org/git/?p=nbds;a=blobdiff_plain;f=struct%2Fskiplist.c;h=a3fab396f8a16cee038c17d7841b279e5267b71c;hp=1d65efa0d191c1175c043276757c75b41be5257f;hb=9b3e566281f7e2ac0683205042796958bfd8939f;hpb=a1fae129c758d7ea83dfdbb5f14ec1df12f0aa34 diff --git a/struct/skiplist.c b/struct/skiplist.c index 1d65efa..a3fab39 100644 --- a/struct/skiplist.c +++ b/struct/skiplist.c @@ -19,60 +19,77 @@ #include "common.h" #include "runtime.h" #include "struct.h" +#include "nstring.h" #include "mem.h" #include "tls.h" // Setting MAX_LEVEL to 0 essentially makes this data structure the Harris-Michael lock-free list -// in list.c +// (in list.c). #define MAX_LEVEL 31 typedef struct node { - uint64_t key; - uint64_t value; + nstring_t *key; + uint64_t val; int top_level; struct node *next[]; } node_t; -typedef struct sl { +struct sl { node_t *head; -} skiplist_t; +}; static int random_level (void) { unsigned r = nbd_rand(); - if (r&1) + if (r & 1) return 0; - int n = __builtin_ctz(r)-1; #if MAX_LEVEL < 31 - if (n > MAX_LEVEL) - return MAX_LEVEL; + r |= 1 << (MAX_LEVEL+1); #endif + int n = __builtin_ctz(r)-1; assert(n <= MAX_LEVEL); return n; } -node_t *node_alloc (int level, uint64_t key, uint64_t value) { +node_t *node_alloc (int level, const void *key_data, uint32_t key_len, uint64_t val) { assert(level >= 0 && level <= MAX_LEVEL); size_t sz = sizeof(node_t) + (level + 1) * sizeof(node_t *); node_t *item = (node_t *)nbd_malloc(sz); memset(item, 0, sz); - item->key = key; - item->value = value; + // If is -1 it indicates is an integer and not a pointer + item->key = (key_len == (unsigned)-1) + ? (void *)TAG_VALUE(key_data) + : ns_alloc(key_data, key_len); + item->val = val; item->top_level = level; return item; } +static void node_free (node_t *item) { + if (!IS_TAGGED(item->key)) { + nbd_free(item->key); + } + nbd_free(item); +} + +static void node_defer_free (node_t *item) { + if (!IS_TAGGED(item->key)) { + nbd_defer_free(item->key); + } + nbd_defer_free(item); +} + skiplist_t *sl_alloc (void) { skiplist_t *sl = (skiplist_t *)nbd_malloc(sizeof(skiplist_t)); - sl->head = node_alloc(MAX_LEVEL, 0, 0); + sl->head = node_alloc(MAX_LEVEL, " ", 0, 0); memset(sl->head->next, 0, (MAX_LEVEL+1) * sizeof(skiplist_t *)); return sl; } -static node_t *find_preds (node_t *preds[MAX_LEVEL+1], int n, skiplist_t *sl, uint64_t key, int help_remove) { +static node_t *find_preds (node_t **preds, node_t **succs, int n, skiplist_t *sl, const void *key_data, uint32_t key_len, int help_remove) { node_t *pred = sl->head; node_t *item = NULL; - TRACE("s3", "find_preds: searching for key %p in sl (head is %p)", key, pred); - + TRACE("s3", "find_preds: searching for key %p in sl (head is %p)", key_data, pred); + int x; int start_level = MAX_LEVEL; #if MAX_LEVEL > 2 // Optimization for small lists. No need to traverse empty higher levels. @@ -95,17 +112,17 @@ static node_t *find_preds (node_t *preds[MAX_LEVEL+1], int n, skiplist_t *sl, ui item = pred->next[level]; if (EXPECT_FALSE(IS_TAGGED(item))) { TRACE("s3", "find_preds: pred %p is marked for removal (item %p); retry", pred, item); - return find_preds(preds, n, sl, key, help_remove); // retry + return find_preds(preds, succs, n, sl, key_data, key_len, help_remove); // retry } while (item != NULL) { 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); + TRACE("s3", "find_preds: key %p", STRIP_TAG(item->key), item->val); - // Marked items are logically removed, but not fully unlinked yet. + // A tag means an item is logically removed but not physically unlinked yet. while (EXPECT_FALSE(IS_TAGGED(next))) { - // Skip over partially removed items. + // Skip over logically removed items. if (!help_remove) { item = (node_t *)STRIP_TAG(item->next); if (EXPECT_FALSE(item == NULL)) @@ -114,7 +131,7 @@ static node_t *find_preds (node_t *preds[MAX_LEVEL+1], int n, skiplist_t *sl, ui continue; } - // Unlink partially removed items. + // Unlink logically removed items. node_t *other; if ((other = SYNC_CAS(&pred->next[level], item, STRIP_TAG(next))) == item) { item = (node_t *)STRIP_TAG(next); @@ -125,11 +142,11 @@ static node_t *find_preds (node_t *preds[MAX_LEVEL+1], int n, skiplist_t *sl, ui 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); } + if (level == 0) { node_defer_free(other); } } else { TRACE("s3", "find_preds: lost race to unlink from pred %p; its link changed to %p", pred, other); if (IS_TAGGED(other)) - return find_preds(preds, n, sl, key, help_remove); // retry + return find_preds(preds, succs, n, sl, key_data, key_len, help_remove); // retry item = other; if (EXPECT_FALSE(item == NULL)) break; @@ -141,7 +158,10 @@ static node_t *find_preds (node_t *preds[MAX_LEVEL+1], int n, skiplist_t *sl, ui break; // If we reached the key (or passed where it should be), we found a pred. Save it and continue down. - if (item->key >= key) { + x = (IS_TAGGED(item->key)) + ? (STRIP_TAG(item->key) - (uint64_t)key_data) + : ns_cmp_raw(item->key, key_data, key_len); + if (x >= 0) { TRACE("s3", "find_preds: found pred %p item %p", pred, item); break; } @@ -149,107 +169,106 @@ static node_t *find_preds (node_t *preds[MAX_LEVEL+1], int n, skiplist_t *sl, ui pred = item; item = next; } - if (preds != NULL) { - preds[level] = pred; + + // The cast to unsigned is for the case when n is -1. + if ((unsigned)level <= (unsigned)n) { + if (preds != NULL) { + preds[level] = pred; + } + if (succs != NULL) { + succs[level] = item; + } } } if (n == -1 && item != NULL) { - assert(preds != NULL); for (int level = start_level + 1; level <= item->top_level; ++level) { preds[level] = sl->head; } } - return item; + return x == 0 ? item : NULL; } // Fast find that does not help unlink partially removed nodes and does not return the node's predecessors. -uint64_t sl_lookup (skiplist_t *sl, uint64_t key) { - TRACE("s3", "sl_lookup: searching for key %p in sl %p", key, sl); - node_t *item = find_preds(NULL, 0, sl, key, FALSE); +uint64_t sl_lookup (skiplist_t *sl, const void *key_data, uint32_t key_len) { + TRACE("s3", "sl_lookup: searching for key %p in skiplist %p", key, sl); + node_t *item = find_preds(NULL, NULL, 0, sl, key_data, key_len, FALSE); // If we found an matching the return its value. - return (item && item->key == key) ? item->value : DOES_NOT_EXIST; + return item != NULL ? item->val : DOES_NOT_EXIST; } // Insert the if it doesn't already exist in -uint64_t sl_add (skiplist_t *sl, uint64_t key, uint64_t value) { - TRACE("s3", "sl_add: inserting key %p value %p", key, value); +uint64_t sl_add (skiplist_t *sl, const void *key_data, uint32_t key_len, uint64_t val) { + TRACE("s3", "sl_add: inserting key %p val %p", key_data, val); node_t *preds[MAX_LEVEL+1]; - node_t *item = NULL; + node_t *nexts[MAX_LEVEL+1]; + node_t *new_item = NULL; + int n = random_level(); do { - int n = random_level(); - node_t *next = find_preds(preds, n, sl, key, TRUE); + node_t *old_item = find_preds(preds, nexts, n, sl, key_data, key_len, TRUE); // If a node matching already exists in , return its value. - if (next != NULL && next->key == key) { - TRACE("s3", "sl_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; + if (old_item != NULL) { + TRACE("s3", "sl_add: there is already an item %p (value %p) with the same key", nexts[0], nexts[0]->val); + return nexts[0]->val; } - // First insert into the bottom level. - if (EXPECT_TRUE(item == NULL)) { item = node_alloc(n, key, value); } - TRACE("s3", "sl_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) { - node_t *pred = preds[level]; - item->next[level] = pred->next[level]; - } + // First insert into the bottom level. + TRACE("s3", "sl_add: attempting to insert item between %p and %p", preds[0], nexts[0]); + new_item = node_alloc(n, key_data, key_len, val); node_t *pred = preds[0]; - node_t *other = SYNC_CAS(&pred->next[0], next, item); + node_t *next = new_item->next[0] = nexts[0]; + for (int level = 1; level <= new_item->top_level; ++level) { + new_item->next[level] = nexts[level]; + } + node_t *other = SYNC_CAS(&pred->next[0], next, new_item); if (other == next) { - TRACE("s3", "sl_add: successfully inserted item %p at level 0", item, 0); + TRACE("s3", "sl_add: successfully inserted item %p at level 0", new_item, 0); break; // success } TRACE("s3", "sl_add: failed to change pred's link: expected %p found %p", next, other); + node_free(new_item); } while (1); - // Insert into from the bottom level up. - for (int level = 1; level <= item->top_level; ++level) { + // Insert into from the bottom level up. + for (int level = 1; level <= new_item->top_level; ++level) { + node_t *pred = preds[level]; + node_t *next = nexts[level]; do { - node_t *pred; - node_t *next; - do { - pred = preds[level]; - next = pred->next[level]; - if (next == NULL) // item goes at the end of the list - break; - if (!IS_TAGGED(next) && next->key > key) // pred's link changed - break; - find_preds(preds, item->top_level, sl, key, TRUE); - } while (1); + TRACE("s3", "sl_add: attempting to insert item between %p and %p", pred, next); + node_t *other = SYNC_CAS(&pred->next[level], next, new_item); + if (other == next) { + TRACE("s3", "sl_add: successfully inserted item %p at level %llu", new_item, level); + break; // success + } + TRACE("s3", "sl_add: failed to change pred's link: expected %p found %p", next, other); + find_preds(preds, nexts, new_item->top_level, sl, key_data, key_len, TRUE); + pred = preds[level]; + next = nexts[level]; + // Update 's next pointer 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]; + node_t *old_next = ((volatile node_t *)new_item)->next[level]; if (IS_TAGGED(old_next)) - return DOES_NOT_EXIST; // success + return val; - // 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) + // Use a CAS so we do not inadvertantly stomp on a mark another thread placed on the item. + if (old_next == next || SYNC_CAS(&new_item->next[level], old_next, next) == old_next) break; } while (1); - - TRACE("s3", "sl_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", "sl_add: successfully inserted item %p at level %llu", item, level); - break; // success - } - TRACE("s3", "sl_add: failed to change pred's link: expected %p found %p", next, other); - } while (1); } - return value; + return val; } -uint64_t sl_remove (skiplist_t *sl, uint64_t key) { - TRACE("s3", "sl_remove: removing item with key %p from sl %p", key, sl); +uint64_t sl_remove (skiplist_t *sl, const void *key_data, uint32_t key_len) { + TRACE("s3", "sl_remove: removing item with key %p from skiplist %p", key_data, sl); node_t *preds[MAX_LEVEL+1]; - node_t *item = find_preds(preds, -1, sl, key, TRUE); - if (item == NULL || item->key != key) { - TRACE("s3", "sl_remove: remove failed, an item with a matching key does not exist in the sl", 0, 0); + node_t *item = find_preds(preds, NULL, -1, sl, key_data, key_len, TRUE); + if (item == NULL) { + TRACE("s3", "sl_remove: remove failed, an item with a matching key does not exist in the skiplist", 0, 0); return DOES_NOT_EXIST; } @@ -272,9 +291,11 @@ uint64_t sl_remove (skiplist_t *sl, uint64_t key) { } } - uint64_t value = item->value; + uint64_t val = item->val; - // Unlink from the top down. + // Unlink from . If we lose a race to another thread just back off. It is safe to leave the + // item partially unlinked for a later call (or some other thread) to physically unlink. By marking the + // item earlier, we logically removed it. int level = item->top_level; while (level >= 0) { node_t *pred = preds[level]; @@ -283,16 +304,14 @@ uint64_t sl_remove (skiplist_t *sl, uint64_t key) { node_t *other = NULL; if ((other = SYNC_CAS(&pred->next[level], item, STRIP_TAG(next))) != item) { TRACE("s3", "sl_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 . - return value; + return val; } --level; } // The thread that completes the unlink should free the memory. - nbd_defer_free(item); - return value; + node_defer_free(item); + return val; } void sl_print (skiplist_t *sl) { @@ -314,7 +333,12 @@ void sl_print (skiplist_t *sl) { node_t *item = sl->head; while (item) { int is_marked = IS_TAGGED(item->next[0]); - printf("%s%p:0x%llx ", is_marked ? "*" : "", item, item->key); + + if (IS_TAGGED(item->key)) { + printf("%s%p:%llx ", is_marked ? "*" : "", item, STRIP_TAG(item->key)); + } else { + printf("%s%p:%s ", is_marked ? "*" : "", item, (char *)ns_data(item->key)); + } if (item != sl->head) { printf("[%d]", item->top_level); } else { @@ -332,93 +356,3 @@ void sl_print (skiplist_t *sl) { 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 *sl_; - -void *worker (void *arg) { - - // 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) { - unsigned r = nbd_rand(); - int key = r & 0xF; - if (r & (1 << 8)) { - sl_add(sl_, key, 1); - } else { - sl_remove(sl_, 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; - } - } - - sl_ = sl_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; - sl_print(sl_); - printf("Th:%ld Time:%dms\n", num_threads_, ms); - - return 0; -} -#endif//skiplist_test