X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fstorage.c;h=dce5fec324870e19f316d02d0efe02b5e6b5d736;hb=6b46b12051fa265bbd9c2a17899a280c8bd82769;hp=6b4eb623f7eb005cca70e6f80a64f152029c075a;hpb=9b767485048053ad0b4c2897a0d3784aa6e9fe10;p=awesomized%2Flibmemcached diff --git a/libmemcached/storage.c b/libmemcached/storage.c index 6b4eb623..dce5fec3 100644 --- a/libmemcached/storage.c +++ b/libmemcached/storage.c @@ -45,8 +45,8 @@ static inline const char *storage_op_string(memcached_storage_action_t verb) } static memcached_return_t memcached_send_binary(memcached_st *ptr, - const char *master_key, - size_t master_key_length, + memcached_server_write_instance_st server, + uint32_t server_key, const char *key, size_t key_length, const char *value, @@ -65,12 +65,12 @@ static inline memcached_return_t memcached_send(memcached_st *ptr, uint64_t cas, memcached_storage_action_t verb) { - char to_write; + bool to_write; size_t write_length; memcached_return_t rc; char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE]; uint32_t server_key; - memcached_server_instance_st *instance; + memcached_server_write_instance_st instance; WATCHPOINT_ASSERT(!(value == NULL && value_length > 0)); @@ -84,112 +84,122 @@ static inline memcached_return_t memcached_send(memcached_st *ptr, if (ptr->flags.verify_key && (memcached_key_test((const char **)&key, &key_length, 1) == MEMCACHED_BAD_KEY_PROVIDED)) return MEMCACHED_BAD_KEY_PROVIDED; - if (ptr->flags.binary_protocol) - { - 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); + server_key= memcached_generate_hash_with_redistribution(ptr, master_key, master_key_length); instance= memcached_server_instance_fetch(ptr, server_key); - if (cas) - { - write_length= (size_t) snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, - "%s %.*s%.*s %u %llu %zu %llu%s\r\n", - storage_op_string(verb), - (int)ptr->prefix_key_length, - ptr->prefix_key, - (int)key_length, key, flags, - (unsigned long long)expiration, value_length, - (unsigned long long)cas, - (ptr->flags.no_reply) ? " noreply" : ""); - } - else - { - char *buffer_ptr= buffer; - const char *command= storage_op_string(verb); - - /* Copy in the command, no space needed, we handle that in the command function*/ - memcpy(buffer_ptr, command, strlen(command)); - - /* Copy in the key prefix, switch to the buffer_ptr */ - buffer_ptr= memcpy((buffer_ptr + strlen(command)), ptr->prefix_key, ptr->prefix_key_length); - - /* Copy in the key, adjust point if a key prefix was used. */ - buffer_ptr= memcpy(buffer_ptr + (ptr->prefix_key_length ? ptr->prefix_key_length : 0), - key, key_length); - buffer_ptr+= key_length; - buffer_ptr[0]= ' '; - buffer_ptr++; - - write_length= (size_t)(buffer_ptr - buffer); - 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.no_reply ? " noreply" : ""); - } + WATCHPOINT_SET(instance->io_wait_count.read= 0); + WATCHPOINT_SET(instance->io_wait_count.write= 0); - if (ptr->flags.use_udp && ptr->flags.buffer_requests) + if (ptr->flags.binary_protocol) { - size_t cmd_size= write_length + value_length + 2; - if (cmd_size > MAX_UDP_DATAGRAM_LENGTH - UDP_DATAGRAM_HEADER_LENGTH) - return MEMCACHED_WRITE_FAILURE; - if (cmd_size + instance->write_buffer_offset > MAX_UDP_DATAGRAM_LENGTH) - memcached_io_write(instance, NULL, 0, true); + rc= memcached_send_binary(ptr, instance, server_key, + key, key_length, + value, value_length, expiration, + flags, cas, verb); + WATCHPOINT_IF_LABELED_NUMBER(instance->io_wait_count.read > 2, "read IO_WAIT", instance->io_wait_count.read); + WATCHPOINT_IF_LABELED_NUMBER(instance->io_wait_count.write > 2, "write_IO_WAIT", instance->io_wait_count.write); } - - if (write_length >= MEMCACHED_DEFAULT_COMMAND_SIZE) + else { - rc= MEMCACHED_WRITE_FAILURE; - goto error; - } - /* Send command header */ - rc= memcached_do(instance, buffer, write_length, 0); - if (rc != MEMCACHED_SUCCESS) - goto error; - - /* Send command body */ - if (memcached_io_write(instance, value, value_length, false) == -1) - { - rc= MEMCACHED_WRITE_FAILURE; - goto error; - } + if (cas) + { + write_length= (size_t) snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, + "%s %.*s%.*s %u %llu %zu %llu%s\r\n", + storage_op_string(verb), + (int)ptr->prefix_key_length, + ptr->prefix_key, + (int)key_length, key, flags, + (unsigned long long)expiration, value_length, + (unsigned long long)cas, + (ptr->flags.no_reply) ? " noreply" : ""); + } + else + { + char *buffer_ptr= buffer; + const char *command= storage_op_string(verb); + + /* Copy in the command, no space needed, we handle that in the command function*/ + memcpy(buffer_ptr, command, strlen(command)); + + /* Copy in the key prefix, switch to the buffer_ptr */ + buffer_ptr= memcpy((buffer_ptr + strlen(command)), ptr->prefix_key, ptr->prefix_key_length); + + /* Copy in the key, adjust point if a key prefix was used. */ + buffer_ptr= memcpy(buffer_ptr + (ptr->prefix_key_length ? ptr->prefix_key_length : 0), + key, key_length); + buffer_ptr+= key_length; + buffer_ptr[0]= ' '; + buffer_ptr++; + + write_length= (size_t)(buffer_ptr - buffer); + 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.no_reply ? " noreply" : ""); + } - if (ptr->flags.buffer_requests && verb == SET_OP) - { - to_write= 0; - } - else - { - to_write= 1; - } + if (ptr->flags.use_udp && ptr->flags.buffer_requests) + { + size_t cmd_size= write_length + value_length + 2; + if (cmd_size > MAX_UDP_DATAGRAM_LENGTH - UDP_DATAGRAM_HEADER_LENGTH) + return MEMCACHED_WRITE_FAILURE; + if (cmd_size + instance->write_buffer_offset > MAX_UDP_DATAGRAM_LENGTH) + memcached_io_write(instance, NULL, 0, true); + } - if (memcached_io_write(instance, "\r\n", 2, to_write) == -1) - { - rc= MEMCACHED_WRITE_FAILURE; - goto error; - } + if (write_length >= MEMCACHED_DEFAULT_COMMAND_SIZE) + { + rc= MEMCACHED_WRITE_FAILURE; + } + else + { + struct __write_vector_st vector[]= + { + { .length= write_length, .buffer= buffer }, + { .length= value_length, .buffer= value }, + { .length= 2, .buffer= "\r\n" } + }; - if (ptr->flags.no_reply) - return (to_write == 0) ? MEMCACHED_BUFFERED : MEMCACHED_SUCCESS; + if (ptr->flags.buffer_requests && verb == SET_OP) + { + to_write= false; + } + else + { + to_write= true; + } - if (to_write == 0) - return MEMCACHED_BUFFERED; + /* Send command header */ + rc= memcached_vdo(instance, vector, 3, to_write); + if (rc == MEMCACHED_SUCCESS) + { - rc= memcached_response(instance, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL); + if (ptr->flags.no_reply) + { + rc= (to_write == false) ? MEMCACHED_BUFFERED : MEMCACHED_SUCCESS; + } + else if (to_write == false) + { + rc= MEMCACHED_BUFFERED; + } + else + { + rc= memcached_response(instance, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL); + + if (rc == MEMCACHED_STORED) + rc= MEMCACHED_SUCCESS; + } + } + } - if (rc == MEMCACHED_STORED) - return MEMCACHED_SUCCESS; - else - return rc; + if (rc == MEMCACHED_WRITE_FAILURE) + memcached_io_reset(instance); + } -error: - memcached_io_reset(instance); + WATCHPOINT_IF_LABELED_NUMBER(instance->io_wait_count.read > 2, "read IO_WAIT", instance->io_wait_count.read); + WATCHPOINT_IF_LABELED_NUMBER(instance->io_wait_count.write > 2, "write_IO_WAIT", instance->io_wait_count.write); return rc; } @@ -430,8 +440,8 @@ static inline uint8_t get_com_code(memcached_storage_action_t verb, bool noreply static memcached_return_t memcached_send_binary(memcached_st *ptr, - const char *master_key, - size_t master_key_length, + memcached_server_write_instance_st server, + uint32_t server_key, const char *key, size_t key_length, const char *value, @@ -441,20 +451,15 @@ static memcached_return_t memcached_send_binary(memcached_st *ptr, uint64_t cas, memcached_storage_action_t verb) { - uint8_t flush; + bool 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_instance_st *server= - memcached_server_instance_fetch(ptr, server_key); bool noreply= server->root->flags.no_reply; request.message.header.request.magic= PROTOCOL_BINARY_REQ; request.message.header.request.opcode= get_com_code(verb, noreply); - request.message.header.request.keylen= htons((uint16_t)key_length); + request.message.header.request.keylen= htons((uint16_t)(key_length + ptr->prefix_key_length)); request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES; if (verb == APPEND_OP || verb == PREPEND_OP) send_length -= 8; /* append & prepend does not contain extras! */ @@ -465,15 +470,15 @@ static memcached_return_t memcached_send_binary(memcached_st *ptr, request.message.body.expiration= htonl((uint32_t)expiration); } - request.message.header.request.bodylen= htonl((uint32_t) (key_length + value_length + + request.message.header.request.bodylen= htonl((uint32_t) (key_length + ptr->prefix_key_length + value_length + request.message.header.request.extlen)); if (cas) request.message.header.request.cas= htonll(cas); - flush= (uint8_t) ((server->root->flags.buffer_requests && verb == SET_OP) ? 0 : 1); + flush= (bool) ((server->root->flags.buffer_requests && verb == SET_OP) ? 0 : 1); - if (server->root->flags.use_udp && !flush) + if (server->root->flags.use_udp && ! flush) { size_t cmd_size= send_length + key_length + value_length; @@ -487,22 +492,30 @@ static memcached_return_t memcached_send_binary(memcached_st *ptr, } } + struct __write_vector_st vector[]= + { + { .length= send_length, .buffer= request.bytes }, + { .length= ptr->prefix_key_length, .buffer= ptr->prefix_key }, + { .length= key_length, .buffer= key }, + { .length= value_length, .buffer= value } + }; + /* write the header */ - if ((memcached_do(server, (const char*)request.bytes, send_length, 0) != MEMCACHED_SUCCESS) || - (memcached_io_write(server, key, key_length, false) == -1) || - (memcached_io_write(server, value, value_length, (char) flush) == -1)) + memcached_return_t rc; + if ((rc= memcached_vdo(server, vector, 4, flush)) != MEMCACHED_SUCCESS) { memcached_io_reset(server); - return MEMCACHED_WRITE_FAILURE; + return (rc == MEMCACHED_SUCCESS) ? MEMCACHED_WRITE_FAILURE : rc; } - unlikely (verb == SET_OP && ptr->number_of_replicas > 0) + if (verb == SET_OP && ptr->number_of_replicas > 0) { request.message.header.request.opcode= PROTOCOL_BINARY_CMD_SETQ; + WATCHPOINT_STRING("replicating"); for (uint32_t x= 0; x < ptr->number_of_replicas; x++) { - memcached_server_instance_st *instance; + memcached_server_write_instance_st instance; ++server_key; if (server_key == memcached_server_count(ptr)) @@ -510,10 +523,7 @@ static memcached_return_t memcached_send_binary(memcached_st *ptr, instance= memcached_server_instance_fetch(ptr, server_key); - if ((memcached_do(instance, (const char*)request.bytes, - send_length, 0) != MEMCACHED_SUCCESS) || - (memcached_io_write(instance, key, key_length, false) == -1) || - (memcached_io_write(instance, value, value_length, (char) flush) == -1)) + if (memcached_vdo(instance, vector, 4, false) != MEMCACHED_SUCCESS) { memcached_io_reset(instance); } @@ -524,7 +534,7 @@ static memcached_return_t memcached_send_binary(memcached_st *ptr, } } - if (flush == 0) + if (flush == false) { return MEMCACHED_BUFFERED; }