X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fprotocol%2Fprotocol_handler.c;h=fcab1419f7c100083a232b0c405cb0ccf65da806;hb=609d07c5a051c301ce6595747c2f64d3819554f5;hp=67f2b63ca200e9aacd47afdce83f566eb6785e85;hpb=e6a6807ad79f55f49826d3c76498373197ad59a7;p=m6w6%2Flibmemcached diff --git a/libmemcached/protocol/protocol_handler.c b/libmemcached/protocol/protocol_handler.c index 67f2b63c..fcab1419 100644 --- a/libmemcached/protocol/protocol_handler.c +++ b/libmemcached/protocol/protocol_handler.c @@ -1,12 +1,13 @@ /* -*- Mode: C; tab-width: 2; c-basic-offset: 2; indent-tabs-mode: nil -*- */ -#include "common.h" +#include "libmemcached/protocol/common.h" #include #include -#include #include #include #include +#include +#include #include /* @@ -27,7 +28,7 @@ * @return the number of bytes transferred of -1 upon error */ static ssize_t default_recv(const void *cookie, - int sock, + memcached_socket_t sock, void *buf, size_t nbytes) { @@ -47,7 +48,7 @@ static ssize_t default_recv(const void *cookie, * @return the number of bytes transferred of -1 upon error */ static ssize_t default_send(const void *cookie, - int fd, + memcached_socket_t fd, const void *buf, size_t nbytes) { @@ -63,11 +64,11 @@ static ssize_t default_send(const void *cookie, * true otherwise (please note that there may be more data to * left in the buffer to send) */ -static bool drain_output(struct memcached_binary_protocol_client_st *client) +static bool drain_output(struct memcached_protocol_client_st *client) { ssize_t len; - // Do we have pending data to send? + /* Do we have pending data to send? */ while (client->output != NULL) { len= client->root->send(client, @@ -77,13 +78,13 @@ static bool drain_output(struct memcached_binary_protocol_client_st *client) if (len == -1) { - if (errno == EWOULDBLOCK) + if (get_socket_errno() == EWOULDBLOCK) { return true; } - else if (errno != EINTR) + else if (get_socket_errno() != EINTR) { - client->error= errno; + client->error= get_socket_errno(); return false; } } @@ -113,21 +114,22 @@ static bool drain_output(struct memcached_binary_protocol_client_st *client) * @param client the client that needs the buffer * @return pointer to the new chunk if the allocation succeeds, NULL otherwise */ -static struct chunk_st* -allocate_output_chunk(struct memcached_binary_protocol_client_st *client) +static struct chunk_st *allocate_output_chunk(struct memcached_protocol_client_st *client) { - struct chunk_st*ret= cache_alloc(client->root->buffer_cache); - if (ret == NULL) { + struct chunk_st *ret= cache_alloc(client->root->buffer_cache); + + if (ret == NULL) + { return NULL; } - ret->offset = ret->nbytes = 0; - ret->next = NULL; - ret->size = CHUNK_BUFFERSIZE; + ret->offset= ret->nbytes= 0; + ret->next= NULL; + ret->size= CHUNK_BUFFERSIZE; ret->data= (void*)(ret + 1); if (client->output == NULL) { - client->output = client->output_tail = ret; + client->output= client->output_tail= ret; } else { @@ -147,12 +149,16 @@ allocate_output_chunk(struct memcached_binary_protocol_client_st *client) * @return PROTOCOL_BINARY_RESPONSE_SUCCESS if success, * PROTOCOL_BINARY_RESPONSE_ENOMEM if we failed to allocate memory */ -static protocol_binary_response_status -spool_output(struct memcached_binary_protocol_client_st *client, - const void *data, - size_t length) +static protocol_binary_response_status spool_output(struct memcached_protocol_client_st *client, + const void *data, + size_t length) { - size_t offset = 0; + if (client->mute) + { + return PROTOCOL_BINARY_RESPONSE_SUCCESS; + } + + size_t offset= 0; struct chunk_st *chunk= client->output; while (offset < length) @@ -165,10 +171,10 @@ spool_output(struct memcached_binary_protocol_client_st *client, } } - size_t bulk = length - offset; + size_t bulk= length - offset; if (bulk > chunk->size - chunk->nbytes) { - bulk = chunk->size - chunk->nbytes; + bulk= chunk->size - chunk->nbytes; } memcpy(chunk->data + chunk->nbytes, data, bulk); @@ -180,985 +186,43 @@ spool_output(struct memcached_binary_protocol_client_st *client, } /** - * Send a preformatted packet back to the client. If the connection is in - * pedantic mode, it will validate the packet and refuse to send it if it - * breaks the specification. + * Try to determine the protocol used on this connection. + * If the first byte contains the magic byte PROTOCOL_BINARY_REQ we should + * be using the binary protocol on the connection. I implemented the support + * for the ASCII protocol by wrapping into the simple interface (aka v1), + * so the implementors needs to provide an implementation of that interface * - * @param cookie client identification - * @param request the original request packet - * @param response the packet to send - * @return The status of the operation - */ -static protocol_binary_response_status -raw_response_handler(const void *cookie, - protocol_binary_request_header *request, - protocol_binary_response_header *response) -{ - struct memcached_binary_protocol_client_st *client= (void*)cookie; - - if (client->root->pedantic && - !memcached_binary_protocol_pedantic_check_response(request, response)) - { - return PROTOCOL_BINARY_RESPONSE_EINVAL; - } - - if (!drain_output(client)) - { - return PROTOCOL_BINARY_RESPONSE_EIO; - } - - size_t len= sizeof(*response) + htonl(response->response.bodylen); - size_t offset= 0; - char *ptr= (void*)response; - - if (client->output == NULL) - { - /* I can write directly to the socket.... */ - do - { - size_t num_bytes= len - offset; - ssize_t nw= client->root->send(client, - client->sock, - ptr + offset, - num_bytes); - if (nw == -1) - { - if (errno == EWOULDBLOCK) - { - break; - } - else if (errno != EINTR) - { - client->error= errno; - return PROTOCOL_BINARY_RESPONSE_EIO; - } - } - else - { - offset += (size_t)nw; - } - } while (offset < len); - } - - return spool_output(client, ptr, len - offset); -} - -/* - * Version 0 of the interface is really low level and protocol specific, - * while the version 1 of the interface is more API focused. We need a - * way to translate between the command codes on the wire and the - * application level interface in V1, so let's just use the V0 of the - * interface as a map instead of creating a huuuge switch :-) - */ - -/** - * Callback for the GET/GETQ/GETK and GETKQ responses - * @param cookie client identifier - * @param key the key for the item - * @param keylen the length of the key - * @param body the length of the body - * @param bodylen the length of the body - * @param flags the flags for the item - * @param cas the CAS id for the item - */ -static protocol_binary_response_status -get_response_handler(const void *cookie, - const void *key, - uint16_t keylen, - const void *body, - uint32_t bodylen, - uint32_t flags, - uint64_t cas) { - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - uint8_t opcode= client->current_command->request.opcode; - - if (opcode == PROTOCOL_BINARY_CMD_GET || opcode == PROTOCOL_BINARY_CMD_GETQ) - { - keylen= 0; - } - - protocol_binary_response_get response= { - .message.header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= opcode, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= client->current_command->request.opaque, - .cas= htonll(cas), - .keylen= htons(keylen), - .extlen= 4, - .bodylen= htonl(bodylen + keylen + 4), - }, - .message.body.flags= htonl(flags), - }; - - protocol_binary_response_status rval; - const protocol_binary_response_status success = PROTOCOL_BINARY_RESPONSE_SUCCESS; - if ((rval= spool_output(client, response.bytes, sizeof(response.bytes))) != success || - (rval= spool_output(client, key, keylen)) != success || - (rval= spool_output(client, body, bodylen)) != success) - { - return rval; - } - - return PROTOCOL_BINARY_RESPONSE_SUCCESS; -} - -/** - * Callback for the STAT responses - * @param cookie client identifier - * @param key the key for the item - * @param keylen the length of the key - * @param body the length of the body - * @param bodylen the length of the body - */ -static protocol_binary_response_status -stat_response_handler(const void *cookie, - const void *key, - uint16_t keylen, - const void *body, - uint32_t bodylen) { - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - - protocol_binary_response_no_extras response= { - .message.header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= client->current_command->request.opcode, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= client->current_command->request.opaque, - .keylen= htons(keylen), - .bodylen= htonl(bodylen + keylen), - }, - }; - - protocol_binary_response_status rval; - const protocol_binary_response_status success = PROTOCOL_BINARY_RESPONSE_SUCCESS; - if ((rval= spool_output(client, response.bytes, sizeof(response.bytes))) != success || - (rval= spool_output(client, key, keylen)) != success || - (rval= spool_output(client, body, bodylen)) != success) - { - return rval; - } - - return PROTOCOL_BINARY_RESPONSE_SUCCESS; -} - -/** - * Callback for the VERSION responses - * @param cookie client identifier - * @param text the length of the body - * @param textlen the length of the body - */ -static protocol_binary_response_status -version_response_handler(const void *cookie, - const void *text, - uint32_t textlen) { - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - - protocol_binary_response_no_extras response= { - .message.header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= client->current_command->request.opcode, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= client->current_command->request.opaque, - .bodylen= htonl(textlen), - }, - }; - - protocol_binary_response_status rval; - const protocol_binary_response_status success = PROTOCOL_BINARY_RESPONSE_SUCCESS; - if ((rval= spool_output(client, response.bytes, sizeof(response.bytes))) != success || - (rval= spool_output(client, text, textlen)) != success) - { - return rval; - } - - return PROTOCOL_BINARY_RESPONSE_SUCCESS; -} - -/** - * Callback for ADD and ADDQ - * @param cookie the calling client - * @param header the add/addq command - * @param response_handler not used - * @return the result of the operation - */ -static protocol_binary_response_status -add_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - protocol_binary_response_status rval; - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.add != NULL) - { - uint16_t keylen= ntohs(header->request.keylen); - uint32_t datalen= ntohl(header->request.bodylen) - keylen - 8; - protocol_binary_request_add *request= (void*)header; - uint32_t flags= ntohl(request->message.body.flags); - uint32_t timeout= ntohl(request->message.body.expiration); - char *key= ((char*)header) + sizeof(*header) + 8; - char *data= key + keylen; - uint64_t cas; - - rval= client->root->callback->interface.v1.add(cookie, key, keylen, - data, datalen, flags, - timeout, &cas); - - if (rval == PROTOCOL_BINARY_RESPONSE_SUCCESS && - header->request.opcode == PROTOCOL_BINARY_CMD_ADD) - { - /* Send a positive request */ - protocol_binary_response_no_extras response= { - .message= { - .header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= PROTOCOL_BINARY_CMD_ADD, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= header->request.opaque, - .cas= ntohll(cas) - } - } - }; - rval= response_handler(cookie, header, (void*)&response); - } - } - else - { - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - } - - return rval; -} - -/** - * Callback for DECREMENT and DECREMENTQ - * @param cookie the calling client - * @param header the command - * @param response_handler not used - * @return the result of the operation - */ -static protocol_binary_response_status -decrement_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - (void)response_handler; - protocol_binary_response_status rval; - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.decrement != NULL) - { - uint16_t keylen= ntohs(header->request.keylen); - protocol_binary_request_decr *request= (void*)header; - uint64_t init= ntohll(request->message.body.initial); - uint64_t delta= ntohll(request->message.body.delta); - uint32_t timeout= ntohl(request->message.body.expiration); - void *key= request->bytes + sizeof(request->bytes); - uint64_t result; - uint64_t cas; - - char buffer[1024] = {0}; - memcpy(buffer, key, keylen); - fprintf(stderr, "%s\n", buffer); - - - rval= client->root->callback->interface.v1.decrement(cookie, key, keylen, - delta, init, timeout, - &result, &cas); - if (rval == PROTOCOL_BINARY_RESPONSE_SUCCESS && - header->request.opcode == PROTOCOL_BINARY_CMD_DECREMENT) - { - /* Send a positive request */ - protocol_binary_response_decr response= { - .message= { - .header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= PROTOCOL_BINARY_CMD_DECREMENT, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= header->request.opaque, - .cas= ntohll(cas), - .bodylen= htonl(8) - }, - .body.value = htonll(result) - } - }; - rval= response_handler(cookie, header, (void*)&response); - } - } - else - { - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - } - - return rval; -} - -/** - * Callback for DELETE and DELETEQ - * @param cookie the calling client - * @param header the command - * @param response_handler not used - * @return the result of the operation - */ -static protocol_binary_response_status -delete_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - (void)response_handler; - protocol_binary_response_status rval; - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.delete != NULL) - { - uint16_t keylen= ntohs(header->request.keylen); - void *key= (header + 1); - uint64_t cas= ntohll(header->request.cas); - rval= client->root->callback->interface.v1.delete(cookie, key, keylen, cas); - if (rval == PROTOCOL_BINARY_RESPONSE_SUCCESS && - header->request.opcode == PROTOCOL_BINARY_CMD_DELETE) - { - /* Send a positive request */ - protocol_binary_response_no_extras response= { - .message= { - .header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= PROTOCOL_BINARY_CMD_DELETE, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= header->request.opaque, - } - } - }; - rval= response_handler(cookie, header, (void*)&response); - } - } - else - { - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - } - - return rval; -} - -/** - * Callback for FLUSH and FLUSHQ - * @param cookie the calling client - * @param header the command - * @param response_handler not used - * @return the result of the operation - */ -static protocol_binary_response_status -flush_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - (void)response_handler; - protocol_binary_response_status rval; - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.flush != NULL) - { - protocol_binary_request_flush *flush= (void*)header; - uint32_t timeout= 0; - if (htonl(header->request.bodylen) == 4) - { - timeout= ntohl(flush->message.body.expiration); - } - - rval= client->root->callback->interface.v1.flush(cookie, timeout); - if (rval == PROTOCOL_BINARY_RESPONSE_SUCCESS && - header->request.opcode == PROTOCOL_BINARY_CMD_FLUSH) - { - /* Send a positive request */ - protocol_binary_response_no_extras response= { - .message= { - .header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= PROTOCOL_BINARY_CMD_FLUSH, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= header->request.opaque, - } - } - }; - rval= response_handler(cookie, header, (void*)&response); - } - } - else - { - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - } - - return rval; -} - -/** - * Callback for GET, GETK, GETQ, GETKQ - * @param cookie the calling client - * @param header the command - * @param response_handler not used - * @return the result of the operation - */ -static protocol_binary_response_status -get_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - (void)response_handler; - protocol_binary_response_status rval; - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.get != NULL) - { - uint16_t keylen= ntohs(header->request.keylen); - void *key= (header + 1); - rval= client->root->callback->interface.v1.get(cookie, key, keylen, - get_response_handler); - - if (rval == PROTOCOL_BINARY_RESPONSE_KEY_ENOENT && - (header->request.opcode == PROTOCOL_BINARY_CMD_GETQ || - header->request.opcode == PROTOCOL_BINARY_CMD_GETKQ)) - { - /* Quiet commands shouldn't respond on cache misses */ - rval= PROTOCOL_BINARY_RESPONSE_SUCCESS; - } - } - else - { - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - } - - return rval; -} - -/** - * Callback for INCREMENT and INCREMENTQ - * @param cookie the calling client - * @param header the command - * @param response_handler not used - * @return the result of the operation */ -static protocol_binary_response_status -increment_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) +static memcached_protocol_event_t determine_protocol(struct memcached_protocol_client_st *client, ssize_t *length, void **endptr) { - (void)response_handler; - protocol_binary_response_status rval; - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.increment != NULL) + if (*client->root->input_buffer == (uint8_t)PROTOCOL_BINARY_REQ) { - uint16_t keylen= ntohs(header->request.keylen); - protocol_binary_request_incr *request= (void*)header; - uint64_t init= ntohll(request->message.body.initial); - uint64_t delta= ntohll(request->message.body.delta); - uint32_t timeout= ntohl(request->message.body.expiration); - void *key= request->bytes + sizeof(request->bytes); - uint64_t cas; - uint64_t result; - - rval= client->root->callback->interface.v1.increment(cookie, key, keylen, - delta, init, timeout, - &result, &cas); - if (rval == PROTOCOL_BINARY_RESPONSE_SUCCESS && - header->request.opcode == PROTOCOL_BINARY_CMD_INCREMENT) - { - /* Send a positive request */ - protocol_binary_response_incr response= { - .message= { - .header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= PROTOCOL_BINARY_CMD_INCREMENT, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= header->request.opaque, - .cas= ntohll(cas), - .bodylen= htonl(8) - }, - .body.value = htonll(result) - } - }; - rval= response_handler(cookie, header, (void*)&response); - } + client->work= memcached_binary_protocol_process_data; } - else + else if (client->root->callback->interface_version == 1) { - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - } - - return rval; -} - -/** - * Callback for noop. Inform the v1 interface about the noop packet, and - * create and send a packet back to the client - * - * @param cookie the calling client - * @param header the command - * @param response_handler the response handler - * @return the result of the operation - */ -static protocol_binary_response_status -noop_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.noop != NULL) - { - client->root->callback->interface.v1.noop(cookie); - } - - protocol_binary_response_no_extras response= { - .message = { - .header.response = { - .magic = PROTOCOL_BINARY_RES, - .opcode= PROTOCOL_BINARY_CMD_NOOP, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= header->request.opaque, - } - } - }; - - return response_handler(cookie, header, (void*)&response); -} - -/** - * Callback for APPEND and APPENDQ - * @param cookie the calling client - * @param header the command - * @param response_handler not used - * @return the result of the operation - */ -static protocol_binary_response_status -append_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - (void)response_handler; - protocol_binary_response_status rval; - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.append != NULL) - { - uint16_t keylen= ntohs(header->request.keylen); - uint32_t datalen= ntohl(header->request.bodylen) - keylen; - char *key= (void*)(header + 1); - char *data= key + keylen; - uint64_t cas= ntohll(header->request.cas); - uint64_t result_cas; - - rval= client->root->callback->interface.v1.append(cookie, key, keylen, - data, datalen, cas, - &result_cas); - if (rval == PROTOCOL_BINARY_RESPONSE_SUCCESS && - header->request.opcode == PROTOCOL_BINARY_CMD_APPEND) - { - /* Send a positive request */ - protocol_binary_response_no_extras response= { - .message= { - .header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= PROTOCOL_BINARY_CMD_APPEND, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= header->request.opaque, - .cas= ntohll(result_cas), - }, - } - }; - rval= response_handler(cookie, header, (void*)&response); - } - } - else - { - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - } - - return rval; -} - -/** - * Callback for PREPEND and PREPENDQ - * @param cookie the calling client - * @param header the command - * @param response_handler not used - * @return the result of the operation - */ -static protocol_binary_response_status -prepend_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - (void)response_handler; - protocol_binary_response_status rval; - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.prepend != NULL) - { - uint16_t keylen= ntohs(header->request.keylen); - uint32_t datalen= ntohl(header->request.bodylen) - keylen; - char *key= (char*)(header + 1); - char *data= key + keylen; - uint64_t cas= ntohll(header->request.cas); - uint64_t result_cas; - rval= client->root->callback->interface.v1.prepend(cookie, key, keylen, - data, datalen, cas, - &result_cas); - if (rval == PROTOCOL_BINARY_RESPONSE_SUCCESS && - header->request.opcode == PROTOCOL_BINARY_CMD_PREPEND) - { - /* Send a positive request */ - protocol_binary_response_no_extras response= { - .message= { - .header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= PROTOCOL_BINARY_CMD_PREPEND, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= header->request.opaque, - .cas= ntohll(result_cas), - }, - } - }; - rval= response_handler(cookie, header, (void*)&response); - } - } - else - { - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - } - - return rval; -} - -/** - * Callback for QUIT and QUITQ. Notify the client and shut down the connection - * @param cookie the calling client - * @param header the command - * @param response_handler not used - * @return the result of the operation - */ -static protocol_binary_response_status -quit_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.quit != NULL) - { - client->root->callback->interface.v1.quit(cookie); - } - - protocol_binary_response_no_extras response= { - .message = { - .header.response = { - .magic= PROTOCOL_BINARY_RES, - .opcode= PROTOCOL_BINARY_CMD_QUIT, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= header->request.opaque - } - } - }; - - if (header->request.opcode == PROTOCOL_BINARY_CMD_QUIT) - { - response_handler(cookie, header, (void*)&response); - } - - /* I need a better way to signal to close the connection */ - return PROTOCOL_BINARY_RESPONSE_EIO; -} - -/** - * Callback for REPLACE and REPLACEQ - * @param cookie the calling client - * @param header the command - * @param response_handler not used - * @return the result of the operation - */ -static protocol_binary_response_status -replace_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - (void)response_handler; - protocol_binary_response_status rval; - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.replace != NULL) - { - uint16_t keylen= ntohs(header->request.keylen); - uint32_t datalen= ntohl(header->request.bodylen) - keylen - 8; - protocol_binary_request_replace *request= (void*)header; - uint32_t flags= ntohl(request->message.body.flags); - uint32_t timeout= ntohl(request->message.body.expiration); - char *key= ((char*)header) + sizeof(*header) + 8; - char *data= key + keylen; - uint64_t cas= ntohll(header->request.cas); - uint64_t result_cas; - - rval= client->root->callback->interface.v1.replace(cookie, key, keylen, - data, datalen, flags, - timeout, cas, - &result_cas); - if (rval == PROTOCOL_BINARY_RESPONSE_SUCCESS && - header->request.opcode == PROTOCOL_BINARY_CMD_REPLACE) - { - /* Send a positive request */ - protocol_binary_response_no_extras response= { - .message= { - .header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= PROTOCOL_BINARY_CMD_REPLACE, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= header->request.opaque, - .cas= ntohll(result_cas), - }, - } - }; - rval= response_handler(cookie, header, (void*)&response); - } - } - else - { - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - } - - return rval; -} - -/** - * Callback for SET and SETQ - * @param cookie the calling client - * @param header the command - * @param response_handler not used - * @return the result of the operation - */ -static protocol_binary_response_status -set_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - (void)response_handler; - protocol_binary_response_status rval; - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.set != NULL) - { - uint16_t keylen= ntohs(header->request.keylen); - uint32_t datalen= ntohl(header->request.bodylen) - keylen - 8; - protocol_binary_request_replace *request= (void*)header; - uint32_t flags= ntohl(request->message.body.flags); - uint32_t timeout= ntohl(request->message.body.expiration); - char *key= ((char*)header) + sizeof(*header) + 8; - char *data= key + keylen; - uint64_t cas= ntohll(header->request.cas); - uint64_t result_cas; - - - rval= client->root->callback->interface.v1.set(cookie, key, keylen, - data, datalen, flags, - timeout, cas, &result_cas); - if (rval == PROTOCOL_BINARY_RESPONSE_SUCCESS && - header->request.opcode == PROTOCOL_BINARY_CMD_SET) - { - /* Send a positive request */ - protocol_binary_response_no_extras response= { - .message= { - .header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= PROTOCOL_BINARY_CMD_SET, - .status= htons(PROTOCOL_BINARY_RESPONSE_SUCCESS), - .opaque= header->request.opaque, - .cas= ntohll(result_cas), - }, - } - }; - rval= response_handler(cookie, header, (void*)&response); - } - } - else - { - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - } - - return rval; -} - -/** - * Callback for STAT - * @param cookie the calling client - * @param header the command - * @param response_handler not used - * @return the result of the operation - */ -static protocol_binary_response_status -stat_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - (void)response_handler; - protocol_binary_response_status rval; - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.stat != NULL) - { - uint16_t keylen= ntohs(header->request.keylen); - - rval= client->root->callback->interface.v1.stat(cookie, - (void*)(header + 1), - keylen, - stat_response_handler); - } - else - { - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - } - - return rval; -} - -/** - * Callback for VERSION - * @param cookie the calling client - * @param header the command - * @param response_handler not used - * @return the result of the operation - */ -static protocol_binary_response_status -version_command_handler(const void *cookie, - protocol_binary_request_header *header, - memcached_binary_protocol_raw_response_handler response_handler) -{ - (void)response_handler; - (void)header; - protocol_binary_response_status rval; - - struct memcached_binary_protocol_client_st *client= (void*)cookie; - if (client->root->callback->interface.v1.version != NULL) - { - rval= client->root->callback->interface.v1.version(cookie, - version_response_handler); + /* + * The ASCII protocol can only be used if the implementors provide + * an implementation for the version 1 of the interface.. + * + * @todo I should allow the implementors to provide an implementation + * for version 0 and 1 at the same time and set the preferred + * interface to use... + */ + client->work= memcached_ascii_protocol_process_data; } else { - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - } - - return rval; -} - -/** - * The map to remap between the com codes and the v1 logical setting - */ -static memcached_binary_protocol_command_handler comcode_v0_v1_remap[256]= { - [PROTOCOL_BINARY_CMD_ADDQ]= add_command_handler, - [PROTOCOL_BINARY_CMD_ADD]= add_command_handler, - [PROTOCOL_BINARY_CMD_APPENDQ]= append_command_handler, - [PROTOCOL_BINARY_CMD_APPEND]= append_command_handler, - [PROTOCOL_BINARY_CMD_DECREMENTQ]= decrement_command_handler, - [PROTOCOL_BINARY_CMD_DECREMENT]= decrement_command_handler, - [PROTOCOL_BINARY_CMD_DELETEQ]= delete_command_handler, - [PROTOCOL_BINARY_CMD_DELETE]= delete_command_handler, - [PROTOCOL_BINARY_CMD_FLUSHQ]= flush_command_handler, - [PROTOCOL_BINARY_CMD_FLUSH]= flush_command_handler, - [PROTOCOL_BINARY_CMD_GETKQ]= get_command_handler, - [PROTOCOL_BINARY_CMD_GETK]= get_command_handler, - [PROTOCOL_BINARY_CMD_GETQ]= get_command_handler, - [PROTOCOL_BINARY_CMD_GET]= get_command_handler, - [PROTOCOL_BINARY_CMD_INCREMENTQ]= increment_command_handler, - [PROTOCOL_BINARY_CMD_INCREMENT]= increment_command_handler, - [PROTOCOL_BINARY_CMD_NOOP]= noop_command_handler, - [PROTOCOL_BINARY_CMD_PREPENDQ]= prepend_command_handler, - [PROTOCOL_BINARY_CMD_PREPEND]= prepend_command_handler, - [PROTOCOL_BINARY_CMD_QUITQ]= quit_command_handler, - [PROTOCOL_BINARY_CMD_QUIT]= quit_command_handler, - [PROTOCOL_BINARY_CMD_REPLACEQ]= replace_command_handler, - [PROTOCOL_BINARY_CMD_REPLACE]= replace_command_handler, - [PROTOCOL_BINARY_CMD_SETQ]= set_command_handler, - [PROTOCOL_BINARY_CMD_SET]= set_command_handler, - [PROTOCOL_BINARY_CMD_STAT]= stat_command_handler, - [PROTOCOL_BINARY_CMD_VERSION]= version_command_handler, -}; - -/** - * Try to execute a command. Fire the pre/post functions and the specialized - * handler function if it's set. If not, the unknown probe should be fired - * if it's present. - * @param client the client connection to operate on - * @param header the command to execute - * @return true if success or false if a fatal error occured so that the - * connection should be shut down. - */ -static bool execute_command(struct memcached_binary_protocol_client_st *client, protocol_binary_request_header *header) -{ - if (client->root->pedantic && - memcached_binary_protocol_pedantic_check_request(header)) - { - /* @todo return invalid command packet */ - } - - /* we got all data available, execute the callback! */ - if (client->root->callback->pre_execute != NULL) - { - client->root->callback->pre_execute(client, header); - } - - protocol_binary_response_status rval; - rval= PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND; - uint8_t cc= header->request.opcode; - - switch (client->root->callback->interface_version) - { - case 0: - if (client->root->callback->interface.v0.comcode[cc] != NULL) { - rval= client->root->callback->interface.v0.comcode[cc](client, header, raw_response_handler); - } - break; - case 1: - if (comcode_v0_v1_remap[cc] != NULL) { - rval= comcode_v0_v1_remap[cc](client, header, raw_response_handler); - } - break; - default: - /* Unknown interface. - * It should be impossible to get here so I'll just call abort - * to avoid getting a compiler warning :-) + /* Let's just output a warning the way it is supposed to look like + * in the ASCII protocol... */ - abort(); - } - - - if (rval == PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND && - client->root->callback->unknown != NULL) - { - rval= client->root->callback->unknown(client, header, raw_response_handler); + const char *err= "CLIENT_ERROR: Unsupported protocol\r\n"; + client->root->spool(client, err, strlen(err)); + client->root->drain(client); + return MEMCACHED_PROTOCOL_ERROR_EVENT; /* Unsupported protocol */ } - if (rval != PROTOCOL_BINARY_RESPONSE_SUCCESS && - rval != PROTOCOL_BINARY_RESPONSE_EIO) - { - protocol_binary_response_no_extras response= { - .message= { - .header.response= { - .magic= PROTOCOL_BINARY_RES, - .opcode= cc, - .status= htons(rval), - .opaque= header->request.opaque, - }, - } - }; - rval= raw_response_handler(client, header, (void*)&response); - } - - if (client->root->callback->post_execute != NULL) - { - client->root->callback->post_execute(client, header); - } - - return rval != PROTOCOL_BINARY_RESPONSE_EIO; + return client->work(client, length, endptr); } /* @@ -1167,13 +231,15 @@ static bool execute_command(struct memcached_binary_protocol_client_st *client, ** * See protocol_handler.h for function description ** ********************************************************************** */ -struct memcached_binary_protocol_st *memcached_binary_protocol_create_instance(void) +struct memcached_protocol_st *memcached_protocol_create_instance(void) { - struct memcached_binary_protocol_st *ret= calloc(1, sizeof(*ret)); + struct memcached_protocol_st *ret= calloc(1, sizeof(*ret)); if (ret != NULL) { ret->recv= default_recv; ret->send= default_send; + ret->drain= drain_output; + ret->spool= spool_output; ret->input_buffer_size= 1 * 1024 * 1024; ret->input_buffer= malloc(ret->input_buffer_size); if (ret->input_buffer == NULL) @@ -1183,10 +249,11 @@ struct memcached_binary_protocol_st *memcached_binary_protocol_create_instance(v return NULL; } - ret->buffer_cache = cache_create("protocol_handler", + ret->buffer_cache= cache_create("protocol_handler", CHUNK_BUFFERSIZE + sizeof(struct chunk_st), 0, NULL, NULL); - if (ret->buffer_cache == NULL) { + if (ret->buffer_cache == NULL) + { free(ret->input_buffer); free(ret); } @@ -1195,57 +262,32 @@ struct memcached_binary_protocol_st *memcached_binary_protocol_create_instance(v return ret; } -void memcached_binary_protocol_destroy_instance(struct memcached_binary_protocol_st *instance) +void memcached_protocol_destroy_instance(struct memcached_protocol_st *instance) { cache_destroy(instance->buffer_cache); free(instance->input_buffer); free(instance); } -struct memcached_binary_protocol_callback_st *memcached_binary_protocol_get_callbacks(struct memcached_binary_protocol_st *instance) -{ - return instance->callback; -} - -void memcached_binary_protocol_set_callbacks(struct memcached_binary_protocol_st *instance, struct memcached_binary_protocol_callback_st *callback) +struct memcached_protocol_client_st *memcached_protocol_create_client(struct memcached_protocol_st *instance, memcached_socket_t sock) { - instance->callback= callback; -} - -memcached_binary_protocol_raw_response_handler memcached_binary_protocol_get_raw_response_handler(const void *cookie) -{ - (void)cookie; - return raw_response_handler; -} - -void memcached_binary_protocol_set_pedantic(struct memcached_binary_protocol_st *instance, bool enable) -{ - instance->pedantic= enable; -} - -bool memcached_binary_protocol_get_pedantic(struct memcached_binary_protocol_st *instance) -{ - return instance->pedantic; -} - -struct memcached_binary_protocol_client_st *memcached_binary_protocol_create_client(struct memcached_binary_protocol_st *instance, int sock) -{ - struct memcached_binary_protocol_client_st *ret= calloc(1, sizeof(*ret)); + struct memcached_protocol_client_st *ret= calloc(1, sizeof(*ret)); if (ret != NULL) { ret->root= instance; ret->sock= sock; + ret->work= determine_protocol; } return ret; } -void memcached_binary_protocol_client_destroy(struct memcached_binary_protocol_client_st *client) +void memcached_protocol_client_destroy(struct memcached_protocol_client_st *client) { free(client); } -enum MEMCACHED_BINARY_PROTOCOL_EVENT memcached_binary_protocol_client_work(struct memcached_binary_protocol_client_st *client) +memcached_protocol_event_t memcached_protocol_client_work(struct memcached_protocol_client_st *client) { /* Try to send data and read from the socket */ bool more_data= true; @@ -1270,43 +312,11 @@ enum MEMCACHED_BINARY_PROTOCOL_EVENT memcached_binary_protocol_client_work(struc client->input_buffer_offset= 0; } - /* try to parse all of the received packets */ - protocol_binary_request_header *header; - header= (void*)client->root->input_buffer; - - if (header->request.magic != (uint8_t)PROTOCOL_BINARY_REQ) - { - client->error= EINVAL; - return ERROR_EVENT; - } - - while (len >= (ssize_t)sizeof(*header) && - (len >= (ssize_t)(sizeof(*header) + ntohl(header->request.bodylen)))) + void *endptr; + memcached_protocol_event_t events= client->work(client, &len, &endptr); + if (events == MEMCACHED_PROTOCOL_ERROR_EVENT) { - - /* I have the complete package */ - client->current_command = header; - if (!execute_command(client, header)) - { - return ERROR_EVENT; - } - - ssize_t total= (ssize_t)(sizeof(*header) + ntohl(header->request.bodylen)); - len -= total; - if (len > 0) - { - intptr_t ptr= (intptr_t)header; - ptr += total; - if ((ptr % 8) == 0) - { - header= (void*)ptr; - } - else - { - memmove(client->root->input_buffer, (void*)ptr, (size_t)len); - header= (void*)client->root->input_buffer; - } - } + return MEMCACHED_PROTOCOL_ERROR_EVENT; } if (len > 0) @@ -1317,9 +327,9 @@ enum MEMCACHED_BINARY_PROTOCOL_EVENT memcached_binary_protocol_client_work(struc if (client->input_buffer == NULL) { client->error= ENOMEM; - return ERROR_EVENT; + return MEMCACHED_PROTOCOL_ERROR_EVENT; } - memcpy(client->input_buffer, header, (size_t)len); + memcpy(client->input_buffer, endptr, (size_t)len); client->input_buffer_offset= (size_t)len; more_data= false; } @@ -1328,24 +338,28 @@ enum MEMCACHED_BINARY_PROTOCOL_EVENT memcached_binary_protocol_client_work(struc { /* Connection closed */ drain_output(client); - return ERROR_EVENT; + return MEMCACHED_PROTOCOL_ERROR_EVENT; } else { - if (errno != EWOULDBLOCK) + if (get_socket_errno() != EWOULDBLOCK) { - client->error= errno; + client->error= get_socket_errno(); /* mark this client as terminated! */ - return ERROR_EVENT; + return MEMCACHED_PROTOCOL_ERROR_EVENT; } - more_data = false; + more_data= false; } } while (more_data); if (!drain_output(client)) { - return ERROR_EVENT; + return MEMCACHED_PROTOCOL_ERROR_EVENT; } - return (client->output) ? READ_WRITE_EVENT : READ_EVENT; + memcached_protocol_event_t ret= MEMCACHED_PROTOCOL_READ_EVENT; + if (client->output) + ret|= MEMCACHED_PROTOCOL_READ_EVENT; + + return ret; }