X-Git-Url: https://pd.if.org/git/?p=nbds;a=blobdiff_plain;f=map%2Fhashtable.c;h=b4d7a5de3976db050fa9cf89a6c84b833b79fe24;hp=1039ff014bf19d3deb246fcf98c79a6aef8f81f8;hb=f3a053a46bbb4ba460bcff0920b93dfc8263e02e;hpb=f1098084dd54496a61f9a254541190df77edd166 diff --git a/map/hashtable.c b/map/hashtable.c index 1039ff0..b4d7a5d 100644 --- a/map/hashtable.c +++ b/map/hashtable.c @@ -16,11 +16,11 @@ #include "common.h" #include "murmur.h" #include "mem.h" -#include "mlocal.h" +#include "hashtable.h" #define GET_PTR(x) ((void *)((x) & MASK(48))) // low-order 48 bits is a pointer to a nstring_t -typedef struct ht_entry { +typedef struct entry { uint64_t key; uint64_t val; } entry_t; @@ -29,29 +29,28 @@ typedef struct hti { volatile entry_t *table; hashtable_t *ht; // parent ht; struct hti *next; - unsigned int scale; + unsigned scale; int max_probe; + int references; int count; // TODO: make these counters distributed int num_entries_copied; - int scan; + int copy_scan; } hti_t; -struct ht { - hti_t *hti; - cmp_fun_t cmp_fun; - hash_fun_t hash_fun; - clone_fun_t clone_fun; +struct ht_iter { + hti_t * hti; + int64_t idx; + uint64_t key; + uint64_t val; }; -static const map_impl_t ht_map_impl = { - (map_alloc_t)ht_alloc, (map_cas_t)ht_cas, (map_get_t)ht_get, (map_remove_t)ht_remove, - (map_count_t)ht_count, (map_print_t)ht_print, (map_free_t)ht_free +struct ht { + hti_t *hti; + const datatype_t *key_type; }; -const map_impl_t *MAP_TYPE_HASHTABLE = &ht_map_impl; - static const uint64_t COPIED_VALUE = -1; -static const uint64_t TOMBSTONE = STRIP_TAG(-1); +static const uint64_t TOMBSTONE = STRIP_TAG(-1, TAG1); static const unsigned ENTRIES_PER_BUCKET = CACHE_LINE_SIZE/sizeof(entry_t); static const unsigned ENTRIES_PER_COPY_CHUNK = CACHE_LINE_SIZE/sizeof(entry_t)*2; @@ -93,13 +92,13 @@ static volatile entry_t *hti_lookup (hti_t *hti, void *key, uint32_t key_hash, i uint64_t ent_key = ent->key; if (ent_key == DOES_NOT_EXIST) { TRACE("h1", "hti_lookup: entry %p for key %p is empty", ent, - (hti->ht->clone_fun == NULL) ? (void *)ent_key : GET_PTR(ent_key)); + (hti->ht->key_type == NULL) ? (void *)ent_key : GET_PTR(ent_key)); *is_empty = 1; // indicate an empty so the caller avoids an expensive key compare return ent; } // Compare with the key in the entry. - if (EXPECT_TRUE(hti->ht->cmp_fun == NULL)) { + if (EXPECT_TRUE(hti->ht->key_type == NULL)) { // fast path for integer keys if (ent_key == (uint64_t)key) { TRACE("h1", "hti_lookup: found entry %p with key %p", ent, ent_key); @@ -109,7 +108,7 @@ static volatile entry_t *hti_lookup (hti_t *hti, void *key, uint32_t key_hash, i // The key in is made up of two parts. The 48 low-order bits are a pointer. The // high-order 16 bits are taken from the hash. The bits from the hash are used as a // quick check to rule out non-equal keys without doing a complete compare. - if ((key_hash >> 16) == (ent_key >> 48) && hti->ht->cmp_fun(GET_PTR(ent_key), key) == 0) { + if ((key_hash >> 16) == (ent_key >> 48) && hti->ht->key_type->cmp(GET_PTR(ent_key), key) == 0) { TRACE("h1", "hti_lookup: found entry %p with key %p", ent, GET_PTR(ent_key)); return ent; } @@ -126,15 +125,13 @@ static volatile entry_t *hti_lookup (hti_t *hti, void *key, uint32_t key_hash, i // Allocate and initialize a hti_t with 2^ entries. static hti_t *hti_alloc (hashtable_t *parent, int scale) { - // Include enough slop to align the actual table on a cache line boundry - size_t n = sizeof(hti_t) - + sizeof(entry_t) * (1 << scale) - + (CACHE_LINE_SIZE - 1); - hti_t *hti = (hti_t *)calloc(n, 1); + hti_t *hti = (hti_t *)nbd_malloc(sizeof(hti_t)); + memset(hti, 0, sizeof(hti_t)); - // Align the table of hash entries on a cache line boundry. - hti->table = (entry_t *)(((uint64_t)hti + sizeof(hti_t) + (CACHE_LINE_SIZE-1)) - & ~(CACHE_LINE_SIZE-1)); + size_t sz = sizeof(entry_t) * (1 << scale); + entry_t *table = nbd_malloc(sz); + memset(table, 0, sz); + hti->table = table; hti->scale = scale; @@ -187,7 +184,7 @@ static int hti_copy_entry (hti_t *ht1, volatile entry_t *ht1_ent, uint32_t key_h assert(ht1->next); assert(ht2); assert(ht1_ent >= ht1->table && ht1_ent < ht1->table + (1 << ht1->scale)); - assert(key_hash == 0 || ht1->ht->hash_fun == NULL || (key_hash >> 16) == (ht1_ent->key >> 48)); + assert(key_hash == 0 || ht1->ht->key_type == NULL || (key_hash >> 16) == (ht1_ent->key >> 48)); uint64_t ht1_ent_val = ht1_ent->val; if (EXPECT_FALSE(ht1_ent_val == COPIED_VALUE)) { @@ -210,7 +207,7 @@ static int hti_copy_entry (hti_t *ht1, volatile entry_t *ht1_ent, uint32_t key_h } // Tag the value in the old entry to indicate a copy is in progress. - ht1_ent_val = SYNC_FETCH_AND_OR(&ht1_ent->val, TAG_VALUE(0)); + ht1_ent_val = SYNC_FETCH_AND_OR(&ht1_ent->val, TAG_VALUE(0, TAG1)); TRACE("h2", "hti_copy_entry: tagged the value %p in old entry %p", ht1_ent_val, ht1_ent); if (ht1_ent_val == COPIED_VALUE) { TRACE("h1", "hti_copy_entry: entry %p already copied to table %p", ht1_ent, ht2); @@ -219,13 +216,13 @@ static int hti_copy_entry (hti_t *ht1, volatile entry_t *ht1_ent, uint32_t key_h // Install the key in the new table. uint64_t ht1_ent_key = ht1_ent->key; - void *key = (ht1->ht->hash_fun == NULL) ? (void *)ht1_ent_key : GET_PTR(ht1_ent_key); + void *key = (ht1->ht->key_type == NULL) ? (void *)ht1_ent_key : GET_PTR(ht1_ent_key); // The old table's dead entries don't need to be copied to the new table, but their keys need to be freed. - assert(COPIED_VALUE == TAG_VALUE(TOMBSTONE)); + assert(COPIED_VALUE == TAG_VALUE(TOMBSTONE, TAG1)); if (ht1_ent_val == TOMBSTONE) { TRACE("h1", "hti_copy_entry: entry %p old value was deleted, now freeing key %p", ht1_ent, key); - if (EXPECT_FALSE(ht1->ht->clone_fun != NULL)) { + if (EXPECT_FALSE(ht1->ht->key_type != NULL)) { nbd_defer_free(key); } return TRUE; @@ -234,7 +231,7 @@ static int hti_copy_entry (hti_t *ht1, volatile entry_t *ht1_ent, uint32_t key_h // We use 0 to indicate that is uninitiallized. Occasionally the key's hash will really be 0 and we // waste time recomputing it every time. It is rare enough (1 in 65k) that it won't hurt performance. if (key_hash == 0) { - key_hash = (ht1->ht->hash_fun == NULL) ? murmur32_8b(ht1_ent_key) : ht1->ht->hash_fun(key); + key_hash = (ht1->ht->key_type == NULL) ? murmur32_8b(ht1_ent_key) : ht1->ht->key_type->hash(key); } int ht2_ent_is_empty; @@ -260,7 +257,7 @@ static int hti_copy_entry (hti_t *ht1, volatile entry_t *ht1_ent, uint32_t key_h } // Copy the value to the entry in the new table. - ht1_ent_val = STRIP_TAG(ht1_ent_val); + ht1_ent_val = STRIP_TAG(ht1_ent_val, TAG1); uint64_t old_ht2_ent_val = SYNC_CAS(&ht2_ent->val, DOES_NOT_EXIST, ht1_ent_val); // If there is a nested copy in progress, we might have installed the key into a dead entry. @@ -286,7 +283,7 @@ static int hti_copy_entry (hti_t *ht1, volatile entry_t *ht1_ent, uint32_t key_h } // Compare with the existing value associated with . If the values match then -// replace the existing value with . If is TOMBSTONE, delete the value associated with +// replace the existing value with . If is DOES_NOT_EXIST, delete the value associated with // the key by replacing it with a TOMBSTONE. // // Return the previous value associated with , or DOES_NOT_EXIST if is not in the table @@ -303,7 +300,7 @@ static uint64_t hti_cas (hti_t *hti, void *key, uint32_t key_hash, uint64_t expe TRACE("h1", "hti_cas: hti %p key %p", hti, key); TRACE("h1", "hti_cas: value %p expect %p", new, expected); assert(hti); - assert(new != DOES_NOT_EXIST && !IS_TAGGED(new)); + assert(!IS_TAGGED(new, TAG1)); assert(key); int is_empty; @@ -324,12 +321,12 @@ static uint64_t hti_cas (hti_t *hti, void *key, uint32_t key_hash, uint64_t expe return DOES_NOT_EXIST; // No need to do anything, is already deleted. - if (new == TOMBSTONE) + if (new == DOES_NOT_EXIST) return DOES_NOT_EXIST; // Allocate . - uint64_t new_key = (uint64_t)((hti->ht->clone_fun == NULL) ? key : hti->ht->clone_fun(key)); - if (EXPECT_FALSE(hti->ht->hash_fun != NULL)) { + uint64_t new_key = (uint64_t)((hti->ht->key_type == NULL) ? key : hti->ht->key_type->clone(key)); + if (EXPECT_FALSE(hti->ht->key_type != NULL)) { // Combine pointer with bits from its hash new_key = ((uint64_t)(key_hash >> 16) << 48) | new_key; } @@ -341,8 +338,8 @@ static uint64_t hti_cas (hti_t *hti, void *key, uint32_t key_hash, uint64_t expe if (old_ent_key != DOES_NOT_EXIST) { TRACE("h0", "hti_cas: lost race to install key %p in entry %p", new_key, ent); TRACE("h0", "hti_cas: found %p instead of NULL", - (hti->ht->clone_fun != NULL) ? (void *)old_ent_key : GET_PTR(old_ent_key), 0); - if (hti->ht->clone_fun != NULL) { + (hti->ht->key_type == NULL) ? (void *)old_ent_key : GET_PTR(old_ent_key), 0); + if (hti->ht->key_type != NULL) { nbd_free(GET_PTR(new_key)); } return hti_cas(hti, key, key_hash, expected, new); // tail-call @@ -351,11 +348,11 @@ static uint64_t hti_cas (hti_t *hti, void *key, uint32_t key_hash, uint64_t expe } TRACE("h0", "hti_cas: entry for key %p is %p", - (hti->ht->clone_fun != NULL) ? (void *)ent->key : GET_PTR(ent->key), ent); + (hti->ht->key_type == NULL) ? (void *)ent->key : GET_PTR(ent->key), ent); // If the entry is in the middle of a copy, the copy must be completed first. uint64_t ent_val = ent->val; - if (EXPECT_FALSE(IS_TAGGED(ent_val))) { + if (EXPECT_FALSE(IS_TAGGED(ent_val, TAG1))) { if (ent_val != COPIED_VALUE) { int did_copy = hti_copy_entry(hti, ent, key_hash, ((volatile hti_t *)hti)->next); if (did_copy) { @@ -379,22 +376,22 @@ static uint64_t hti_cas (hti_t *hti, void *key, uint32_t key_hash, uint64_t expe } // No need to update if value is unchanged. - if ((new == TOMBSTONE && !old_existed) || ent_val == new) { + if ((new == DOES_NOT_EXIST && !old_existed) || ent_val == new) { TRACE("h1", "hti_cas: old value and new value were the same", 0, 0); return ent_val; } // CAS the value into the entry. Retry if it fails. - uint64_t v = SYNC_CAS(&ent->val, ent_val, new); + uint64_t v = SYNC_CAS(&ent->val, ent_val, new == DOES_NOT_EXIST ? TOMBSTONE : new); if (EXPECT_FALSE(v != ent_val)) { TRACE("h0", "hti_cas: value CAS failed; expected %p found %p", ent_val, v); return hti_cas(hti, key, key_hash, expected, new); // recursive tail-call } // The set succeeded. Adjust the value count. - if (old_existed && new == TOMBSTONE) { + if (old_existed && new == DOES_NOT_EXIST) { SYNC_ADD(&hti->count, -1); - } else if (!old_existed && new != TOMBSTONE) { + } else if (!old_existed && new != DOES_NOT_EXIST) { SYNC_ADD(&hti->count, 1); } @@ -422,7 +419,7 @@ static uint64_t hti_get (hti_t *hti, void *key, uint32_t key_hash) { // If the entry is being copied, finish the copy and retry on the next table. uint64_t ent_val = ent->val; - if (EXPECT_FALSE(IS_TAGGED(ent_val))) { + if (EXPECT_FALSE(IS_TAGGED(ent_val, TAG1))) { if (EXPECT_FALSE(ent_val != COPIED_VALUE)) { int did_copy = hti_copy_entry(hti, ent, key_hash, ((volatile hti_t *)hti)->next); if (did_copy) { @@ -437,28 +434,20 @@ static uint64_t hti_get (hti_t *hti, void *key, uint32_t key_hash) { // uint64_t ht_get (hashtable_t *ht, void *key) { - uint32_t hash = (ht->hash_fun == NULL) ? murmur32_8b((uint64_t)key) : ht->hash_fun(key); + uint32_t hash = (ht->key_type == NULL) ? murmur32_8b((uint64_t)key) : ht->key_type->hash(key); return hti_get(ht->hti, key, hash); } -// -uint64_t ht_cas (hashtable_t *ht, void *key, uint64_t expected_val, uint64_t new_val) { - - TRACE("h2", "ht_cas: key %p ht %p", key, ht); - TRACE("h2", "ht_cas: expected val %p new val %p", expected_val, new_val); - assert(key != DOES_NOT_EXIST); - assert(!IS_TAGGED(new_val) && new_val != DOES_NOT_EXIST && new_val != TOMBSTONE); - - hti_t *hti = ht->hti; - - // Help with an ongoing copy. - if (EXPECT_FALSE(hti->next != NULL)) { - volatile entry_t *ent; - uint64_t limit; - int num_copied = 0; - int x = hti->scan; +// returns TRUE if copy is done +int hti_help_copy (hti_t *hti) { + volatile entry_t *ent; + uint64_t limit; + uint64_t total_copied = hti->num_entries_copied; + int num_copied = 0; + int x = hti->copy_scan; - TRACE("h1", "ht_cas: help copy. scan is %llu, size is %llu", x, 1<scale); + TRACE("h1", "ht_cas: help copy. scan is %llu, size is %llu", x, 1<scale); + if (total_copied == (1 << hti->scale)) { // Panic if we've been around the array twice and still haven't finished the copy. int panic = (x >= (1 << (hti->scale + 1))); if (!panic) { @@ -466,9 +455,9 @@ uint64_t ht_cas (hashtable_t *ht, void *key, uint64_t expected_val, uint64_t new // Reserve some entries for this thread to copy. There is a race condition here because the // fetch and add isn't atomic, but that is ok. - hti->scan = x + ENTRIES_PER_COPY_CHUNK; + hti->copy_scan = x + ENTRIES_PER_COPY_CHUNK; - // scan> might be larger than the size of the table, if some thread stalls while + // might be larger than the size of the table, if some thread stalls while // copying. In that case we just wrap around to the begining and make another pass through // the table. ent = hti->table + (x & MASK(hti->scale)); @@ -485,20 +474,43 @@ uint64_t ht_cas (hashtable_t *ht, void *key, uint64_t expected_val, uint64_t new assert(ent <= hti->table + (1 << hti->scale)); } if (num_copied != 0) { - SYNC_ADD(&hti->num_entries_copied, num_copied); + total_copied = SYNC_ADD(&hti->num_entries_copied, num_copied); } + } + + return (total_copied == (1 << hti->scale)); +} + +// +uint64_t ht_cas (hashtable_t *ht, void *key, uint64_t expected_val, uint64_t new_val) { + + TRACE("h2", "ht_cas: key %p ht %p", key, ht); + TRACE("h2", "ht_cas: expected val %p new val %p", expected_val, new_val); + assert(key != DOES_NOT_EXIST); + assert(!IS_TAGGED(new_val, TAG1) && new_val != DOES_NOT_EXIST && new_val != TOMBSTONE); + + hti_t *hti = ht->hti; + + // Help with an ongoing copy. + if (EXPECT_FALSE(hti->next != NULL)) { + int done = hti_help_copy(hti); // Dispose of fully copied tables. - if (hti->num_entries_copied == (1 << hti->scale) || panic) { - assert(hti->next); - if (SYNC_CAS(&ht->hti, hti, hti->next) == hti) { - nbd_defer_free(hti); + if (done && hti->references == 0) { + + int r = SYNC_CAS(&hti->references, 0, -1); + if (r == 0) { + assert(hti->next); + if (SYNC_CAS(&ht->hti, hti, hti->next) == hti) { + nbd_defer_free((void *)hti->table); + nbd_defer_free(hti); + } } } } uint64_t old_val; - uint32_t key_hash = (ht->hash_fun == NULL) ? murmur32_8b((uint64_t)key) : ht->hash_fun(key); + uint32_t key_hash = (ht->key_type == NULL) ? murmur32_8b((uint64_t)key) : ht->key_type->hash(key); while ((old_val = hti_cas(hti, key, key_hash, expected_val, new_val)) == COPIED_VALUE) { assert(hti->next); hti = hti->next; @@ -512,9 +524,9 @@ uint64_t ht_cas (hashtable_t *ht, void *key, uint64_t expected_val, uint64_t new uint64_t ht_remove (hashtable_t *ht, void *key) { hti_t *hti = ht->hti; uint64_t val; - uint32_t key_hash = (ht->hash_fun == NULL) ? murmur32_8b((uint64_t)key) : ht->hash_fun(key); + uint32_t key_hash = (ht->key_type == NULL) ? murmur32_8b((uint64_t)key) : ht->key_type->hash(key); do { - val = hti_cas(hti, key, key_hash, CAS_EXPECT_WHATEVER, TOMBSTONE); + val = hti_cas(hti, key, key_hash, CAS_EXPECT_WHATEVER, DOES_NOT_EXIST); if (val != COPIED_VALUE) return val == TOMBSTONE ? DOES_NOT_EXIST : val; assert(hti->next); @@ -535,11 +547,9 @@ uint64_t ht_count (hashtable_t *ht) { } // Allocate and initialize a new hash table. -hashtable_t *ht_alloc (cmp_fun_t cmp_fun, hash_fun_t hash_fun, clone_fun_t clone_fun) { +hashtable_t *ht_alloc (const datatype_t *key_type) { hashtable_t *ht = nbd_malloc(sizeof(hashtable_t)); - ht->cmp_fun = cmp_fun; - ht->hash_fun = hash_fun; - ht->clone_fun = clone_fun; + ht->key_type = key_type; ht->hti = (hti_t *)hti_alloc(ht, MIN_SCALE); return ht; } @@ -549,12 +559,13 @@ void ht_free (hashtable_t *ht) { hti_t *hti = ht->hti; do { for (uint32_t i = 0; i < (1 << hti->scale); ++i) { - assert(hti->table[i].val == COPIED_VALUE || !IS_TAGGED(hti->table[i].val)); - if (ht->clone_fun != NULL && hti->table[i].key != DOES_NOT_EXIST) { + assert(hti->table[i].val == COPIED_VALUE || !IS_TAGGED(hti->table[i].val, TAG1)); + if (ht->key_type != NULL && hti->table[i].key != DOES_NOT_EXIST) { nbd_free(GET_PTR(hti->table[i].key)); } } hti_t *next = hti->next; + nbd_free((void *)hti->table); nbd_free(hti); hti = next; } while (hti); @@ -576,3 +587,70 @@ void ht_print (hashtable_t *ht) { hti = hti->next; } } + +ht_iter_t *ht_iter_start (hashtable_t *ht, void *key) { + hti_t *hti = ht->hti; + int rcount; + do { + while (((volatile hti_t *)hti)->next != NULL) { + do { } while (hti_help_copy(hti) != TRUE); + hti = hti->next; + } + + int old = hti->references; + do { + rcount = old; + if (rcount != -1) { + old = SYNC_CAS(&hti->references, rcount, rcount + 1); + } + } while (rcount != old); + } while (rcount == -1); + + ht_iter_t *iter = nbd_malloc(sizeof(ht_iter_t)); + iter->hti = hti; + iter->idx = -1; + + return iter; +} + +ht_iter_t *ht_iter_next (ht_iter_t *iter) { + volatile entry_t *ent; + uint64_t key; + uint64_t val; + uint64_t table_size = (1 << iter->hti->scale); + do { + if (++iter->idx == table_size) { + ht_iter_free(iter); + return NULL; + } + ent = &iter->hti->table[++iter->idx]; + key = ent->key; + val = ent->val; + + } while (key == DOES_NOT_EXIST || val == DOES_NOT_EXIST || val == TOMBSTONE); + + iter->key = key; + if (val == COPIED_VALUE) { + uint32_t hash = (iter->hti->ht->key_type == NULL) + ? murmur32_8b(key) + : iter->hti->ht->key_type->hash((void *)key); + iter->val = hti_get(iter->hti->next, (void *)ent->key, hash); + } else { + iter->val = val; + } + + return iter; +} + +uint64_t ht_iter_val (ht_iter_t *iter) { + return iter->val; +} + +uint64_t ht_iter_key (ht_iter_t *iter) { + return iter->key; +} + +void ht_iter_free (ht_iter_t *iter) { + SYNC_ADD(&iter->hti->references, -1); +} +