X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=lib%2Fmemcached_storage.c;h=aac9b9227cae8d1c77f62ee0a912a86f53e39fb3;hb=dbce10442220e3c776226a3a2f4050cb6bcfbe42;hp=1ee6c4bafcd51bfef55b86a27bbb9625fe95982d;hpb=806525899ce3dff894b829416307a4290cd882e2;p=m6w6%2Flibmemcached diff --git a/lib/memcached_storage.c b/lib/memcached_storage.c index 1ee6c4ba..aac9b922 100644 --- a/lib/memcached_storage.c +++ b/lib/memcached_storage.c @@ -7,81 +7,307 @@ */ -#include - -static memcached_return memcached_send(memcached_st *ptr, - char *key, size_t key_length, - char *value, size_t value_length, - time_t expiration, - uint16_t flags, - char *verb) +#include "common.h" +#include "memcached_io.h" + +typedef enum { + SET_OP, + REPLACE_OP, + ADD_OP, + PREPEND_OP, + APPEND_OP, + CAS_OP, +} memcached_storage_action; + +/* Inline this */ +static char *storage_op_string(memcached_storage_action verb) +{ + switch (verb) + { + case SET_OP: + return "set"; + case REPLACE_OP: + return "replace"; + case ADD_OP: + return "add"; + case PREPEND_OP: + return "prepend"; + case APPEND_OP: + return "append"; + case CAS_OP: + return "cas"; + }; + + return SET_OP; +} + +static inline memcached_return memcached_send(memcached_st *ptr, + char *master_key, size_t master_key_length, + char *key, size_t key_length, + char *value, size_t value_length, + time_t expiration, + uint32_t flags, + uint64_t cas, + memcached_storage_action verb) { + char to_write; size_t write_length; ssize_t sent_length; - memcached_return rc; + memcached_return rc[MEMCACHED_MAX_REPLICAS]; char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE]; unsigned int server_key; + uint8_t replicas= 0; + + unlikely (key_length == 0) + return MEMCACHED_NO_KEY_PROVIDED; - assert(value); - assert(value_length); + unlikely (ptr->number_of_hosts == 0) + return MEMCACHED_NO_SERVERS; - rc= memcached_connect(ptr); - if (rc != MEMCACHED_SUCCESS) - return rc; + if ((ptr->flags & MEM_VERIFY_KEY) && (memcachd_key_test(&key, &key_length, 1) == MEMCACHED_BAD_KEY_PROVIDED)) + return MEMCACHED_BAD_KEY_PROVIDED; - server_key= memcached_generate_hash(key, key_length) % ptr->number_of_hosts; + server_key= memcached_generate_hash(ptr, master_key, master_key_length); + + if (cas) + write_length= snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, + "%s %.*s %u %llu %zu %llu\r\n", storage_op_string(verb), + (int)key_length, key, flags, + (unsigned long long)expiration, value_length, + (unsigned long long)cas); + else + write_length= snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, + "%s %.*s %u %llu %zu\r\n", storage_op_string(verb), + (int)key_length, key, flags, + (unsigned long long)expiration, value_length); - write_length= snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, - "%s %.*s %x %llu %zu\r\n", verb, - (int)key_length, key, flags, - (unsigned long long)expiration, value_length); if (write_length >= MEMCACHED_DEFAULT_COMMAND_SIZE) return MEMCACHED_WRITE_FAILURE; - if ((sent_length= write(ptr->hosts[server_key].fd, buffer, write_length)) == -1) - return MEMCACHED_WRITE_FAILURE; - assert(write_length == sent_length); - if ((sent_length= write(ptr->hosts[server_key].fd, value, value_length)) == -1) - return MEMCACHED_WRITE_FAILURE; - assert(value_length == sent_length); + if ((ptr->flags & MEM_BUFFER_REQUESTS) && verb == SET_OP) + to_write= 0; + else + to_write= 1; - if ((sent_length= write(ptr->hosts[server_key].fd, "\r\n", 2)) == -1) - return MEMCACHED_WRITE_FAILURE; - assert(2 == sent_length); + do + { + char response_buffer[MEMCACHED_DEFAULT_COMMAND_SIZE]; - sent_length= read(ptr->hosts[server_key].fd, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE); + rc[replicas]= memcached_do(&ptr->hosts[server_key], buffer, write_length, 0); - if (sent_length && buffer[0] == 'S') /* STORED */ - return MEMCACHED_SUCCESS; - else if (write_length && buffer[0] == 'N') /* NOT_STORED */ - return MEMCACHED_NOTSTORED; - else - return MEMCACHED_READ_FAILURE; + if (rc[replicas] != MEMCACHED_SUCCESS) + goto error; + + if ((sent_length= memcached_io_write(&ptr->hosts[server_key], value, value_length, 0)) == -1) + { + rc[replicas]= MEMCACHED_WRITE_FAILURE; + goto error; + } + + if ((sent_length= memcached_io_write(&ptr->hosts[server_key], "\r\n", 2, to_write)) == -1) + { + rc[replicas]= MEMCACHED_WRITE_FAILURE; + goto error; + } + + if (to_write == 0) + return MEMCACHED_BUFFERED; + else + rc[replicas]= memcached_response(&ptr->hosts[server_key], + response_buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL); + + /* On error we just jump to the next potential server */ +error: + if (replicas > 1 && ptr->distribution == MEMCACHED_DISTRIBUTION_CONSISTENT) + { + if (server_key == (ptr->number_of_hosts - 1)) + server_key= 0; + else + server_key++; + } + } while ((++replicas) < ptr->number_of_replicas); + + /* As long as one object gets stored, we count this as a success */ + while (replicas--) + { + if (rc[replicas] == MEMCACHED_STORED) + return MEMCACHED_SUCCESS; + } + + return rc[0]; } memcached_return memcached_set(memcached_st *ptr, char *key, size_t key_length, char *value, size_t value_length, time_t expiration, - uint16_t flags) + uint32_t flags) { - return memcached_send(ptr, key, key_length, value, value_length, - expiration, flags, "set"); + memcached_return rc; + LIBMEMCACHED_MEMCACHED_SET_START(); + rc= memcached_send(ptr, key, key_length, + key, key_length, value, value_length, + expiration, flags, 0, SET_OP); + LIBMEMCACHED_MEMCACHED_SET_END(); + return rc; } -memcached_return memcached_add(memcached_st *ptr, char *key, size_t key_length, +memcached_return memcached_add(memcached_st *ptr, + char *key, size_t key_length, char *value, size_t value_length, time_t expiration, - uint16_t flags) + uint32_t flags) +{ + memcached_return rc; + LIBMEMCACHED_MEMCACHED_ADD_START(); + rc= memcached_send(ptr, key, key_length, + key, key_length, value, value_length, + expiration, flags, 0, ADD_OP); + LIBMEMCACHED_MEMCACHED_ADD_END(); + return rc; +} + +memcached_return memcached_replace(memcached_st *ptr, + char *key, size_t key_length, + char *value, size_t value_length, + time_t expiration, + uint32_t flags) { - return memcached_send(ptr, key, key_length, value, value_length, - expiration, flags, "add"); + memcached_return rc; + LIBMEMCACHED_MEMCACHED_REPLACE_START(); + rc= memcached_send(ptr, key, key_length, + key, key_length, value, value_length, + expiration, flags, 0, REPLACE_OP); + LIBMEMCACHED_MEMCACHED_REPLACE_END(); + return rc; } -memcached_return memcached_replace(memcached_st *ptr, char *key, size_t key_length, +memcached_return memcached_prepend(memcached_st *ptr, + char *key, size_t key_length, char *value, size_t value_length, time_t expiration, - uint16_t flags) + uint32_t flags) +{ + memcached_return rc; + rc= memcached_send(ptr, key, key_length, + key, key_length, value, value_length, + expiration, flags, 0, PREPEND_OP); + return rc; +} + +memcached_return memcached_append(memcached_st *ptr, + char *key, size_t key_length, + char *value, size_t value_length, + time_t expiration, + uint32_t flags) +{ + memcached_return rc; + rc= memcached_send(ptr, key, key_length, + key, key_length, value, value_length, + expiration, flags, 0, APPEND_OP); + return rc; +} + +memcached_return memcached_cas(memcached_st *ptr, + char *key, size_t key_length, + char *value, size_t value_length, + time_t expiration, + uint32_t flags, + uint64_t cas) +{ + memcached_return rc; + rc= memcached_send(ptr, key, key_length, + key, key_length, value, value_length, + expiration, flags, cas, CAS_OP); + return rc; +} + +memcached_return memcached_set_by_key(memcached_st *ptr, + char *master_key, size_t master_key_length, + char *key, size_t key_length, + char *value, size_t value_length, + time_t expiration, + uint32_t flags) +{ + memcached_return rc; + LIBMEMCACHED_MEMCACHED_SET_START(); + rc= memcached_send(ptr, key, key_length, + key, key_length, value, value_length, + expiration, flags, 0, SET_OP); + LIBMEMCACHED_MEMCACHED_SET_END(); + return rc; +} + +memcached_return memcached_add_by_key(memcached_st *ptr, + char *master_key, size_t master_key_length, + char *key, size_t key_length, + char *value, size_t value_length, + time_t expiration, + uint32_t flags) +{ + memcached_return rc; + LIBMEMCACHED_MEMCACHED_ADD_START(); + rc= memcached_send(ptr, key, key_length, + key, key_length, value, value_length, + expiration, flags, 0, ADD_OP); + LIBMEMCACHED_MEMCACHED_ADD_END(); + return rc; +} + +memcached_return memcached_replace_by_key(memcached_st *ptr, + char *master_key, size_t master_key_length, + char *key, size_t key_length, + char *value, size_t value_length, + time_t expiration, + uint32_t flags) +{ + memcached_return rc; + LIBMEMCACHED_MEMCACHED_REPLACE_START(); + rc= memcached_send(ptr, key, key_length, + key, key_length, value, value_length, + expiration, flags, 0, REPLACE_OP); + LIBMEMCACHED_MEMCACHED_REPLACE_END(); + return rc; +} + +memcached_return memcached_prepend_by_key(memcached_st *ptr, + char *master_key, size_t master_key_length, + char *key, size_t key_length, + char *value, size_t value_length, + time_t expiration, + uint32_t flags) +{ + memcached_return rc; + rc= memcached_send(ptr, key, key_length, + key, key_length, value, value_length, + expiration, flags, 0, PREPEND_OP); + return rc; +} + +memcached_return memcached_append_by_key(memcached_st *ptr, + char *master_key, size_t master_key_length, + char *key, size_t key_length, + char *value, size_t value_length, + time_t expiration, + uint32_t flags) +{ + memcached_return rc; + rc= memcached_send(ptr, key, key_length, + key, key_length, value, value_length, + expiration, flags, 0, APPEND_OP); + return rc; +} + +memcached_return memcached_cas_by_key(memcached_st *ptr, + char *master_key, size_t master_key_length, + char *key, size_t key_length, + char *value, size_t value_length, + time_t expiration, + uint32_t flags, + uint64_t cas) { - return memcached_send(ptr, key, key_length, value, value_length, - expiration, flags, "replace"); + memcached_return rc; + rc= memcached_send(ptr, key, key_length, + key, key_length, value, value_length, + expiration, flags, cas, CAS_OP); + return rc; }