X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fstorage.c;h=be0040a4197f6b621f0a900e662c92117ea8b89b;hb=d5c34fa54897336e3c92789d1abcceca8c5b213b;hp=c0db470ca80fd5f1b2c54890c1f334ab746e21ff;hpb=3928e14fbe1ed17cefe9bdbbb282fb7ecf053e7a;p=awesomized%2Flibmemcached diff --git a/libmemcached/storage.c b/libmemcached/storage.c index c0db470c..be0040a4 100644 --- a/libmemcached/storage.c +++ b/libmemcached/storage.c @@ -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)); @@ -92,14 +92,15 @@ static inline memcached_return_t memcached_send(memcached_st *ptr, 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", + "%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, @@ -115,10 +116,10 @@ static inline memcached_return_t memcached_send(memcached_st *ptr, 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, strlen(ptr->prefix_key)); + 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 ? strlen(ptr->prefix_key) : 0), + buffer_ptr= memcpy(buffer_ptr + (ptr->prefix_key_length ? ptr->prefix_key_length : 0), key, key_length); buffer_ptr+= key_length; buffer_ptr[0]= ' '; @@ -138,57 +139,53 @@ static inline memcached_return_t memcached_send(memcached_st *ptr, 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, 1); + memcached_io_write(instance, NULL, 0, true); } if (write_length >= MEMCACHED_DEFAULT_COMMAND_SIZE) { 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, 0) == -1) - { - rc= MEMCACHED_WRITE_FAILURE; - goto error; - } - - if (ptr->flags.buffer_requests && verb == SET_OP) - { - to_write= 0; } else { - to_write= 1; - } + struct __write_vector_st vector[]= + { + { .length= write_length, .buffer= buffer }, + { .length= value_length, .buffer= value }, + { .length= 2, .buffer= "\r\n" } + }; - if (memcached_io_write(instance, "\r\n", 2, to_write) == -1) - { - rc= MEMCACHED_WRITE_FAILURE; - goto error; - } + if (ptr->flags.buffer_requests && verb == SET_OP) + { + to_write= false; + } + else + { + to_write= true; + } - if (ptr->flags.no_reply) - return (to_write == 0) ? MEMCACHED_BUFFERED : MEMCACHED_SUCCESS; + /* Send command header */ + rc= memcached_vdo(instance, vector, 3, to_write); + if (rc == MEMCACHED_SUCCESS) + { - if (to_write == 0) - return MEMCACHED_BUFFERED; + if (ptr->flags.no_reply) + return (to_write == false) ? MEMCACHED_BUFFERED : MEMCACHED_SUCCESS; - rc= memcached_response(instance, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL); + if (to_write == false) + return MEMCACHED_BUFFERED; - if (rc == MEMCACHED_STORED) - return MEMCACHED_SUCCESS; - else - return rc; + rc= memcached_response(instance, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL); -error: - memcached_io_reset(instance); + if (rc == MEMCACHED_STORED) + return MEMCACHED_SUCCESS; + else + return rc; + } + } + + if (rc == MEMCACHED_WRITE_FAILURE) + memcached_io_reset(instance); return rc; } @@ -440,20 +437,20 @@ 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); + uint32_t server_key= memcached_generate_hash_with_redistribution(ptr, master_key, + master_key_length); - memcached_server_instance_st *server= + memcached_server_write_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! */ @@ -464,15 +461,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; @@ -482,17 +479,24 @@ static memcached_return_t memcached_send_binary(memcached_st *ptr, } if (cmd_size + server->write_buffer_offset > MAX_UDP_DATAGRAM_LENGTH) { - memcached_io_write(server, NULL, 0, 1); + memcached_io_write(server, NULL, 0, true); } } + 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, 0) == -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) @@ -501,7 +505,7 @@ static memcached_return_t memcached_send_binary(memcached_st *ptr, 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)) @@ -509,10 +513,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, 0) == -1) || - (memcached_io_write(instance, value, value_length, (char) flush) == -1)) + if (memcached_vdo(instance, vector, 4, false) != MEMCACHED_SUCCESS) { memcached_io_reset(instance); } @@ -523,7 +524,7 @@ static memcached_return_t memcached_send_binary(memcached_st *ptr, } } - if (flush == 0) + if (flush == false) { return MEMCACHED_BUFFERED; }