X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fmemcached_get.c;h=f00cfafbc261b94fa26ce27f838ebba2e5811f6a;hb=3f4d3e26db2dafabf270e964ba2be0dd72756c4d;hp=05b317a6268b79d19b5a711d49dfb05cf4dc3cb1;hpb=1d3aea18eadb5a2ca6893ab860bb5dbace09b94a;p=m6w6%2Flibmemcached diff --git a/libmemcached/memcached_get.c b/libmemcached/memcached_get.c index 05b317a6..f00cfafb 100644 --- a/libmemcached/memcached_get.c +++ b/libmemcached/memcached_get.c @@ -37,7 +37,7 @@ char *memcached_get_by_key(memcached_st *ptr, *error= memcached_mget_by_key(ptr, master_key, master_key_length, - (char **)&key, &key_length, 1); + (const char **)&key, &key_length, 1); value= memcached_fetch(ptr, NULL, NULL, value_length, flags, error); @@ -102,8 +102,8 @@ char *memcached_get_by_key(memcached_st *ptr, } memcached_return memcached_mget(memcached_st *ptr, - char **keys, size_t *key_length, - unsigned int number_of_keys) + const char **keys, size_t *key_length, + size_t number_of_keys) { return memcached_mget_by_key(ptr, NULL, 0, keys, key_length, number_of_keys); } @@ -111,19 +111,19 @@ memcached_return memcached_mget(memcached_st *ptr, static memcached_return binary_mget_by_key(memcached_st *ptr, unsigned int master_server_key, bool is_master_key_set, - char **keys, size_t *key_length, - unsigned int number_of_keys); + const char **keys, size_t *key_length, + size_t number_of_keys); memcached_return memcached_mget_by_key(memcached_st *ptr, const char *master_key, size_t master_key_length, - char **keys, + const char **keys, size_t *key_length, - unsigned int number_of_keys) + size_t number_of_keys) { unsigned int x; memcached_return rc= MEMCACHED_NOTFOUND; - char *get_command= "get "; + const char *get_command= "get "; uint8_t get_command_length= 4; unsigned int master_server_key= (unsigned int)-1; /* 0 is a valid server id! */ bool is_master_key_set= false; @@ -143,15 +143,9 @@ memcached_return memcached_mget_by_key(memcached_st *ptr, if ((ptr->flags & MEM_VERIFY_KEY) && (memcached_key_test(keys, key_length, number_of_keys) == MEMCACHED_BAD_KEY_PROVIDED)) return MEMCACHED_BAD_KEY_PROVIDED; - if (ptr->flags & MEM_SUPPORT_CAS) - { - get_command= "gets "; - get_command_length= 5; - } - if (master_key && master_key_length) { - if ((ptr->flags & MEM_VERIFY_KEY) && (memcached_key_test((char **)&master_key, &master_key_length, 1) == MEMCACHED_BAD_KEY_PROVIDED)) + if ((ptr->flags & MEM_VERIFY_KEY) && (memcached_key_test((const char **)&master_key, &master_key_length, 1) == MEMCACHED_BAD_KEY_PROVIDED)) return MEMCACHED_BAD_KEY_PROVIDED; master_server_key= memcached_generate_hash(ptr, master_key, master_key_length); is_master_key_set= true; @@ -181,6 +175,12 @@ memcached_return memcached_mget_by_key(memcached_st *ptr, return binary_mget_by_key(ptr, master_server_key, is_master_key_set, keys, key_length, number_of_keys); + if (ptr->flags & MEM_SUPPORT_CAS) + { + get_command= "gets "; + get_command_length= 5; + } + /* If a server fails we warn about errors and start all over with sending keys to the server. @@ -256,11 +256,11 @@ memcached_return memcached_mget_by_key(memcached_st *ptr, return rc; } -static memcached_return binary_mget_by_key(memcached_st *ptr, +static memcached_return simple_binary_mget(memcached_st *ptr, unsigned int master_server_key, bool is_master_key_set, - char **keys, size_t *key_length, - unsigned int number_of_keys) + const char **keys, size_t *key_length, + size_t number_of_keys) { memcached_return rc= MEMCACHED_NOTFOUND; uint32_t x; @@ -357,3 +357,148 @@ static memcached_return binary_mget_by_key(memcached_st *ptr, return rc; } + +static memcached_return replication_binary_mget(memcached_st *ptr, + uint32_t* hash, bool* dead_servers, + const char **keys, size_t *key_length, + size_t number_of_keys) +{ + memcached_return rc= MEMCACHED_NOTFOUND; + uint32_t x; + + int flush= number_of_keys == 1; + + for (uint32_t replica= 0; replica <= ptr->number_of_replicas; ++replica) + { + bool success= true; + + for (x= 0; x < number_of_keys; ++x) + { + if (hash[x] == ptr->number_of_hosts) + continue; /* Already successfully sent */ + + uint32_t server= hash[x] + replica; + while (server >= ptr->number_of_hosts) + server -= ptr->number_of_hosts; + + if (dead_servers[server]) + continue; + + if (memcached_server_response_count(&ptr->hosts[server]) == 0) + { + rc= memcached_connect(&ptr->hosts[server]); + if (rc != MEMCACHED_SUCCESS) + { + memcached_io_reset(&ptr->hosts[server]); + dead_servers[server]= true; + success= false; + continue; + } + } + + protocol_binary_request_getk request= {.bytes= {0}}; + request.message.header.request.magic= PROTOCOL_BINARY_REQ; + if (number_of_keys == 1) + request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK; + else + request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETKQ; + + request.message.header.request.keylen= htons((uint16_t)key_length[x]); + request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES; + request.message.header.request.bodylen= htonl(key_length[x]); + + if ((memcached_io_write(&ptr->hosts[server], request.bytes, + sizeof(request.bytes), 0) == -1) || + (memcached_io_write(&ptr->hosts[server], keys[x], + key_length[x], flush) == -1)) + { + memcached_io_reset(&ptr->hosts[server]); + dead_servers[server]= true; + success= false; + continue; + } + memcached_server_response_increment(&ptr->hosts[server]); + } + + if (number_of_keys > 1) + { + /* + * Send a noop command to flush the buffers + */ + protocol_binary_request_noop request= {.bytes= {0}}; + request.message.header.request.magic= PROTOCOL_BINARY_REQ; + request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP; + request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES; + + for (x= 0; x < ptr->number_of_hosts; x++) + if (memcached_server_response_count(&ptr->hosts[x])) + { + if (memcached_io_write(&ptr->hosts[x], request.bytes, + sizeof(request.bytes), 1) == -1) + { + memcached_io_reset(&ptr->hosts[x]); + dead_servers[x]= true; + success= false; + } + memcached_server_response_increment(&ptr->hosts[x]); + + /* mark all of the messages bound for this server as sent! */ + for (x= 0; x < number_of_keys; ++x) + if (hash[x] == x) + hash[x]= ptr->number_of_hosts; + } + } + + if (success) + break; + } + + return rc; +} + +static memcached_return binary_mget_by_key(memcached_st *ptr, + unsigned int master_server_key, + bool is_master_key_set, + const char **keys, size_t *key_length, + size_t number_of_keys) +{ + memcached_return rc; + + if (ptr->number_of_replicas == 0) + { + rc= simple_binary_mget(ptr, master_server_key, is_master_key_set, + keys, key_length, number_of_keys); + } + else + { + uint32_t* hash; + bool* dead_servers; + + hash= ptr->call_malloc(ptr, sizeof(uint32_t) * number_of_keys); + dead_servers= ptr->call_calloc(ptr, ptr->number_of_hosts, sizeof(bool)); + + if (hash == NULL || dead_servers == NULL) + { + ptr->call_free(ptr, hash); + ptr->call_free(ptr, dead_servers); + return MEMCACHED_MEMORY_ALLOCATION_FAILURE; + } + + if (is_master_key_set) + for (unsigned int x= 0; x < number_of_keys; x++) + hash[x]= master_server_key; + else + for (unsigned int x= 0; x < number_of_keys; x++) + hash[x]= memcached_generate_hash(ptr, keys[x], key_length[x]); + + rc= replication_binary_mget(ptr, hash, dead_servers, keys, + key_length, number_of_keys); + + ptr->call_free(ptr, hash); + ptr->call_free(ptr, dead_servers); + + return MEMCACHED_SUCCESS; + } + + return rc; +}