From 4df86fe8a036d271bcb03392f65d8bf815969d73 Mon Sep 17 00:00:00 2001 From: Date: Sun, 20 Apr 2008 13:39:49 -0700 Subject: [PATCH] Temp commit to test with valgrind on linux (memory issue) --- libmemcached/common.h | 1 + libmemcached/memcached.h | 7 ++ libmemcached/memcached_constants.h | 5 ++ libmemcached/memcached_hash.c | 117 ++++++++++++++++++++++++++++- libmemcached/memcached_hosts.c | 88 +++++++++++++++++++++- 5 files changed, 213 insertions(+), 5 deletions(-) diff --git a/libmemcached/common.h b/libmemcached/common.h index de55dbbe..44c50606 100644 --- a/libmemcached/common.h +++ b/libmemcached/common.h @@ -111,4 +111,5 @@ memcached_return memcachd_key_test(char **keys, size_t *key_length, void sort_hosts(memcached_st *ptr); +uint32_t generate_hash(memcached_st *ptr, char *key, size_t key_length); #endif /* __COMMON_H__ */ diff --git a/libmemcached/memcached.h b/libmemcached/memcached.h index 2b547184..99c942d8 100644 --- a/libmemcached/memcached.h +++ b/libmemcached/memcached.h @@ -33,6 +33,11 @@ extern "C" { /* string value */ #define LIBMEMCACHED_VERSION_STRING "0.19" +struct continuum_item{ + uint32_t index; + uint32_t value; +}; + struct memcached_stat_st { uint32_t pid; uint32_t uptime; @@ -77,6 +82,7 @@ struct memcached_st { memcached_server_distribution distribution; void *user_data; unsigned int wheel[MEMCACHED_WHEEL_SIZE]; + struct continuum_item continuum[MEMCACHED_CONTINUUM_SIZE]; memcached_clone_func on_clone; memcached_cleanup_func on_cleanup; memcached_free_function call_free; @@ -90,6 +96,7 @@ struct memcached_st { #endif }; + /* Public API */ const char * memcached_lib_version(void); diff --git a/libmemcached/memcached_constants.h b/libmemcached/memcached_constants.h index 254e9116..b2b819f2 100644 --- a/libmemcached/memcached_constants.h +++ b/libmemcached/memcached_constants.h @@ -18,7 +18,10 @@ extern "C" { #define MEMCACHED_MAX_KEY 251 /* We add one to have it null terminated */ #define MEMCACHED_MAX_BUFFER 8196 #define MEMCACHED_MAX_HOST_LENGTH 64 +#define MEMCACHED_MAX_HOST_SORT_LENGTH 86 /* Used for Ketama */ #define MEMCACHED_WHEEL_SIZE 1024 +#define MEMCACHED_CONTINUUM_SIZE 2048 +#define MEMCACHED_POINTS_PER_SERVER 100 #define MEMCACHED_STRIDE 4 #define MEMCACHED_DEFAULT_TIMEOUT INT32_MAX @@ -62,6 +65,8 @@ typedef enum { typedef enum { MEMCACHED_DISTRIBUTION_MODULA, MEMCACHED_DISTRIBUTION_CONSISTENT, + MEMCACHED_DISTRIBUTION_CONSISTENT_WHEEL, + MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA, } memcached_server_distribution; typedef enum { diff --git a/libmemcached/memcached_hash.c b/libmemcached/memcached_hash.c index f365cc0e..8d269faa 100644 --- a/libmemcached/memcached_hash.c +++ b/libmemcached/memcached_hash.c @@ -12,10 +12,11 @@ static uint32_t internal_generate_hash(char *key, size_t key_length); static uint32_t internal_generate_md5(char *key, size_t key_length); static uint32_t internal_generate_ketama_md5(char *key, size_t key_length); -unsigned int memcached_generate_hash(memcached_st *ptr, char *key, size_t key_length) +uint32_t generate_hash(memcached_st *ptr, char *key, size_t key_length) { uint32_t hash= 1; /* Just here to remove compile warning */ - unsigned int x; + uint32_t x= 0; + WATCHPOINT_ASSERT(ptr->number_of_hosts); @@ -98,12 +99,70 @@ unsigned int memcached_generate_hash(memcached_st *ptr, char *key, size_t key_le } WATCHPOINT_ASSERT(hash); + return hash; +} + +unsigned int dispatch_host(memcached_st *ptr, uint32_t hash) +{ + switch (ptr->distribution) + { + case MEMCACHED_DISTRIBUTION_CONSISTENT: + case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA: + { + int num= ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER; + + hash= hash; + struct continuum_item *begin, *end, *left, *right, *middle; + begin= left= ptr->continuum; + end= right= ptr->continuum + (num - 1); + + while (1) + { + struct continuum_item *rmiddle; + + middle = left + (right - left) / 2; + + if (middle==end) + return begin->index; + + if (middle==begin) + return end->index; + + rmiddle = middle+1; + + if (hashvalue && hash>=middle->value) + return middle->index; + + if (middle->value < hash) + left = middle + 1; + else if (middle->value > hash) + right = middle - 1; + + if (left>right) + return left->index; + } + } + break; + case MEMCACHED_DISTRIBUTION_CONSISTENT_WHEEL: + { + unsigned int server_key; + + server_key= hash % MEMCACHED_WHEEL_SIZE; + + return ptr->wheel[server_key]; + } + case MEMCACHED_DISTRIBUTION_MODULA: + return hash % ptr->number_of_hosts; + default: + WATCHPOINT_ASSERT(0); /* We have added a distribution without extending the logic */ + return hash % ptr->number_of_hosts; + } if (ptr->distribution == MEMCACHED_DISTRIBUTION_MODULA) { return hash % ptr->number_of_hosts; } - else + else if (ptr->distribution == MEMCACHED_DISTRIBUTION_CONSISTENT) { unsigned int server_key; @@ -111,6 +170,58 @@ unsigned int memcached_generate_hash(memcached_st *ptr, char *key, size_t key_le return ptr->wheel[server_key]; } + else if (ptr->distribution == MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA) + { + int num = ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER; + + hash = hash; + struct continuum_item *begin, *end, *left, *right, *middle; + begin = left = ptr->continuum; + end = right = ptr->continuum + (num - 1); + while(1) + { + middle = left + (right - left) / 2; + if(middle==end) + { + return begin->index; + } + if(middle==begin) + { + return end->index; + } + struct continuum_item *rmiddle = middle+1; + if(hashvalue && hash>=middle->value) + return middle->index; + + if(middle->value < hash) { + left = middle + 1; + }else if(middle->value > hash) { + right = middle - 1; + } + + if (left>right) + return left->index; + } + } + else + { + WATCHPOINT_ASSERT(0); + } +} + +unsigned int memcached_generate_hash(memcached_st *ptr, char *key, size_t key_length) +{ + uint32_t hash= 1; /* Just here to remove compile warning */ + + WATCHPOINT_ASSERT(ptr->number_of_hosts); + + if (ptr->number_of_hosts == 1) + return 0; + + hash = generate_hash(ptr, key, key_length); + + WATCHPOINT_ASSERT(hash); + return dispatch_host(ptr, hash); } static uint32_t internal_generate_hash(char *key, size_t key_length) diff --git a/libmemcached/memcached_hosts.c b/libmemcached/memcached_hosts.c index 76fd5a01..65dcc546 100644 --- a/libmemcached/memcached_hosts.c +++ b/libmemcached/memcached_hosts.c @@ -91,6 +91,64 @@ void server_list_free(memcached_st *ptr, memcached_server_st *servers) free(servers); } +static int continuum_item_cmp(const void *t1, const void *t2) +{ + struct continuum_item *ct1 = (struct continuum_item *)t1; + struct continuum_item *ct2 = (struct continuum_item *)t2; + if(ct1->value == ct2->value) + return 0; + else if(ct1->value > ct2->value) + return 1; + else + return -1; +} + +static uint32_t internal_generate_ketama_md5(char *key, size_t key_length) +{ + unsigned char results[16]; + + md5_signature((unsigned char*)key, (unsigned int)key_length, results); + + return ( (results[3] ) << 24) + | ( (results[2] ) << 16) + | ( (results[1] ) << 8) + | ( results[0] ); +} + +void update_continuum(memcached_st *ptr) +{ + int index; + int host_index; + int continuum_index= 0; + int value; + memcached_server_st *list = ptr->hosts; + + for (host_index = 0; host_index < ptr->number_of_hosts; ++host_index) + { + for(index= 1; index <= MEMCACHED_POINTS_PER_SERVER; ++index) + { + char sort_host[MEMCACHED_MAX_HOST_SORT_LENGTH]= ""; + size_t sort_host_length; + + sort_host_length= snprintf(sort_host, MEMCACHED_MAX_HOST_SORT_LENGTH, "%s:%d-%d", + list[host_index].hostname, list[host_index].port, index); + WATCHPOINT_ASSERT(sort_host_length); + value= internal_generate_ketama_md5(sort_host, sort_host_length); + ptr->continuum[continuum_index].index= host_index; + ptr->continuum[continuum_index++].value= value; + } + } + + qsort(ptr->continuum, ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER, sizeof(struct continuum_item), continuum_item_cmp); +#ifdef HAVE_DEBUG + for(index= 0; index < ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER - 1; ++index) + { + WATCHPOINT_ASSERT(ptr->continuum[index].value < ptr->continuum[index + 1].value); + } +#endif +} + + memcached_return memcached_server_push(memcached_st *ptr, memcached_server_st *list) { unsigned int x; @@ -128,7 +186,20 @@ memcached_return memcached_server_push(memcached_st *ptr, memcached_server_st *l if (ptr->flags & MEM_USE_SORT_HOSTS) sort_hosts(ptr); - rebalance_wheel(ptr); + switch (ptr->distribution) + { + case MEMCACHED_DISTRIBUTION_CONSISTENT: + case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA: + update_continuum(ptr); + break; + case MEMCACHED_DISTRIBUTION_CONSISTENT_WHEEL: + rebalance_wheel(ptr); + break; + case MEMCACHED_DISTRIBUTION_MODULA: + break; + default: + WATCHPOINT_ASSERT(0); /* We have added a distribution without extending the logic */ + } return MEMCACHED_SUCCESS; } @@ -194,7 +265,20 @@ static memcached_return server_add(memcached_st *ptr, char *hostname, ptr->hosts[0].count= ptr->number_of_hosts; - rebalance_wheel(ptr); + switch (ptr->distribution) + { + case MEMCACHED_DISTRIBUTION_CONSISTENT: + case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA: + update_continuum(ptr); + break; + case MEMCACHED_DISTRIBUTION_CONSISTENT_WHEEL: + rebalance_wheel(ptr); + break; + case MEMCACHED_DISTRIBUTION_MODULA: + break; + default: + WATCHPOINT_ASSERT(0); /* We have added a distribution without extending the logic */ + } LIBMEMCACHED_MEMCACHED_SERVER_ADD_END(); -- 2.30.2