X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=lib%2Fmemcached_storage.c;h=7b48afeedea1e29a5b1146004ee73a59040324bc;hb=264a1ae95d273b6b32afed45d7c7c7461bfa54e4;hp=a40296dffca76384d2d8e6cc24e675190adb4a8a;hpb=1f34771f7ba527e3afb675f1a448d6c5ab66596b;p=awesomized%2Flibmemcached diff --git a/lib/memcached_storage.c b/lib/memcached_storage.c index a40296df..7b48afee 100644 --- a/lib/memcached_storage.c +++ b/lib/memcached_storage.c @@ -7,74 +7,217 @@ */ -#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 */ +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 *key, size_t key_length, + char *value, size_t value_length, + time_t expiration, + uint16_t flags, + uint64_t cas, + memcached_storage_action verb) { - size_t send_length; + char to_write; + size_t write_length; + ssize_t sent_length; memcached_return rc; char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE]; + unsigned int server_key; + + WATCHPOINT_ASSERT(!(value == NULL && value_length > 0)); + WATCHPOINT_ASSERT(!(value && value_length == 0)); + + if (key_length == 0) + return MEMCACHED_NO_KEY_PROVIDED; + + /* Leaving this WATCHPOINT_ASSERT in since only a library fubar could blow this */ +#ifdef NOT_DONE + if (!(ptr->flags & MEM_NO_BLOCK) && ptr->write_buffer_offset != 0) + WATCHPOINT_ASSERT(0); +#endif + + if (ptr->hosts == NULL || ptr->number_of_hosts == 0) + return MEMCACHED_NO_SERVERS; + + server_key= memcached_generate_hash(ptr, key, key_length); + + rc= memcached_connect(ptr, server_key); + if (rc != MEMCACHED_SUCCESS) + return rc; - rc= memcached_connect(ptr); - send_length= snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, - "%s %.*s %u %u %u\r\n", verb, - key_length, key, flags, expiration, value_length); - if ((send(ptr->fd, buffer, send_length, 0) == -1)) + 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); + + if (write_length >= MEMCACHED_DEFAULT_COMMAND_SIZE) { - fprintf(stderr, "failed set on %.*s TCP\n", key_length+1, key); + rc= MEMCACHED_WRITE_FAILURE; + goto error; + } - return MEMCACHED_WRITE_FAILURE; + /* + We have to flush after sending the command. Memcached is not smart enough + to just keep reading from the socket :( + */ + if ((sent_length= memcached_io_write(ptr, server_key, buffer, write_length, 0)) == -1) + { + rc= MEMCACHED_WRITE_FAILURE; + goto error; } - send_length= snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, - "%.*s\r\n", - value_length, value); - if ((send(ptr->fd, buffer, send_length, 0) == -1)) + if ((sent_length= memcached_io_write(ptr, server_key, value, value_length, 0)) == -1) { - fprintf(stderr, "failed set on %.*s TCP\n", key_length+1, key); + rc= MEMCACHED_WRITE_FAILURE; + goto error; + } + + if ((ptr->flags & MEM_NO_BLOCK) && verb == SET_OP) + to_write= 0; + else + to_write= 1; - return MEMCACHED_WRITE_FAILURE; + if ((sent_length= memcached_io_write(ptr, server_key, "\r\n", 2, to_write)) == -1) + { + memcached_quit_server(ptr, server_key); + rc= MEMCACHED_WRITE_FAILURE; + goto error; } - - send_length= read(ptr->fd, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE); - if (send_length && buffer[0] == 'S') /* STORED */ - return MEMCACHED_SUCCESS; - else if (send_length && buffer[0] == 'N') /* NOT_STORED */ - return MEMCACHED_NOTSTORED; + if ((ptr->flags & MEM_NO_BLOCK) && verb == SET_OP) + { + rc= MEMCACHED_SUCCESS; + memcached_server_response_increment(ptr, server_key); + } else - return MEMCACHED_READ_FAILURE; + { + rc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, server_key); + } + if (rc == MEMCACHED_STORED) + return MEMCACHED_SUCCESS; + else + return rc; + +error: + memcached_io_reset(ptr, server_key); + + return rc; } 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) + uint16_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, 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) + uint16_t flags) +{ + memcached_return rc; + LIBMEMCACHED_MEMCACHED_ADD_START(); + rc= memcached_send(ptr, 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, + uint16_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, 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) + uint16_t flags) { - return memcached_send(ptr, key, key_length, value, value_length, - expiration, flags, "replace"); + memcached_return rc; + rc= memcached_send(ptr, 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, + uint16_t flags) +{ + memcached_return rc; + rc= memcached_send(ptr, 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, + uint16_t flags, + uint64_t cas) +{ + memcached_return rc; + rc= memcached_send(ptr, key, key_length, value, value_length, + expiration, flags, cas, APPEND_OP); + return rc; }