From f5e3ab2d916513bea2bcf4bec757c671701c26f5 Mon Sep 17 00:00:00 2001 From: Sears Russell Date: Mon, 19 Oct 2009 21:55:25 +0000 Subject: [PATCH] concurrent hashtable seems to be working; still need to work out atomicity primitives (eg; options to leave the bucket locked on success) --- benchmarks/multicore/CMakeLists.txt | 1 + .../multicore/hashtableDifferentPages.c | 54 ++++++ src/stasis/CMakeLists.txt | 2 +- src/stasis/concurrentHash.c | 141 +++++++++++++++ stasis/concurrentHash.h | 24 +++ test/stasis/CMakeLists.txt | 3 +- test/stasis/check_concurrentHash.c | 161 ++++++++++++++++++ 7 files changed, 384 insertions(+), 2 deletions(-) create mode 100644 benchmarks/multicore/hashtableDifferentPages.c create mode 100644 src/stasis/concurrentHash.c create mode 100644 stasis/concurrentHash.h create mode 100644 test/stasis/check_concurrentHash.c diff --git a/benchmarks/multicore/CMakeLists.txt b/benchmarks/multicore/CMakeLists.txt index 6d78edd..986443c 100644 --- a/benchmarks/multicore/CMakeLists.txt +++ b/benchmarks/multicore/CMakeLists.txt @@ -4,6 +4,7 @@ CREATE_EXECUTABLE(smallLogEntry) CREATE_EXECUTABLE(noopTransactions) CREATE_EXECUTABLE(pinSamePage) CREATE_EXECUTABLE(pinDifferentPages) +CREATE_EXECUTABLE(hashtableDifferentPages) CREATE_EXECUTABLE(readLatch) CREATE_EXECUTABLE(readLatches) CREATE_EXECUTABLE(writeLatch) diff --git a/benchmarks/multicore/hashtableDifferentPages.c b/benchmarks/multicore/hashtableDifferentPages.c new file mode 100644 index 0000000..646e900 --- /dev/null +++ b/benchmarks/multicore/hashtableDifferentPages.c @@ -0,0 +1,54 @@ +/* + * hashtableDifferentPages.c + * + * Created on: Oct 19, 2009 + * Author: sears + */ +#include +#include +#include +#include + + +char * usage = "%s numthreads numops\n"; + +unsigned long numops; + +hashtable_t * ht; + +static void* worker(void* arg) { + pageid_t pid = *(pageid_t*)arg; + hashtable_insert(ht, pid, &pid); + for(unsigned long i = 0; i < numops; i++) { + void * ptr = hashtable_lookup(ht, pid); + assert(ptr == &pid); + // Page * p = loadPage(-1, pid); +// releasePage(p); + } + + return 0; +} + +int main(int argc, char * argv[]) { + if(argc != 3) { printf(usage, argv[0]); abort(); } + char * endptr; + unsigned long numthreads = strtoul(argv[1], &endptr, 10); + if(*endptr != 0) { printf(usage, argv[0]); abort(); } + numops= strtoul(argv[2], &endptr, 10) / numthreads; + if(*endptr != 0) { printf(usage, argv[0]); abort(); } + + pthread_t workers[numthreads]; + pageid_t pids[numthreads]; + + ht = hashtable_init(numthreads * 10, 0); + + for(int i = 0; i < numthreads; i++) { + pids[i] = i*2 ; + pthread_create(&workers[i], 0, worker, &pids[i]); + } + for(int i = 0; i < numthreads; i++) { + pthread_join(workers[i], 0); + } + + hashtable_deinit(ht); +} diff --git a/src/stasis/CMakeLists.txt b/src/stasis/CMakeLists.txt index 805616e..86cd904 100644 --- a/src/stasis/CMakeLists.txt +++ b/src/stasis/CMakeLists.txt @@ -1,4 +1,4 @@ -ADD_LIBRARY(stasis crc32.c redblack.c lhtable.c rw.c doubleLinkedList.c +ADD_LIBRARY(stasis crc32.c redblack.c lhtable.c concurrentHash.c rw.c doubleLinkedList.c common.c flags.c stats.c bufferManager.c linkedlist.c operations.c pageHandle.c bufferManager/legacy/pageFile.c diff --git a/src/stasis/concurrentHash.c b/src/stasis/concurrentHash.c new file mode 100644 index 0000000..8b8bdb4 --- /dev/null +++ b/src/stasis/concurrentHash.c @@ -0,0 +1,141 @@ +/* + * concurrenthash.c + * + * Created on: Oct 15, 2009 + * Author: sears + */ +#include +#include + +typedef struct bucket { + pageid_t key; + pthread_mutex_t mut; + void * val; +} bucket_t; + +typedef struct hashtable { + bucket_t* buckets; + pageid_t maxbucketid; + pageid_t numentries; + char tracknum; +} hashtable_t; + +hashtable_t * hashtable_init(pageid_t size, int tracknum) { + pageid_t newsize = 1; + for(int i = 0; size; i++) { + size /= 2; + newsize *= 2; + } + hashtable_t *ht = malloc(sizeof(*ht)); + + ht->maxbucketid = (newsize) - 1; + ht->buckets = calloc(ht->maxbucketid+1, sizeof(bucket_t)); + for(pageid_t i = 0; i <= ht->maxbucketid; i++) { + pthread_mutex_init(&ht->buckets[i].mut, 0); + } + + ht->numentries = 0; + ht->tracknum = (tracknum == 0 ? 0 : 1); + + return ht; +} +void hashtable_deinit(hashtable_t * ht) { + for(pageid_t i = 0; i < ht->maxbucketid; i++) { + pthread_mutex_destroy(&ht->buckets[i].mut); + } + free(ht->buckets); + free(ht); +} +static inline pageid_t hashtable_func(hashtable_t *ht, pageid_t p) { + return p & ht->maxbucketid; +} +typedef enum { + LOOKUP, + INSERT, + TRYINSERT, + REMOVE +} hashtable_mode; +static inline void * hashtable_op(hashtable_mode mode, hashtable_t *ht, pageid_t p, void *val) { + pageid_t idx = hashtable_func(ht, p); + void * ret; + bucket_t *b1 = &ht->buckets[idx], *b2 = NULL; + pthread_mutex_lock(&b1->mut); // start crabbing + + int num_incrs = 0; + + while(1) { + // Loop invariants: + // b1 is latched, b2 is unlatched + assert(num_incrs < (ht->maxbucketid/4)); + num_incrs++; + if(b1->key == p) { ret = b1->val; break; } + if(b1->val == NULL) { ret = NULL; break; } + idx = hashtable_func(ht, idx+1); + b2 = b1; + b1 = &ht->buckets[idx]; + pthread_mutex_lock(&b1->mut); + pthread_mutex_unlock(&b2->mut); + } + if(mode == INSERT || (mode == TRYINSERT && ret == NULL)) { + b1->key = p; + b1->val = val; + } else if(mode == REMOVE && ret != NULL) { + pageid_t idx2 = idx; + idx = hashtable_func(ht, idx+1); + b2 = b1; + b1 = &ht->buckets[idx]; + pthread_mutex_lock(&b1->mut); + while(1) { + // Loop invariants: b2 needs to be overwritten. + // b1 and b2 are latched + // b1 is the next bucket to consider for copying into b2. + + // What to do with b1? + // Case 1: It is null, we win. + pageid_t newidx = hashtable_func(ht, b1->key); + if(b1->val == NULL) { +// printf("d\n"); fflush(0); + b2->key = 0; + b2->val = NULL; + break; + // Case 2: b1 belongs "after" b2 + // Subcase 1: newidx is higher than idx2, so newidx should stay where it is. + // Subcase 2: newidx wrapped, so it is less than idx2, but more than half way around the ring. + } else if(idx2 < newidx || (idx2 > newidx + (ht->maxbucketid/2))) { + // skip this b1. +// printf("s\n"); fflush(0); + idx = hashtable_func(ht, idx+1); + pthread_mutex_unlock(&b1->mut); + b1 = &ht->buckets[idx]; + pthread_mutex_lock(&b1->mut); + // Case 3: we can compact b1 into b2's slot. + } else { +// printf("c %lld %lld %lld %lld\n", startidx, idx2, newidx, ht->maxbucketid); fflush(0); + b2->key = b1->key; + b2->val = b1->val; + pthread_mutex_unlock(&b2->mut); + // now we need to overwrite b1, so it is the new b2. + idx2 = idx; + idx = hashtable_func(ht, idx+1); + b2 = b1; + b1 = &ht->buckets[idx]; + pthread_mutex_lock(&b1->mut); + } + } + pthread_mutex_unlock(&b2->mut); + } + pthread_mutex_unlock(&b1->mut); // stop crabbing + return ret; +} +void * hashtable_insert(hashtable_t *ht, pageid_t p, void * val) { + return hashtable_op(INSERT, ht, p, val); +} +void * hashtable_test_and_set(hashtable_t *ht, pageid_t p, void * val) { + return hashtable_op(TRYINSERT, ht, p, val); +} +void * hashtable_lookup(hashtable_t *ht, pageid_t p) { + return hashtable_op(LOOKUP, ht, p, NULL); +} +void * hashtable_remove(hashtable_t *ht, pageid_t p) { + return hashtable_op(REMOVE, ht, p, NULL); +} diff --git a/stasis/concurrentHash.h b/stasis/concurrentHash.h new file mode 100644 index 0000000..d8f3b6f --- /dev/null +++ b/stasis/concurrentHash.h @@ -0,0 +1,24 @@ +/* + * concurrentHash.h + * + * Created on: Oct 15, 2009 + * Author: sears + */ + +#ifndef CONCURRENTHASH_H_ +#define CONCURRENTHASH_H_ +#include + +typedef struct hashtable_t hashtable_t; + +hashtable_t * hashtable_init(pageid_t size, int tracknum); +void hashtable_deinit(hashtable_t * ht); +void * hashtable_insert(hashtable_t *ht, pageid_t p, void * val); +/** Atomically insert a value if the key was not already defined + * @return NULL if val was inserted + */ +void * hashtable_test_and_set(hashtable_t *ht, pageid_t p, void * val); +void * hashtable_lookup(hashtable_t *ht, pageid_t p); +void * hashtable_remove(hashtable_t *ht, pageid_t p); + +#endif /* CONCURRENTHASH_H_ */ diff --git a/test/stasis/CMakeLists.txt b/test/stasis/CMakeLists.txt index 3ed784f..91b063d 100644 --- a/test/stasis/CMakeLists.txt +++ b/test/stasis/CMakeLists.txt @@ -1,4 +1,5 @@ SUBDIRS(fault_injection) +CREATE_CHECK(check_concurrentHash) CREATE_CHECK(check_lhtable) CREATE_CHECK(check_logEntry) CREATE_CHECK(check_logWriter) @@ -29,4 +30,4 @@ CREATE_CHECK(check_rangeTracker) CREATE_CHECK(check_replacementPolicy) CREATE_CHECK(check_lsmTree) CREATE_CHECK(check_groupBy) -CREATE_CHECK(check_boundedLog) \ No newline at end of file +CREATE_CHECK(check_boundedLog) diff --git a/test/stasis/check_concurrentHash.c b/test/stasis/check_concurrentHash.c new file mode 100644 index 0000000..1a09b05 --- /dev/null +++ b/test/stasis/check_concurrentHash.c @@ -0,0 +1,161 @@ +/* + * check_concurrentHash.c + * + * Created on: Oct 15, 2009 + * Author: sears + */ +/*--- +This software is copyrighted by the Regents of the University of +California, and other parties. The following terms apply to all files +associated with the software unless explicitly disclaimed in +individual files. + +The authors hereby grant permission to use, copy, modify, distribute, +and license this software and its documentation for any purpose, +provided that existing copyright notices are retained in all copies +and that this notice is included verbatim in any distributions. No +written agreement, license, or royalty fee is required for any of the +authorized uses. Modifications to this software may be copyrighted by +their authors and need not follow the licensing terms described here, +provided that the new terms are clearly indicated on the first page of +each file where they apply. + +IN NO EVENT SHALL THE AUTHORS OR DISTRIBUTORS BE LIABLE TO ANY PARTY +FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES +ARISING OUT OF THE USE OF THIS SOFTWARE, ITS DOCUMENTATION, OR ANY +DERIVATIVES THEREOF, EVEN IF THE AUTHORS HAVE BEEN ADVISED OF THE +POSSIBILITY OF SUCH DAMAGE. + +THE AUTHORS AND DISTRIBUTORS SPECIFICALLY DISCLAIM ANY WARRANTIES, +INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, AND +NON-INFRINGEMENT. THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, AND +THE AUTHORS AND DISTRIBUTORS HAVE NO OBLIGATION TO PROVIDE +MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS. + +GOVERNMENT USE: If you are acquiring this software on behalf of the +U.S. government, the Government shall have only "Restricted Rights" in +the software and related documentation as defined in the Federal +Acquisition Regulations (FARs) in Clause 52.227.19 (c) (2). If you are +acquiring the software on behalf of the Department of Defense, the +software shall be classified as "Commercial Computer Software" and the +Government shall have only "Restricted Rights" as defined in Clause +252.227-7013 (c) (1) of DFARs. Notwithstanding the foregoing, the +authors grant the U.S. Government and others acting in its behalf +permission to use and distribute the software in accordance with the +terms specified in this license. +---*/ + +#define _GNU_SOURCE +#include "../check_includes.h" + +#include + +#include +#include +#include +#include +#include +#include + +#define LOG_NAME "check_lhtable.log" + + +#define NUM_OPS 100000000 +#define NUM_ENTRIES 10000 +#define NUM_THREADS 100 +#define THREAD_ENTRIES ((NUM_ENTRIES/NUM_THREADS)-1) + +hashtable_t * ht; + +void * worker(void * arg) { + int stride = *(int*) arg; + + pageid_t *data = malloc(sizeof(pageid_t) * THREAD_ENTRIES); + + for(int i = 1; i <= THREAD_ENTRIES; i++) { + data[i-1] = -1 * (stride + (i * NUM_THREADS)); + } + for(int j = 0; j < NUM_OPS/ NUM_THREADS; j++) { + int op = myrandom(2); + + int i = myrandom(THREAD_ENTRIES); + + pageid_t scratch = data[i]; + if(data[i] < 0) { + scratch *= -1; + } + switch(op) { + case 0: { + void * ret; + if(data[i] < 0) { + ret = hashtable_insert(ht, scratch, &data[i]); + assert(ret == NULL); + data[i] *= -1; + } else { + ret = hashtable_remove(ht, scratch); + assert(ret == &data[i]); + data[i] *= -1; + } + } break; + case 1: { + void * ret = hashtable_lookup(ht, scratch); + if(data[i] < 0) { + assert(ret == NULL); + } else { + assert(ret == &data[i]); + } + } break; + default: + abort(); + } + } + free(data); + return 0; +} + +START_TEST(singleThreadHashTest) { + ht = hashtable_init((pageid_t)((double)THREAD_ENTRIES * 1.1), 0); + int i = 0; + worker(&i); + hashtable_deinit(ht); +} END_TEST + +START_TEST(concurrentHashTest) { + ht = hashtable_init((pageid_t)((double)NUM_ENTRIES * 1.1), 0); + pthread_t workers[NUM_THREADS]; + for(int i = 0 ; i < NUM_THREADS; i++) { + int * ip = malloc(sizeof(int)); + *ip = i; + pthread_create(&workers[i], 0, worker, ip); + } + for(int i = 0 ; i < NUM_THREADS; i++) { + pthread_join(workers[i],0); + } + hashtable_deinit(ht); +} END_TEST + +Suite * check_suite(void) { + Suite *s = suite_create("lhtable"); + /* Begin a new test */ + TCase *tc = tcase_create("lhtable"); + + tcase_set_timeout(tc, 0); // disable timeouts + + srandom(43); + + /* Sub tests are added, one per line, here */ + tcase_add_test(tc, singleThreadHashTest); + tcase_add_test(tc, concurrentHashTest); + + /* --------------------------------------------- */ + + tcase_add_checked_fixture(tc, setup, teardown); + + + suite_add_tcase(s, tc); + return s; +} + +#include "../check_setup.h" +