From 957f0cc7c87115224087f2c0de83a4c4aaa2c36d Mon Sep 17 00:00:00 2001 From: Date: Wed, 23 Jul 2008 12:16:31 -0700 Subject: [PATCH] Pushing weighted ketama code. --- ChangeLog | 1 + THANKS | 2 +- docs/memcached_behavior.pod | 10 ++++++ libmemcached/common.h | 2 ++ libmemcached/memcached.c | 1 + libmemcached/memcached.h | 2 ++ libmemcached/memcached_behavior.c | 19 +++++++++++- libmemcached/memcached_constants.h | 2 ++ libmemcached/memcached_hash.c | 27 ++++++++++------ libmemcached/memcached_hosts.c | 36 ++++++++++++++++++++-- libmemcached/memcached_server.h | 1 + tests/function.c | 49 ++++++++++++++++++++++++++++-- tests/server.c | 18 ++++++++--- 13 files changed, 151 insertions(+), 19 deletions(-) diff --git a/ChangeLog b/ChangeLog index c8e04e6f..6500f2a2 100644 --- a/ChangeLog +++ b/ChangeLog @@ -1,6 +1,7 @@ 0.23 * Added strings.h header for Solaris 9 * Solaris 64bit fix. + * Support for weighted Ketama from Yin Chen. 0.22 Mon Jul 14 09:24:11 PDT 2008 * Fix where master key was no being checked for "bad key" diff --git a/THANKS b/THANKS index bf6975fa..9221e3a6 100644 --- a/THANKS +++ b/THANKS @@ -9,4 +9,4 @@ Kevin Dalley - Bug Fixes Patrick Galbraith - work on C++ interface Ross McFarland - Idea for sorting servers. Marcelo Fernandez - TCP/IP timeout pieces -Yin Chen - Ketama support +Yin Chen - Ketama support/weighted support diff --git a/docs/memcached_behavior.pod b/docs/memcached_behavior.pod index c33b2af5..f717984d 100755 --- a/docs/memcached_behavior.pod +++ b/docs/memcached_behavior.pod @@ -77,6 +77,16 @@ Support CAS operations (this is not enabled by default at this point in the serv Sets the default distribution to MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA and the hash to MEMCACHED_HASH_MD5. +=item MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED + +Sets the default distribution to MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA with the weighted support. +and the hash to MEMCACHED_HASH_MD5. + +=item MEMCACHED_BEHAVIOR_KETAMA_HASH + +Sets the hashing algorithm for host mapping on continuum. The value can be set +to either MEMCACHED_HASH_DEFAULT, MEMCACHED_HASH_MD5, MEMCACHED_HASH_CRC, MEMCACHED_HASH_FNV1_64, MEMCACHED_HASH_FNV1A_64, MEMCACHED_HASH_FNV1_32, and MEMCACHED_HASH_FNV1A_32. + =item MEMCACHED_BEHAVIOR_POLL_TIMEOUT Modify the timeout value that is used by poll(). The default value is -1. An signed int pointer must be passed to memcached_behavior_set() to change this value. For memcached_behavior_get() a signed int value will be cast and returned as the unsigned long long. diff --git a/libmemcached/common.h b/libmemcached/common.h index 8cef723a..0c9b87ce 100644 --- a/libmemcached/common.h +++ b/libmemcached/common.h @@ -78,6 +78,8 @@ typedef enum { MEM_BUFFER_REQUESTS= (1 << 8), MEM_USE_SORT_HOSTS= (1 << 9), MEM_VERIFY_KEY= (1 << 10), + /* 11 used for weighted ketama */ + MEM_KETAMA_WEIGHTED= (1 << 11), } memcached_flags; /* Hashing algo */ diff --git a/libmemcached/memcached.c b/libmemcached/memcached.c index 2efcc7d5..dcf1a4a0 100644 --- a/libmemcached/memcached.c +++ b/libmemcached/memcached.c @@ -103,6 +103,7 @@ memcached_st *memcached_clone(memcached_st *clone, memcached_st *ptr) new_clone->retry_timeout= ptr->retry_timeout; new_clone->distribution= ptr->distribution; new_clone->hash= ptr->hash; + new_clone->hash_continuum= ptr->hash_continuum; new_clone->user_data= ptr->user_data; new_clone->on_clone= ptr->on_clone; diff --git a/libmemcached/memcached.h b/libmemcached/memcached.h index d1daf775..911b0444 100644 --- a/libmemcached/memcached.h +++ b/libmemcached/memcached.h @@ -93,6 +93,8 @@ struct memcached_st { memcached_trigger_delete_key delete_trigger; char prefix_key[MEMCACHED_PREFIX_KEY_MAX_SIZE]; size_t prefix_key_length; + memcached_hash hash_continuum; + uint32_t continuum_points_counter; }; diff --git a/libmemcached/memcached_behavior.c b/libmemcached/memcached_behavior.c index e89473cf..c001c461 100644 --- a/libmemcached/memcached_behavior.c +++ b/libmemcached/memcached_behavior.c @@ -58,9 +58,21 @@ memcached_return memcached_behavior_set(memcached_st *ptr, run_distribution(ptr); break; } + case MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED: + { + ptr->hash= MEMCACHED_HASH_MD5; + ptr->distribution= MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA; + set_behavior_flag(ptr, MEM_KETAMA_WEIGHTED, data); + run_distribution(ptr); + break; + } case MEMCACHED_BEHAVIOR_HASH: ptr->hash= (memcached_hash)(data); break; + case MEMCACHED_BEHAVIOR_KETAMA_HASH: + ptr->hash_continuum= (memcached_hash)(data); + run_distribution(ptr); + break; case MEMCACHED_BEHAVIOR_CACHE_LOOKUPS: set_behavior_flag(ptr, MEM_USE_CACHE_LOOKUPS, data); memcached_quit(ptr); @@ -124,12 +136,17 @@ uint64_t memcached_behavior_get(memcached_st *ptr, case MEMCACHED_BEHAVIOR_VERIFY_KEY: temp_flag= MEM_VERIFY_KEY; break; + case MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED: + temp_flag= MEM_KETAMA_WEIGHTED; + break; case MEMCACHED_BEHAVIOR_DISTRIBUTION: return ptr->distribution; case MEMCACHED_BEHAVIOR_KETAMA: - return (ptr->distribution == MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA && ptr->hash == MEMCACHED_HASH_MD5 ) ? 1 : 0; + return (ptr->distribution == MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA) ? 1 : 0; case MEMCACHED_BEHAVIOR_HASH: return ptr->hash; + case MEMCACHED_BEHAVIOR_KETAMA_HASH: + return ptr->hash_continuum; case MEMCACHED_BEHAVIOR_SORT_HOSTS: temp_flag= MEM_USE_SORT_HOSTS; break; diff --git a/libmemcached/memcached_constants.h b/libmemcached/memcached_constants.h index fd2f2b4d..69c7666b 100644 --- a/libmemcached/memcached_constants.h +++ b/libmemcached/memcached_constants.h @@ -86,6 +86,8 @@ typedef enum { MEMCACHED_BEHAVIOR_VERIFY_KEY, MEMCACHED_BEHAVIOR_CONNECT_TIMEOUT, MEMCACHED_BEHAVIOR_RETRY_TIMEOUT, + MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED, + MEMCACHED_BEHAVIOR_KETAMA_HASH, } memcached_behavior; typedef enum { diff --git a/libmemcached/memcached_hash.c b/libmemcached/memcached_hash.c index 5bb0fcee..75a1537a 100644 --- a/libmemcached/memcached_hash.c +++ b/libmemcached/memcached_hash.c @@ -11,18 +11,12 @@ static uint32_t FNV_32_PRIME= 16777619; static uint32_t internal_generate_hash(const char *key, size_t key_length); static uint32_t internal_generate_md5(const char *key, size_t key_length); -uint32_t generate_hash(memcached_st *ptr, const char *key, size_t key_length) +uint32_t generate_hash_value(const char *key, size_t key_length, memcached_hash hash_algorithm) { uint32_t hash= 1; /* Just here to remove compile warning */ uint32_t x= 0; - - WATCHPOINT_ASSERT(ptr->number_of_hosts); - - if (ptr->number_of_hosts == 1) - return 0; - - switch (ptr->hash) + switch (hash_algorithm) { case MEMCACHED_HASH_DEFAULT: hash= internal_generate_hash(key, key_length); @@ -91,7 +85,20 @@ uint32_t generate_hash(memcached_st *ptr, const char *key, size_t key_length) break; } } + return hash; +} + +uint32_t generate_hash(memcached_st *ptr, const 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_value(key, key_length, ptr->hash); WATCHPOINT_ASSERT(hash); return hash; } @@ -103,7 +110,7 @@ unsigned int dispatch_host(memcached_st *ptr, uint32_t hash) case MEMCACHED_DISTRIBUTION_CONSISTENT: case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA: { - uint32_t num= ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER; + uint32_t num= ptr->continuum_points_counter; WATCHPOINT_ASSERT(ptr->continuum); hash= hash; @@ -156,6 +163,7 @@ unsigned int dispatch_host(memcached_st *ptr, uint32_t hash) uint32_t memcached_generate_hash(memcached_st *ptr, const char *key, size_t key_length) { uint32_t hash= 1; /* Just here to remove compile warning */ + uint32_t result= 1; WATCHPOINT_ASSERT(ptr->number_of_hosts); @@ -165,6 +173,7 @@ uint32_t memcached_generate_hash(memcached_st *ptr, const char *key, size_t key_ hash = generate_hash(ptr, key, key_length); WATCHPOINT_ASSERT(hash); + return dispatch_host(ptr, hash); } diff --git a/libmemcached/memcached_hosts.c b/libmemcached/memcached_hosts.c index d75da5d2..55d7a49d 100644 --- a/libmemcached/memcached_hosts.c +++ b/libmemcached/memcached_hosts.c @@ -1,4 +1,5 @@ #include "common.h" +#include /* Protoypes (static) */ static memcached_return server_add(memcached_st *ptr, char *hostname, @@ -107,6 +108,12 @@ memcached_return update_continuum(memcached_st *ptr) uint32_t continuum_index= 0; uint32_t value; memcached_server_st *list; + uint32_t pointer_counter= 0; + uint32_t pointer_per_server= MEMCACHED_POINTS_PER_SERVER; + memcached_return rc; + uint64_t total_mem_bytes= 0; + memcached_stat_st *stat_p= NULL; + uint32_t is_ketama_weighted= 0; if (ptr->number_of_hosts > ptr->continuum_count) { @@ -125,9 +132,29 @@ memcached_return update_continuum(memcached_st *ptr) } list = ptr->hosts; + + is_ketama_weighted= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED); + if(is_ketama_weighted) + { + stat_p = memcached_stat(ptr, NULL, &rc); + for (host_index = 0; host_index < ptr->number_of_hosts; ++host_index) + { + list[host_index].limit_maxbytes= (stat_p + host_index)->limit_maxbytes; + total_mem_bytes += (stat_p + host_index)->limit_maxbytes; + } + } + for (host_index = 0; host_index < ptr->number_of_hosts; ++host_index) { - for(index= 1; index <= MEMCACHED_POINTS_PER_SERVER; ++index) + if(is_ketama_weighted) + { + float pct = (float)list[host_index].limit_maxbytes/ (float)total_mem_bytes; + pointer_per_server= floorf( pct * MEMCACHED_POINTS_PER_SERVER * (float)(ptr->number_of_hosts)); +#ifdef HAVE_DEBUG + printf("ketama_weighted:%s|%d|%llu|%u\n", list[host_index].hostname, list[host_index].port, list[host_index].limit_maxbytes, pointer_per_server); +#endif + } + for(index= 1; index <= pointer_per_server; ++index) { char sort_host[MEMCACHED_MAX_HOST_SORT_LENGTH]= ""; size_t sort_host_length; @@ -135,14 +162,19 @@ memcached_return update_continuum(memcached_st *ptr) 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= generate_hash(ptr, sort_host, sort_host_length); + value= generate_hash_value(sort_host, sort_host_length, ptr->hash_continuum); ptr->continuum[continuum_index].index= host_index; ptr->continuum[continuum_index++].value= value; } + pointer_counter+= pointer_per_server; } WATCHPOINT_ASSERT(ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER <= MEMCACHED_CONTINUUM_SIZE); qsort(ptr->continuum, ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER, sizeof(memcached_continuum_item_st), continuum_item_cmp); + + ptr->continuum_points_counter= pointer_counter; + memcached_stat_free(NULL, stat_p); + #ifdef HAVE_DEBUG for (index= 0; index < ((ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER) - 1); index++) { diff --git a/libmemcached/memcached_server.h b/libmemcached/memcached_server.h index f0eff961..c566ddbe 100644 --- a/libmemcached/memcached_server.h +++ b/libmemcached/memcached_server.h @@ -34,6 +34,7 @@ struct memcached_server_st { uint16_t count; time_t next_retry; memcached_st *root; + uint64_t limit_maxbytes; }; #define memcached_server_count(A) (A)->number_of_hosts diff --git a/tests/function.c b/tests/function.c index 59e1f2b9..d83b9729 100644 --- a/tests/function.c +++ b/tests/function.c @@ -24,8 +24,9 @@ #include "test.h" -#define GLOBAL_COUNT 100000 -#define GLOBAL2_COUNT 1000 +#define GLOBAL_COUNT 10000 +#define GLOBAL2_COUNT 100 +#define SERVERS_TO_CREATE 5 static uint32_t global_count; static pairs_st *global_pairs; @@ -2237,6 +2238,23 @@ test_return generate_data(memcached_st *memc) return 0; } +test_return generate_data_with_stats(memcached_st *memc) +{ + memcached_stat_st *stat_p= NULL; + memcached_return rc; + int host_index= 0; + execute_set(memc, global_pairs, global_count); + + //TODO: hosts used size stats + stat_p = memcached_stat(memc, NULL, &rc); + for (host_index = 0; host_index < SERVERS_TO_CREATE; ++host_index) + { + printf("\nserver %d|%s|%d bytes: %lld\n", host_index, (memc->hosts)[host_index].hostname, (memc->hosts)[host_index].port, (stat_p + host_index)->bytes); + } + + + return 0; +} test_return generate_buffer_data(memcached_st *memc) { int latch= 0; @@ -2521,6 +2539,24 @@ memcached_return pre_behavior_ketama(memcached_st *memc) return MEMCACHED_SUCCESS; } +memcached_return pre_behavior_ketama_weighted(memcached_st *memc) +{ + memcached_return rc; + uint64_t value; + + rc= memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED, 1); + assert(rc == MEMCACHED_SUCCESS); + + value= memcached_behavior_get(memc, MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED); + assert(value == 1); + + rc= memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_KETAMA_HASH, MEMCACHED_HASH_MD5); + assert(rc == MEMCACHED_SUCCESS); + + value= memcached_behavior_get(memc, MEMCACHED_BEHAVIOR_KETAMA_HASH); + assert(value == MEMCACHED_HASH_MD5); + return MEMCACHED_SUCCESS; +} void my_free(memcached_st *ptr, void *mem) { free(mem); @@ -2862,6 +2898,14 @@ test_st consistent_tests[] ={ {0, 0, 0} }; +test_st consistent_weighted_tests[] ={ + {"generate_pairs", 1, generate_pairs }, + {"generate_data", 1, generate_data_with_stats }, + {"get_read", 0, get_read_count }, + {"cleanup", 1, cleanup_pairs }, + {0, 0, 0} +}; + collection_st collection[] ={ {"block", 0, 0, tests}, {"nonblock", pre_nonblock, 0, tests}, @@ -2896,6 +2940,7 @@ collection_st collection[] ={ {"generate_nonblock", pre_nonblock, 0, generate_tests}, {"consistent_not", 0, 0, consistent_tests}, {"consistent_ketama", pre_behavior_ketama, 0, consistent_tests}, + {"consistent_ketama_weighted", pre_behavior_ketama_weighted, 0, consistent_weighted_tests}, {0, 0, 0, 0} }; diff --git a/tests/server.c b/tests/server.c index 1248981b..29f6699d 100644 --- a/tests/server.c +++ b/tests/server.c @@ -37,10 +37,20 @@ void server_startup(server_startup_st *construct) int count; int status; - if (construct->udp) - sprintf(buffer, "memcached -d -P /tmp/%umemc.pid -t 1 -U %u", x, x+ TEST_PORT_BASE); - else - sprintf(buffer, "memcached -d -P /tmp/%umemc.pid -t 1 -p %u", x, x+ TEST_PORT_BASE); + if (construct->udp){ + if(x == 0) { + sprintf(buffer, "memcached -d -P /tmp/%umemc.pid -t 1 -U %u -m 128", x, x+ TEST_PORT_BASE); + } else { + sprintf(buffer, "memcached -d -P /tmp/%umemc.pid -t 1 -U %u", x, x+ TEST_PORT_BASE); + } + } + else{ + if(x == 0) { + sprintf(buffer, "memcached -d -P /tmp/%umemc.pid -t 1 -p %u -m 128", x, x+ TEST_PORT_BASE); + } else { + sprintf(buffer, "memcached -d -P /tmp/%umemc.pid -t 1 -p %u", x, x+ TEST_PORT_BASE); + } + } status= system(buffer); count= sprintf(end_ptr, "localhost:%u,", x + TEST_PORT_BASE); end_ptr+= count; -- 2.30.2