From 1c5347ac90a725929777e4a89463bf6504e8b187 Mon Sep 17 00:00:00 2001 From: Unbit Date: Mon, 11 Feb 2013 18:57:58 +0100 Subject: [PATCH] added router_memcached plugin --- plugins/router_memcached/router_memcached.c | 232 ++++++++++++++++++++ plugins/router_memcached/uwsgiplugin.py | 6 + 2 files changed, 238 insertions(+) create mode 100644 plugins/router_memcached/router_memcached.c create mode 100644 plugins/router_memcached/uwsgiplugin.py diff --git a/plugins/router_memcached/router_memcached.c b/plugins/router_memcached/router_memcached.c new file mode 100644 index 00000000..19bf2bf6 --- /dev/null +++ b/plugins/router_memcached/router_memcached.c @@ -0,0 +1,232 @@ +#include + +#define MEMCACHED_BUFSIZE 8192 + +extern struct uwsgi_server uwsgi; + +/* + + memcached internal router + + route = /^foobar1(.*)/ memcached:addr=127.0.0.1:11211,key=foo$1poo,type=body + +*/ + +struct uwsgi_router_memcached_conf { + + char *addr; + + char *key; + size_t key_len; + + char *content_type; + size_t content_type_len; + + // 0 -> full, 1 -> body + char *type; + int type_num; + +}; + +static size_t memcached_firstline_parse(char *buf, size_t len) { + // check for "VALUE x 0 0" + if (len < 11) return 0; + char *flags = memchr(buf + 6, ' ', len-6); + if (!flags) return 0; + size_t skip = (flags-buf)+1; + if (skip+1 >= len) return 0; + char *bytes = memchr(buf + skip + 1, ' ', len - (skip+1)); + if (!bytes) return 0; + skip = (bytes-buf)+1; + if (skip+1 > len) return 0; + char *bytes_end = memchr(buf + skip + 1, ' ', len - (skip+1)); + if (bytes_end) { + return uwsgi_str_num(bytes + 1, bytes_end - (bytes+1)); + } + else { + return uwsgi_str_num(bytes + 1, len-skip); + } +} + +static int uwsgi_routing_func_memcached(struct wsgi_request *wsgi_req, struct uwsgi_route *ur){ + // this is the buffer for the memcached response + char buf[MEMCACHED_BUFSIZE]; + size_t i; + char last_char = 0; + + struct uwsgi_router_memcached_conf *urmc = (struct uwsgi_router_memcached_conf *) ur->data2; + + char **subject = (char **) (((char *)(wsgi_req))+ur->subject); + uint16_t *subject_len = (uint16_t *) (((char *)(wsgi_req))+ur->subject_len); + + char *k = uwsgi_regexp_apply_ovec(*subject, *subject_len, urmc->key, urmc->key_len, ur->ovector, ur->ovn); + size_t k_len = strlen(k); + + int fd = uwsgi_connect(urmc->addr, 0, 1); + if (fd < 0) goto end; + + // wait for connection; + int ret = uwsgi.wait_write_hook(fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + if (ret <= 0) { + close(fd); + goto end; + } + + // build the request and send it + char *cmd = uwsgi_concat3n("get ", 4, k, k_len, "\r\n", 2); + if (uwsgi_write_true_nb(fd, cmd, 6+k_len, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT])) { + free(cmd); + close(fd); + goto end; + } + free(cmd); + + // ok, start reading the response... + // first we need to get a full line; + size_t found = 0; + size_t pos = 0; + for(;;) { + ssize_t len = read(fd, buf + pos, MEMCACHED_BUFSIZE - pos); + if (len > 0) { + pos += len; + goto read; + } + if (len < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS) goto wait; + } + close(fd); + goto end; +wait: + ret = uwsgi.wait_read_hook(fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + // when we have a chunk try to read the first line + if (ret > 0) { + len = read(fd, buf + pos, MEMCACHED_BUFSIZE - pos); + if (len > 0) { + pos += len; + goto read; + } + } + close(fd); + goto end; +read: + for(i=0;itype_num == 1) { + if (uwsgi_response_prepare_headers(wsgi_req, "200 OK", 6)) { close(fd); goto end; } + if (uwsgi_response_add_content_type(wsgi_req, urmc->content_type, urmc->content_type_len)) { close(fd); goto end; } + if (uwsgi_response_add_content_length(wsgi_req, response_size)) { close(fd); goto end; } + } + size_t remains = pos-(found+2); + if (remains >= response_size) { + uwsgi_response_write_body_do(wsgi_req, buf+found+2, response_size); + close(fd); + goto end; + } + + // send what we have + if (uwsgi_response_write_body_do(wsgi_req, buf+found+2, remains)) { + close(fd); + goto end; + } + + // and now start reading til the output is consumed + response_size -= remains; + while(response_size > 0) { + ssize_t len = read(fd, buf, UMIN(MEMCACHED_BUFSIZE, response_size)); + if (len > 0) goto write; + if (len < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS) goto wait2; + } + goto error; +wait2: + ret = uwsgi.wait_read_hook(fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + if (ret > 0) { + len = read(fd, buf, UMIN(MEMCACHED_BUFSIZE, response_size)); + if (len > 0) goto write; + } + goto error; +write: + if (uwsgi_response_write_body_do(wsgi_req, buf, len)) { + goto error; + } + response_size -= len; + } + + close(fd); + return UWSGI_ROUTE_BREAK; + +error: + close(fd); + +end: + if (ur->custom) + return UWSGI_ROUTE_NEXT; + + return UWSGI_ROUTE_BREAK; +} + +static int uwsgi_router_memcached(struct uwsgi_route *ur, char *args) { + ur->func = uwsgi_routing_func_memcached; + ur->data = args; + ur->data_len = strlen(args); + struct uwsgi_router_memcached_conf *urmc = uwsgi_calloc(sizeof(struct uwsgi_router_memcached_conf)); + if (uwsgi_kvlist_parse(ur->data, ur->data_len, ',', '=', + "addr", &urmc->addr, + "key", &urmc->key, + "content_type", &urmc->content_type, + "type", &urmc->type, NULL)) { + uwsgi_log("invalid route syntax: %s\n", args); + exit(1); + } + + if (!urmc->key || !urmc->addr) { + uwsgi_log("invalid route syntax: you need to specify a memcached address and key pattern\n"); + exit(1); + } + + urmc->key_len = strlen(urmc->key); + + if (!urmc->type) urmc->type = "full"; + if (!urmc->content_type) urmc->content_type = "text/html"; + + urmc->content_type_len = strlen(urmc->content_type); + + if (!strcmp(urmc->type, "body")) { + urmc->type_num = 1; + } + + ur->data2 = urmc; + return 0; +} + +static int uwsgi_router_memcached_continue(struct uwsgi_route *ur, char *args) { + uwsgi_router_memcached(ur, args); + ur->custom = 1; + return 0; +} + +static void router_memcached_register() { + uwsgi_register_router("memcached", uwsgi_router_memcached); + uwsgi_register_router("memcached-continue", uwsgi_router_memcached_continue); +} + +struct uwsgi_plugin router_memcached_plugin = { + .name = "router_memcached", + .on_load = router_memcached_register, +}; diff --git a/plugins/router_memcached/uwsgiplugin.py b/plugins/router_memcached/uwsgiplugin.py new file mode 100644 index 00000000..b72e6ad8 --- /dev/null +++ b/plugins/router_memcached/uwsgiplugin.py @@ -0,0 +1,6 @@ +NAME='router_memcached' + +CFLAGS = [] +LDFLAGS = [] +LIBS = [] +GCC_LIST = ['router_memcached']