X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fmemcached_storage.c;h=ecefc56674134a47b1e4dc545be9d5a6c02db2d4;hb=a7a011c2ea4a63368b3a96a332da00820ed402cb;hp=dd7e23b0ddf2d8d99610b9b04687edefa841d78e;hpb=469a069004a24764960f98b0fd6101ce84199733;p=m6w6%2Flibmemcached diff --git a/libmemcached/memcached_storage.c b/libmemcached/memcached_storage.c index dd7e23b0..ecefc566 100644 --- a/libmemcached/memcached_storage.c +++ b/libmemcached/memcached_storage.c @@ -19,7 +19,7 @@ typedef enum { } memcached_storage_action; /* Inline this */ -static char *storage_op_string(memcached_storage_action verb) +static const char *storage_op_string(memcached_storage_action verb) { switch (verb) { @@ -42,7 +42,9 @@ static char *storage_op_string(memcached_storage_action verb) /* NOTREACHED */ } -static memcached_return memcached_send_binary(memcached_server_st* server, +static memcached_return memcached_send_binary(memcached_st *ptr, + const char *master_key, + size_t master_key_length, const char *key, size_t key_length, const char *value, @@ -77,25 +79,26 @@ static inline memcached_return memcached_send(memcached_st *ptr, unlikely (ptr->number_of_hosts == 0) return MEMCACHED_NO_SERVERS; - if ((ptr->flags & MEM_VERIFY_KEY) && (memcached_key_test((char **)&key, &key_length, 1) == MEMCACHED_BAD_KEY_PROVIDED)) + if ((ptr->flags & MEM_VERIFY_KEY) && (memcached_key_test((const char **)&key, &key_length, 1) == MEMCACHED_BAD_KEY_PROVIDED)) return MEMCACHED_BAD_KEY_PROVIDED; - server_key= memcached_generate_hash(ptr, master_key, master_key_length); - if (ptr->flags & MEM_BINARY_PROTOCOL) - return memcached_send_binary(&ptr->hosts[server_key], key, key_length, - value, value_length, expiration, + return memcached_send_binary(ptr, master_key, master_key_length, + key, key_length, + value, value_length, expiration, flags, cas, verb); + server_key= memcached_generate_hash(ptr, master_key, master_key_length); + if (cas) - write_length= snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, - "%s %s%.*s %u %llu %zu %llu%s\r\n", - storage_op_string(verb), - ptr->prefix_key, - (int)key_length, key, flags, - (unsigned long long)expiration, value_length, - (unsigned long long)cas, - (ptr->flags & MEM_NOREPLY) ? " noreply" : ""); + write_length= (size_t) snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, + "%s %s%.*s %u %llu %zu %llu%s\r\n", + storage_op_string(verb), + ptr->prefix_key, + (int)key_length, key, flags, + (unsigned long long)expiration, value_length, + (unsigned long long)cas, + (ptr->flags & MEM_NOREPLY) ? " noreply" : ""); else { char *buffer_ptr= buffer; @@ -115,11 +118,11 @@ static inline memcached_return memcached_send(memcached_st *ptr, buffer_ptr++; write_length= (size_t)(buffer_ptr - buffer); - write_length+= snprintf(buffer_ptr, MEMCACHED_DEFAULT_COMMAND_SIZE, - "%u %llu %zu%s\r\n", - flags, - (unsigned long long)expiration, value_length, - (ptr->flags & MEM_NOREPLY) ? " noreply" : ""); + write_length+= (size_t) snprintf(buffer_ptr, MEMCACHED_DEFAULT_COMMAND_SIZE, + "%u %llu %zu%s\r\n", + flags, + (unsigned long long)expiration, value_length, + (ptr->flags & MEM_NOREPLY) ? " noreply" : ""); } if (ptr->flags & MEM_USE_UDP && ptr->flags & MEM_BUFFER_REQUESTS) @@ -414,8 +417,10 @@ static inline uint8_t get_com_code(memcached_storage_action verb, bool noreply) -static memcached_return memcached_send_binary(memcached_server_st* server, - const char *key, +static memcached_return memcached_send_binary(memcached_st *ptr, + const char *master_key, + size_t master_key_length, + const char *key, size_t key_length, const char *value, size_t value_length, @@ -424,9 +429,12 @@ static memcached_return memcached_send_binary(memcached_server_st* server, uint64_t cas, memcached_storage_action verb) { - char flush; + uint8_t flush; protocol_binary_request_set request= {.bytes= {0}}; size_t send_length= sizeof(request.bytes); + uint32_t server_key= memcached_generate_hash(ptr, master_key, + master_key_length); + memcached_server_st *server= &ptr->hosts[server_key]; bool noreply= server->root->flags & MEM_NOREPLY; request.message.header.request.magic= PROTOCOL_BINARY_REQ; @@ -442,13 +450,13 @@ static memcached_return memcached_send_binary(memcached_server_st* server, request.message.body.expiration= htonl((uint32_t)expiration); } - request.message.header.request.bodylen= htonl(key_length + value_length + - request.message.header.request.extlen); + request.message.header.request.bodylen= htonl((uint32_t) (key_length + value_length + + request.message.header.request.extlen)); if (cas) request.message.header.request.cas= htonll(cas); - flush= ((server->root->flags & MEM_BUFFER_REQUESTS) && verb == SET_OP) ? 0 : 1; + flush= (uint8_t) (((server->root->flags & MEM_BUFFER_REQUESTS) && verb == SET_OP) ? 0 : 1); if ((server->root->flags & MEM_USE_UDP) && !flush) { @@ -462,12 +470,33 @@ static memcached_return memcached_send_binary(memcached_server_st* server, /* write the header */ if ((memcached_do(server, (const char*)request.bytes, send_length, 0) != MEMCACHED_SUCCESS) || (memcached_io_write(server, key, key_length, 0) == -1) || - (memcached_io_write(server, value, value_length, flush) == -1)) + (memcached_io_write(server, value, value_length, (char) flush) == -1)) { memcached_io_reset(server); return MEMCACHED_WRITE_FAILURE; } - + + unlikely (verb == SET_OP && ptr->number_of_replicas > 0) + { + request.message.header.request.opcode= PROTOCOL_BINARY_CMD_SETQ; + + for (uint32_t x= 0; x < ptr->number_of_replicas; x++) + { + ++server_key; + if (server_key == ptr->number_of_hosts) + server_key= 0; + + memcached_server_st *srv= &ptr->hosts[server_key]; + if ((memcached_do(srv, (const char*)request.bytes, + send_length, 0) != MEMCACHED_SUCCESS) || + (memcached_io_write(srv, key, key_length, 0) == -1) || + (memcached_io_write(srv, value, value_length, (char) flush) == -1)) + memcached_io_reset(srv); + else + memcached_server_response_decrement(srv); + } + } + if (flush == 0) return MEMCACHED_BUFFERED;