X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fmemcached_response.c;h=193bfd8c2c3b3ad6f5fafd469504688e24cfcace;hb=a91a68a7c685f82bc6f46c09fa2036e44a82e7f7;hp=f81f51548dbe58be3788513188be5cfc2ecb080b;hpb=232a337899ee121d8ad4852862ee56bfb4b864fa;p=awesomized%2Flibmemcached diff --git a/libmemcached/memcached_response.c b/libmemcached/memcached_response.c index f81f5154..193bfd8c 100644 --- a/libmemcached/memcached_response.c +++ b/libmemcached/memcached_response.c @@ -8,95 +8,215 @@ #include "common.h" #include "memcached_io.h" -static memcached_return binary_response(memcached_server_st *ptr, - char *buffer, size_t buffer_length, - memcached_result_st *result); - -memcached_return memcached_response(memcached_server_st *ptr, - char *buffer, size_t buffer_length, - memcached_result_st *result) +static memcached_return textual_read_one_response(memcached_server_st *ptr, + char *buffer, size_t buffer_length, + memcached_result_st *result); +static memcached_return binary_read_one_response(memcached_server_st *ptr, + char *buffer, size_t buffer_length, + memcached_result_st *result); + +memcached_return memcached_read_one_response(memcached_server_st *ptr, + char *buffer, size_t buffer_length, + memcached_result_st *result) { - unsigned int x; - size_t send_length; - char *buffer_ptr; - unsigned int max_messages; + memcached_server_response_decrement(ptr); + if (result == NULL) + result = &ptr->root->result; - send_length= 0; - /* UDP at the moment is odd...*/ - if (ptr->type == MEMCACHED_CONNECTION_UDP) - { - char buffer[8]; - ssize_t read_length; + memcached_return rc; + if (ptr->root->flags & MEM_BINARY_PROTOCOL) + rc= binary_read_one_response(ptr, buffer, buffer_length, result); + else + rc= textual_read_one_response(ptr, buffer, buffer_length, result); - return MEMCACHED_SUCCESS; + unlikely(rc == MEMCACHED_UNKNOWN_READ_FAILURE || + rc == MEMCACHED_PROTOCOL_ERROR || + rc == MEMCACHED_CLIENT_ERROR || + rc == MEMCACHED_MEMORY_ALLOCATION_FAILURE) + memcached_io_reset(ptr); - read_length= memcached_io_read(ptr, buffer, 8); - } + return rc; +} +memcached_return memcached_response(memcached_server_st *ptr, + char *buffer, size_t buffer_length, + memcached_result_st *result) +{ /* We may have old commands in the buffer not set, first purge */ if (ptr->root->flags & MEM_NO_BLOCK) (void)memcached_io_write(ptr, NULL, 0, 1); - if (ptr->root->flags & MEM_BINARY_PROTOCOL) - return binary_response(ptr, buffer, buffer_length, result); - - max_messages= memcached_server_response_count(ptr); - for (x= 0; x < max_messages; x++) - { - size_t total_length= 0; - buffer_ptr= buffer; + /* + * The previous implementation purged all pending requests and just + * returned the last one. Purge all pending messages to ensure backwards + * compatibility. + */ + if ((ptr->root->flags & MEM_BINARY_PROTOCOL) == 0) + while (memcached_server_response_count(ptr) > 1) + { + memcached_return rc= memcached_read_one_response(ptr, buffer, buffer_length, result); + + unlikely (rc != MEMCACHED_END && + rc != MEMCACHED_STORED && + rc != MEMCACHED_SUCCESS && + rc != MEMCACHED_STAT && + rc != MEMCACHED_DELETED && + rc != MEMCACHED_NOTFOUND && + rc != MEMCACHED_NOTSTORED && + rc != MEMCACHED_DATA_EXISTS) + return rc; + } + return memcached_read_one_response(ptr, buffer, buffer_length, result); +} - while (1) - { - ssize_t read_length; +static memcached_return textual_value_fetch(memcached_server_st *ptr, + char *buffer, + memcached_result_st *result) +{ + memcached_return rc= MEMCACHED_SUCCESS; + char *string_ptr; + char *end_ptr; + char *next_ptr; + size_t value_length; + size_t to_read; + char *value_ptr; - read_length= memcached_io_read(ptr, buffer_ptr, 1); - WATCHPOINT_ASSERT(isgraph(*buffer_ptr) || isspace(*buffer_ptr)); + if (ptr->root->flags & MEM_USE_UDP) + return MEMCACHED_NOT_SUPPORTED; - if (read_length != 1) - { - memcached_io_reset(ptr); - return MEMCACHED_UNKNOWN_READ_FAILURE; - } + WATCHPOINT_ASSERT(ptr->root); + end_ptr= buffer + MEMCACHED_DEFAULT_COMMAND_SIZE; - if (*buffer_ptr == '\n') - break; - else - buffer_ptr++; + memcached_result_reset(result); - total_length++; - WATCHPOINT_ASSERT(total_length <= buffer_length); + string_ptr= buffer; + string_ptr+= 6; /* "VALUE " */ - if (total_length >= buffer_length) + + /* We load the key */ + { + char *key; + size_t prefix_length; + + key= result->key; + result->key_length= 0; + + for (prefix_length= ptr->root->prefix_key_length; !(iscntrl(*string_ptr) || isspace(*string_ptr)) ; string_ptr++) + { + if (prefix_length == 0) { - memcached_io_reset(ptr); - return MEMCACHED_PROTOCOL_ERROR; + *key= *string_ptr; + key++; + result->key_length++; } + else + prefix_length--; } - buffer_ptr++; - *buffer_ptr= 0; + result->key[result->key_length]= 0; + } + + if (end_ptr == string_ptr) + goto read_error; + + /* Flags fetch move past space */ + string_ptr++; + if (end_ptr == string_ptr) + goto read_error; + for (next_ptr= string_ptr; isdigit(*string_ptr); string_ptr++); + result->flags= strtoul(next_ptr, &string_ptr, 10); + + if (end_ptr == string_ptr) + goto read_error; - memcached_server_response_decrement(ptr); + /* Length fetch move past space*/ + string_ptr++; + if (end_ptr == string_ptr) + goto read_error; + + for (next_ptr= string_ptr; isdigit(*string_ptr); string_ptr++); + value_length= (size_t)strtoull(next_ptr, &string_ptr, 10); + + if (end_ptr == string_ptr) + goto read_error; + + /* Skip spaces */ + if (*string_ptr == '\r') + { + /* Skip past the \r\n */ + string_ptr+= 2; + } + else + { + string_ptr++; + for (next_ptr= string_ptr; isdigit(*string_ptr); string_ptr++); + result->cas= strtoull(next_ptr, &string_ptr, 10); + } + + if (end_ptr < string_ptr) + goto read_error; + + /* We add two bytes so that we can walk the \r\n */ + rc= memcached_string_check(&result->value, value_length+2); + if (rc != MEMCACHED_SUCCESS) + { + value_length= 0; + return MEMCACHED_MEMORY_ALLOCATION_FAILURE; + } + + value_ptr= memcached_string_value(&result->value); + /* + We read the \r\n into the string since not doing so is more + cycles then the waster of memory to do so. + + We are null terminating through, which will most likely make + some people lazy about using the return length. + */ + to_read= (value_length) + 2; + ssize_t read_length= 0; + memcached_return rrc= memcached_io_read(ptr, value_ptr, to_read, &read_length); + if (rrc != MEMCACHED_SUCCESS) + return rrc; + + if (read_length != (ssize_t)(value_length + 2)) + { + goto read_error; + } + +/* This next bit blows the API, but this is internal....*/ + { + char *char_ptr; + char_ptr= memcached_string_value(&result->value);; + char_ptr[value_length]= 0; + char_ptr[value_length + 1]= 0; + memcached_string_set_length(&result->value, value_length); } - uint64_t auto_return_value= 0; + return MEMCACHED_SUCCESS; + +read_error: + memcached_io_reset(ptr); + + return MEMCACHED_PARTIAL_READ; +} + +static memcached_return textual_read_one_response(memcached_server_st *ptr, + char *buffer, size_t buffer_length, + memcached_result_st *result) +{ + memcached_return rc= memcached_io_readline(ptr, buffer, buffer_length); + if (rc != MEMCACHED_SUCCESS) + return rc; + switch(buffer[0]) { case 'V': /* VALUE || VERSION */ if (buffer[1] == 'A') /* VALUE */ { - memcached_return rc; - /* We add back in one because we will need to search for END */ memcached_server_response_increment(ptr); - if (result) - rc= value_fetch(ptr, buffer, result); - else - rc= value_fetch(ptr, buffer, &ptr->root->result); - - return rc; + return textual_value_fetch(ptr, buffer, result); } else if (buffer[1] == 'E') /* VERSION */ { @@ -106,7 +226,6 @@ memcached_return memcached_response(memcached_server_st *ptr, { WATCHPOINT_STRING(buffer); WATCHPOINT_ASSERT(0); - memcached_io_reset(ptr); return MEMCACHED_UNKNOWN_READ_FAILURE; } case 'O': /* OK */ @@ -126,7 +245,6 @@ memcached_return memcached_response(memcached_server_st *ptr, { WATCHPOINT_STRING(buffer); WATCHPOINT_ASSERT(0); - memcached_io_reset(ptr); return MEMCACHED_UNKNOWN_READ_FAILURE; } } @@ -139,43 +257,33 @@ memcached_return memcached_response(memcached_server_st *ptr, else if (buffer[4] == 'S') return MEMCACHED_NOTSTORED; else - { - memcached_io_reset(ptr); return MEMCACHED_UNKNOWN_READ_FAILURE; - } } case 'E': /* PROTOCOL ERROR or END */ { if (buffer[1] == 'N') return MEMCACHED_END; else if (buffer[1] == 'R') - { - memcached_io_reset(ptr); return MEMCACHED_PROTOCOL_ERROR; - } else if (buffer[1] == 'X') - { - memcached_io_reset(ptr); return MEMCACHED_DATA_EXISTS; - } else - { - memcached_io_reset(ptr); return MEMCACHED_UNKNOWN_READ_FAILURE; - } } case 'C': /* CLIENT ERROR */ - memcached_io_reset(ptr); return MEMCACHED_CLIENT_ERROR; default: - if(sscanf(buffer, "%lld", &auto_return_value) == 1) + { + unsigned long long auto_return_value; + + if (sscanf(buffer, "%llu", &auto_return_value) == 1) return MEMCACHED_SUCCESS; - memcached_io_reset(ptr); - return MEMCACHED_UNKNOWN_READ_FAILURE; + return MEMCACHED_UNKNOWN_READ_FAILURE; + } } - return MEMCACHED_SUCCESS; + /* NOTREACHED */ } char *memcached_result_value(memcached_result_st *ptr) @@ -190,46 +298,18 @@ size_t memcached_result_length(memcached_result_st *ptr) return memcached_string_length(sptr); } -/** - * Read a given number of bytes from the server and place it into a specific - * buffer. Reset the IO channel or this server if an error occurs. - */ -static memcached_return safe_read(memcached_server_st *ptr, void *dta, - size_t size) -{ - int offset= 0; - char *data= dta; - while (offset < size) - { - ssize_t nread= memcached_io_read(ptr, data + offset, size - offset); - if (nread <= 0) - { - memcached_io_reset(ptr); - return MEMCACHED_UNKNOWN_READ_FAILURE; - } - offset += nread; - } - - return MEMCACHED_SUCCESS; -} - -static memcached_return binary_response(memcached_server_st *ptr, - char *buffer, - size_t buffer_length, - memcached_result_st *result) +static memcached_return binary_read_one_response(memcached_server_st *ptr, + char *buffer, size_t buffer_length, + memcached_result_st *result) { protocol_binary_response_header header; - memcached_server_response_decrement(ptr); - unlikely (safe_read(ptr, &header.bytes, - sizeof(header.bytes)) != MEMCACHED_SUCCESS) + unlikely (memcached_safe_read(ptr, &header.bytes, + sizeof(header.bytes)) != MEMCACHED_SUCCESS) return MEMCACHED_UNKNOWN_READ_FAILURE; unlikely (header.response.magic != PROTOCOL_BINARY_RES) - { - memcached_io_reset(ptr); return MEMCACHED_PROTOCOL_ERROR; - } /* ** Convert the header to host local endian! @@ -242,120 +322,135 @@ static memcached_return binary_response(memcached_server_st *ptr, if (header.response.status == 0) { - if ((header.response.opcode == PROTOCOL_BINARY_CMD_GETK) || - (header.response.opcode == PROTOCOL_BINARY_CMD_GETKQ)) + switch (header.response.opcode) { - uint16_t keylen= header.response.keylen; - memcached_result_reset(result); - result->cas= header.response.cas; - - if (safe_read(ptr, &result->flags, - sizeof(result->flags)) != MEMCACHED_SUCCESS) + case PROTOCOL_BINARY_CMD_GETK: + case PROTOCOL_BINARY_CMD_GETKQ: { - return MEMCACHED_UNKNOWN_READ_FAILURE; - } - result->flags= ntohl(result->flags); - bodylen -= header.response.extlen; - - result->key_length= keylen; - if (safe_read(ptr, result->key, keylen) != MEMCACHED_SUCCESS) - { - return MEMCACHED_UNKNOWN_READ_FAILURE; - } + uint16_t keylen= header.response.keylen; + memcached_result_reset(result); + result->cas= header.response.cas; - bodylen -= keylen; - if (memcached_string_check(&result->value, - bodylen) != MEMCACHED_SUCCESS) - { - memcached_io_reset(ptr); - return MEMCACHED_MEMORY_ALLOCATION_FAILURE; - } + if (memcached_safe_read(ptr, &result->flags, + sizeof (result->flags)) != MEMCACHED_SUCCESS) + return MEMCACHED_UNKNOWN_READ_FAILURE; - char *vptr= memcached_string_value(&result->value); - if (safe_read(ptr, vptr, bodylen) != MEMCACHED_SUCCESS) - { - return MEMCACHED_UNKNOWN_READ_FAILURE; - } + result->flags= ntohl(result->flags); + bodylen -= header.response.extlen; - memcached_string_set_length(&result->value, bodylen); - } - else if ((header.response.opcode == PROTOCOL_BINARY_CMD_INCREMENT) || - (header.response.opcode == PROTOCOL_BINARY_CMD_DECREMENT)) - { - if (bodylen != sizeof(uint64_t) || buffer_length != sizeof(uint64_t)) + result->key_length= keylen; + if (memcached_safe_read(ptr, result->key, keylen) != MEMCACHED_SUCCESS) + return MEMCACHED_UNKNOWN_READ_FAILURE; + + bodylen -= keylen; + if (memcached_string_check(&result->value, + bodylen) != MEMCACHED_SUCCESS) + return MEMCACHED_MEMORY_ALLOCATION_FAILURE; + + char *vptr= memcached_string_value(&result->value); + if (memcached_safe_read(ptr, vptr, bodylen) != MEMCACHED_SUCCESS) + return MEMCACHED_UNKNOWN_READ_FAILURE; + + memcached_string_set_length(&result->value, bodylen); + } + break; + case PROTOCOL_BINARY_CMD_INCREMENT: + case PROTOCOL_BINARY_CMD_DECREMENT: { - return MEMCACHED_PROTOCOL_ERROR; - } + if (bodylen != sizeof(uint64_t) || buffer_length != sizeof(uint64_t)) + return MEMCACHED_PROTOCOL_ERROR; + + WATCHPOINT_ASSERT(bodylen == buffer_length); + uint64_t val; + if (memcached_safe_read(ptr, &val, sizeof(val)) != MEMCACHED_SUCCESS) + return MEMCACHED_UNKNOWN_READ_FAILURE; - WATCHPOINT_ASSERT(bodylen == buffer_length); - uint64_t val; - if (safe_read(ptr, &val, sizeof(val)) != MEMCACHED_SUCCESS) + val= ntohll(val); + memcpy(buffer, &val, sizeof(val)); + } + break; + case PROTOCOL_BINARY_CMD_VERSION: { - return MEMCACHED_UNKNOWN_READ_FAILURE; + memset(buffer, 0, buffer_length); + if (bodylen >= buffer_length) + /* not enough space in buffer.. should not happen... */ + return MEMCACHED_UNKNOWN_READ_FAILURE; + else if (memcached_safe_read(ptr, buffer, bodylen) != MEMCACHED_SUCCESS) + return MEMCACHED_UNKNOWN_READ_FAILURE; + } + break; + case PROTOCOL_BINARY_CMD_FLUSH: + case PROTOCOL_BINARY_CMD_QUIT: + case PROTOCOL_BINARY_CMD_SET: + case PROTOCOL_BINARY_CMD_ADD: + case PROTOCOL_BINARY_CMD_REPLACE: + case PROTOCOL_BINARY_CMD_APPEND: + case PROTOCOL_BINARY_CMD_PREPEND: + case PROTOCOL_BINARY_CMD_DELETE: + { + WATCHPOINT_ASSERT(bodylen == 0); + return MEMCACHED_SUCCESS; + } + break; + case PROTOCOL_BINARY_CMD_NOOP: + { + WATCHPOINT_ASSERT(bodylen == 0); + return MEMCACHED_END; } - - val= ntohll(val); - memcpy(buffer, &val, sizeof(val)); - } - else if (header.response.opcode == PROTOCOL_BINARY_CMD_VERSION) - { - memset(buffer, 0, buffer_length); - if (bodylen >= buffer_length) - /* not enough space in buffer.. should not happen... */ - return MEMCACHED_UNKNOWN_READ_FAILURE; - else - safe_read(ptr, buffer, bodylen); - } - else if ((header.response.opcode == PROTOCOL_BINARY_CMD_FLUSH) || - (header.response.opcode == PROTOCOL_BINARY_CMD_QUIT) || - (header.response.opcode == PROTOCOL_BINARY_CMD_SET) || - (header.response.opcode == PROTOCOL_BINARY_CMD_ADD) || - (header.response.opcode == PROTOCOL_BINARY_CMD_REPLACE) || - (header.response.opcode == PROTOCOL_BINARY_CMD_APPEND) || - (header.response.opcode == PROTOCOL_BINARY_CMD_PREPEND) || - (header.response.opcode == PROTOCOL_BINARY_CMD_DELETE)) - { - WATCHPOINT_ASSERT(bodylen == 0); - return MEMCACHED_SUCCESS; - } - else if (header.response.opcode == PROTOCOL_BINARY_CMD_NOOP) - { - WATCHPOINT_ASSERT(bodylen == 0); - return MEMCACHED_END; - } - else if (header.response.opcode == PROTOCOL_BINARY_CMD_STAT) - { - if (bodylen == 0) + break; + case PROTOCOL_BINARY_CMD_STAT: + { + if (bodylen == 0) return MEMCACHED_END; - else if (bodylen + 1 > buffer_length) + else if (bodylen + 1 > buffer_length) /* not enough space in buffer.. should not happen... */ return MEMCACHED_UNKNOWN_READ_FAILURE; - else - { + else + { size_t keylen= header.response.keylen; memset(buffer, 0, buffer_length); - safe_read(ptr, buffer, keylen); - safe_read(ptr, buffer + keylen + 1, bodylen - keylen); - } - } - else - { - /* Command not implemented yet! */ - WATCHPOINT_ASSERT(0); - memcached_io_reset(ptr); - return MEMCACHED_PROTOCOL_ERROR; - } + if (memcached_safe_read(ptr, buffer, keylen) != MEMCACHED_SUCCESS || + memcached_safe_read(ptr, buffer + keylen + 1, + bodylen - keylen) != MEMCACHED_SUCCESS) + return MEMCACHED_UNKNOWN_READ_FAILURE; + } + } + break; + default: + { + /* Command not implemented yet! */ + WATCHPOINT_ASSERT(0); + return MEMCACHED_PROTOCOL_ERROR; + } + } } else if (header.response.bodylen) { /* What should I do with the error message??? just discard it for now */ - char hole[1024]; + char hole[SMALL_STRING_LEN]; while (bodylen > 0) { - size_t nr= (bodylen > sizeof(hole)) ? sizeof(hole) : bodylen; - safe_read(ptr, hole, nr); + size_t nr= (bodylen > SMALL_STRING_LEN) ? SMALL_STRING_LEN : bodylen; + if (memcached_safe_read(ptr, hole, nr) != MEMCACHED_SUCCESS) + return MEMCACHED_UNKNOWN_READ_FAILURE; bodylen -= nr; } + + /* This might be an error from one of the quiet commands.. if + * so, just throw it away and get the next one. What about creating + * a callback to the user with the error information? + */ + switch (header.response.opcode) + { + case PROTOCOL_BINARY_CMD_SETQ: + case PROTOCOL_BINARY_CMD_ADDQ: + case PROTOCOL_BINARY_CMD_REPLACEQ: + case PROTOCOL_BINARY_CMD_APPENDQ: + case PROTOCOL_BINARY_CMD_PREPENDQ: + return binary_read_one_response(ptr, buffer, buffer_length, result); + default: + break; + } } memcached_return rc= MEMCACHED_SUCCESS;