X-Git-Url: https://pd.if.org/git/?p=nbds;a=blobdiff_plain;f=struct%2Fskiplist.c;h=e8060ddb3e6e820282e79d8a2032d3ebba444a2d;hp=1d65efa0d191c1175c043276757c75b41be5257f;hb=2d93f3b29622488bde80b6cd18661fd7eb603eee;hpb=a1fae129c758d7ea83dfdbb5f14ec1df12f0aa34 diff --git a/struct/skiplist.c b/struct/skiplist.c index 1d65efa..e8060dd 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("s2", "find_preds: searching for key %p in sl (head is %p)", key_data, pred); + int d; int start_level = MAX_LEVEL; #if MAX_LEVEL > 2 // Optimization for small lists. No need to traverse empty higher levels. @@ -94,42 +111,42 @@ static node_t *find_preds (node_t *preds[MAX_LEVEL+1], int n, skiplist_t *sl, ui TRACE("s3", "find_preds: level %llu", level, 0); 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 + TRACE("s2", "find_preds: pred %p is marked for removal (item %p); retry", pred, item); + 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); - // 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)) break; + TRACE("s3", "find_preds: skipping marked item %p (next is %p)", item, next); next = item->next[level]; continue; } - // Unlink partially removed items. + // Unlink logically removed items. node_t *other; + TRACE("s3", "find_preds: unlinking marked item %p; next is %p", item, next); if ((other = SYNC_CAS(&pred->next[level], item, STRIP_TAG(next))) == item) { item = (node_t *)STRIP_TAG(next); if (EXPECT_FALSE(item == NULL)) break; 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); + TRACE("s3", "find_preds: now the current 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); + TRACE("s2", "find_preds: lost race to unlink item %p from pred %p", item, pred); + TRACE("s2", "find_preds: pred's link changed to %p", other, 0); 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; @@ -140,116 +157,179 @@ static node_t *find_preds (node_t *preds[MAX_LEVEL+1], int n, skiplist_t *sl, ui if (EXPECT_FALSE(item == NULL)) 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) { - TRACE("s3", "find_preds: found pred %p item %p", pred, item); + TRACE("s3", "find_preds: visiting item %p (next is %p)", item, next); + TRACE("s4", "find_preds: key %p val %p", STRIP_TAG(item->key), item->val); + + // A tagged key is an integer, otherwise it is a pointer to a string + if (IS_TAGGED(item->key)) { + d = (STRIP_TAG(item->key) - (uint64_t)key_data); + } else { + int item_key_len = item->key->len; + int len = (key_len < item_key_len) ? key_len : item_key_len; + d = memcmp(item->key->data, key_data, len); + if (d == 0) { d = item_key_len - key_len; } + } + + if (d >= 0) { + TRACE("s2", "find_preds: found pred %p item %p", pred, item); break; } 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; + } } } + + // fill in empty levels 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; + + if (d == 0) { + TRACE("s2", "find_preds: found matching item %p in skiplist, pred is %p", item, pred); + return item; + } + TRACE("s2", "find_preds: found proper place for key %p in skiplist, pred is %p. returning null", key_data, pred); + return 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("s1", "sl_lookup: searching for key %p in skiplist %p", key_data, 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; + if (item != NULL) { + uint64_t val = item->val; + if (val != DOES_NOT_EXIST) { + TRACE("s1", "sl_lookup: found item %p. val %p. returning item", item, item->val); + return val; + } + } + + TRACE("l1", "sl_lookup: no item in the skiplist matched the key", 0, 0); + return 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_cas (skiplist_t *sl, const void *key_data, uint32_t key_len, uint64_t expectation, uint64_t new_val) { + TRACE("s1", "sl_cas: key %p skiplist %p", key_data, sl); + TRACE("s1", "sl_cas: expectation %p new value %p", expectation, new_val); + ASSERT((int64_t)new_val > 0); + 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); - - // 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; - } + node_t *old_item = find_preds(preds, nexts, n, sl, key_data, key_len, TRUE); + if (old_item == NULL) { - // 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]; - } - node_t *pred = preds[0]; - node_t *other = SYNC_CAS(&pred->next[0], next, item); - if (other == next) { - TRACE("s3", "sl_add: successfully inserted item %p at level 0", item, 0); - break; // success + // There was not an item in the skiplist that matches the key. + if (EXPECT_FALSE((int64_t)expectation > 0 || expectation == EXPECT_EXISTS)) { + TRACE("l1", "sl_cas: the expectation was not met, the skiplist was not changed", 0, 0); + return DOES_NOT_EXIST; // failure + } + + ASSERT(expectation == EXPECT_DOES_NOT_EXIST || expectation == EXPECT_WHATEVER); + + // First insert into the bottom level. + TRACE("s3", "sl_cas: attempting to insert item between %p and %p", preds[0], nexts[0]); + new_item = node_alloc(n, key_data, key_len, new_val); + node_t *pred = preds[0]; + 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_cas: successfully inserted item %p at level 0", new_item, 0); + break; // success + } + TRACE("s3", "sl_cas: failed to change pred's link: expected %p found %p", next, other); + node_free(new_item); + continue; } - TRACE("s3", "sl_add: failed to change pred's link: expected %p found %p", next, other); + // Found an item in the skiplist that matches the key. + uint64_t old_item_val = old_item->val; + do { + // If the item's value is DOES_NOT_EXIST it means another thread removed the node out from under us. + if (EXPECT_FALSE(old_item_val == DOES_NOT_EXIST)) { + TRACE("s2", "sl_cas: lost a race, found an item but another thread removed it. retry", 0, 0); + break; // retry + } + + if (EXPECT_FALSE(expectation == EXPECT_DOES_NOT_EXIST)) { + TRACE("s1", "sl_cas: found an item %p in the skiplist that matched the key. the expectation was " + "not met, the skiplist was not changed", old_item, old_item_val); + return old_item_val; // failure + } + + // Use a CAS and not a SWAP. If the node is in the process of being removed and we used a SWAP, we could + // replace DOES_NOT_EXIST with our value. Then another thread that is updating the value could think it + // succeeded and return our value even though we indicated that the node has been removed. If the CAS + // fails it means another thread either removed the node or updated its value. + uint64_t ret_val = SYNC_CAS(&old_item->val, old_item_val, new_val); + if (ret_val == old_item_val) { + TRACE("s1", "sl_cas: the CAS succeeded. updated the value of the item", 0, 0); + return ret_val; // success + } + TRACE("s2", "sl_cas: lost a race. the CAS failed. another thread changed the item's value", 0, 0); + + old_item_val = ret_val; + } while (1); } while (1); - // Insert into from the bottom level up. - for (int level = 1; level <= item->top_level; ++level) { + // Link into from the bottom 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_cas: 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_cas: successfully inserted item %p at level %llu", new_item, level); + break; // success + } + TRACE("s3", "sl_cas: 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 new_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 new_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("s1", "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; } @@ -263,36 +343,43 @@ uint64_t sl_remove (skiplist_t *sl, uint64_t key) { return DOES_NOT_EXIST; continue; } - node_t *next = SYNC_FETCH_AND_OR(&item->next[level], TAG); - if (EXPECT_FALSE(IS_TAGGED(next))) { - TRACE("s3", "sl_remove: lost race -- %p is already marked for removal by another thread", item, 0); - if (level == 0) - return DOES_NOT_EXIST; - continue; - } + node_t *next; + node_t *old_next = item->next[level]; + do { + next = old_next; + old_next = SYNC_CAS(&item->next[level], next, TAG_VALUE(next)); + if (IS_TAGGED(old_next)) { + TRACE("s2", "sl_remove: lost race -- %p is already marked for removal by another thread", item, 0); + if (level == 0) + return DOES_NOT_EXIST; + } + } while (!IS_TAGGED(old_next) || next != old_next); } - uint64_t value = item->value; + // This has to be an atomic swap in case another thread is updating the item while we are removing it. + uint64_t val = SYNC_SWAP(&item->val, DOES_NOT_EXIST); + TRACE("s2", "sl_remove: replaced item's val %p with DOES_NOT_EXIT", val, 0); - // 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]; node_t *next = item->next[level]; - TRACE("s3", "sl_remove: link item's pred %p to it's successor %p", pred, STRIP_TAG(next)); + TRACE("s2", "sl_remove: unlink the item by linking its 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", "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; + TRACE("s1", "sl_remove: unlink failed; pred's link changed from %p to %p", item, other); + return val; } --level; } // The thread that completes the unlink should free the memory. - nbd_defer_free(item); - return value; + TRACE("s1", "sl_remove: successfully unlinked item %p from the skiplist", item, 0); + node_defer_free(item); + return val; } void sl_print (skiplist_t *sl) { @@ -314,7 +401,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 *)item->key->data); + } if (item != sl->head) { printf("[%d]", item->top_level); } else { @@ -332,93 +424,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