X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fmemcached_storage.c;h=ea8b1d249b98d5d1189d4d151bf5191120f3f5b6;hb=f05cd5b77ca7b17440bfc7ed9f48f7c11d269767;hp=0f817c476524811e136b589dac90d6b1a7035d6e;hpb=b0e52309fcc12077506ff7d0d7d55311d64a073e;p=awesomized%2Flibmemcached diff --git a/libmemcached/memcached_storage.c b/libmemcached/memcached_storage.c index 0f817c47..ea8b1d24 100644 --- a/libmemcached/memcached_storage.c +++ b/libmemcached/memcached_storage.c @@ -39,7 +39,7 @@ static char *storage_op_string(memcached_storage_action verb) return "tosserror"; /* This is impossible, fixes issue for compiler warning in VisualStudio */ }; - return SET_OP; + /* NOTREACHED */ } static memcached_return memcached_send_binary(memcached_server_st* server, @@ -77,7 +77,7 @@ static inline memcached_return memcached_send(memcached_st *ptr, unlikely (ptr->number_of_hosts == 0) return MEMCACHED_NO_SERVERS; - if ((ptr->flags & MEM_VERIFY_KEY) && (memcachd_key_test((char **)&key, &key_length, 1) == MEMCACHED_BAD_KEY_PROVIDED)) + if ((ptr->flags & MEM_VERIFY_KEY) && (memcached_key_test((char **)&key, &key_length, 1) == MEMCACHED_BAD_KEY_PROVIDED)) return MEMCACHED_BAD_KEY_PROVIDED; server_key= memcached_generate_hash(ptr, master_key, master_key_length); @@ -89,17 +89,30 @@ static inline memcached_return memcached_send(memcached_st *ptr, if (cas) write_length= snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, - "%s %s%.*s %u %llu %zu %llu\r\n", storage_op_string(verb), + "%s %s%.*s %u %llu %zu %llu%s\r\n", + storage_op_string(verb), ptr->prefix_key, (int)key_length, key, flags, (unsigned long long)expiration, value_length, - (unsigned long long)cas); + (unsigned long long)cas, + (ptr->flags & MEM_NOREPLY) ? " noreply" : ""); else write_length= snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, - "%s %s%.*s %u %llu %zu\r\n", storage_op_string(verb), + "%s %s%.*s %u %llu %zu%s\r\n", + storage_op_string(verb), ptr->prefix_key, (int)key_length, key, flags, - (unsigned long long)expiration, value_length); + (unsigned long long)expiration, value_length, + (ptr->flags & MEM_NOREPLY) ? " noreply" : ""); + + if (ptr->flags & MEM_USE_UDP && ptr->flags & MEM_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 + ptr->hosts[server_key].write_buffer_offset > MAX_UDP_DATAGRAM_LENGTH) + memcached_io_write(&ptr->hosts[server_key], NULL, 0, 1); + } if (write_length >= MEMCACHED_DEFAULT_COMMAND_SIZE) { @@ -124,25 +137,15 @@ static inline memcached_return memcached_send(memcached_st *ptr, else to_write= 1; - if (ptr->flags & MEM_NOREPLY) - { - if (memcached_io_write(&ptr->hosts[server_key], " noreply\r\n", - 10, 1) == -1) - { - rc= MEMCACHED_WRITE_FAILURE; - goto error; - } - - memcached_server_response_decrement(&ptr->hosts[server_key]); - return MEMCACHED_SUCCESS; - } - if ((sent_length= memcached_io_write(&ptr->hosts[server_key], "\r\n", 2, to_write)) == -1) { rc= MEMCACHED_WRITE_FAILURE; goto error; } + if (ptr->flags & MEM_NOREPLY) + return (to_write == 0) ? MEMCACHED_BUFFERED : MEMCACHED_SUCCESS; + if (to_write == 0) return MEMCACHED_BUFFERED; @@ -336,6 +339,56 @@ memcached_return memcached_cas_by_key(memcached_st *ptr, return rc; } +static inline uint8_t get_com_code(memcached_storage_action verb, bool noreply) +{ + uint8_t ret; + + if (noreply) + switch (verb) + { + case SET_OP: + ret=PROTOCOL_BINARY_CMD_SETQ; + break; + case ADD_OP: + ret=PROTOCOL_BINARY_CMD_ADDQ; + break; + case CAS_OP: /* FALLTHROUGH */ + case REPLACE_OP: + ret=PROTOCOL_BINARY_CMD_REPLACEQ; + break; + case APPEND_OP: + ret=PROTOCOL_BINARY_CMD_APPENDQ; + break; + case PREPEND_OP: + ret=PROTOCOL_BINARY_CMD_PREPENDQ; + break; + } + else + switch (verb) + { + case SET_OP: + ret=PROTOCOL_BINARY_CMD_SET; + break; + case ADD_OP: + ret=PROTOCOL_BINARY_CMD_ADD; + break; + case CAS_OP: /* FALLTHROUGH */ + case REPLACE_OP: + ret=PROTOCOL_BINARY_CMD_REPLACE; + break; + case APPEND_OP: + ret=PROTOCOL_BINARY_CMD_APPEND; + break; + case PREPEND_OP: + ret=PROTOCOL_BINARY_CMD_PREPEND; + break; + } + + return ret; +} + + + static memcached_return memcached_send_binary(memcached_server_st* server, const char *key, size_t key_length, @@ -349,30 +402,10 @@ static memcached_return memcached_send_binary(memcached_server_st* server, char flush; protocol_binary_request_set request= {.bytes= {0}}; size_t send_length= sizeof(request.bytes); + bool noreply= server->root->flags & MEM_NOREPLY; request.message.header.request.magic= PROTOCOL_BINARY_REQ; - switch (verb) - { - case SET_OP: - request.message.header.request.opcode= PROTOCOL_BINARY_CMD_SET; - break; - case ADD_OP: - request.message.header.request.opcode= PROTOCOL_BINARY_CMD_ADD; - break; - case REPLACE_OP: - request.message.header.request.opcode= PROTOCOL_BINARY_CMD_REPLACE; - break; - case APPEND_OP: - request.message.header.request.opcode= PROTOCOL_BINARY_CMD_APPEND; - break; - case PREPEND_OP: - request.message.header.request.opcode= PROTOCOL_BINARY_CMD_PREPEND; - break; - case CAS_OP: - request.message.header.request.opcode= PROTOCOL_BINARY_CMD_REPLACE; - break; - } - + request.message.header.request.opcode= get_com_code(verb, noreply); request.message.header.request.keylen= htons((uint16_t)key_length); request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES; if (verb == APPEND_OP || verb == PREPEND_OP) @@ -392,17 +425,15 @@ static memcached_return memcached_send_binary(memcached_server_st* server, flush= ((server->root->flags & MEM_BUFFER_REQUESTS) && verb == SET_OP) ? 0 : 1; - /* The binary protocol does not implement NOREPLY right now, so I need to - * parse the return message from the server. The problem we tried to solve - * is the fact that the async mode will buffer commands, and if we issue a - * lot of sets followed by a get we have to flush the send buffer before we - * can send the get command (and all those commands have to be executed). - * The workaround for the binary is that we send the command, but we await - * parsing of the result message until we really need to do it.. - */ - if (server->root->flags & MEM_NOREPLY) - flush=1; - + if ((server->root->flags & MEM_USE_UDP) && !flush) + { + size_t cmd_size= send_length + key_length + value_length; + if (cmd_size > MAX_UDP_DATAGRAM_LENGTH - UDP_DATAGRAM_HEADER_LENGTH) + return MEMCACHED_WRITE_FAILURE; + if (cmd_size + server->write_buffer_offset > MAX_UDP_DATAGRAM_LENGTH) + memcached_io_write(server,NULL,0, 1); + } + /* 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) || @@ -411,10 +442,13 @@ static memcached_return memcached_send_binary(memcached_server_st* server, memcached_io_reset(server); return MEMCACHED_WRITE_FAILURE; } - - if (flush == 0 || server->root->flags & MEM_NOREPLY) + + if (flush == 0) return MEMCACHED_BUFFERED; + if (noreply) + return MEMCACHED_SUCCESS; + return memcached_response(server, NULL, 0, NULL); }