From e8f39685cc83f13dc44fdebd9506f1f5f378ee1a Mon Sep 17 00:00:00 2001 From: "roberto@quantal64" Date: Sat, 1 Sep 2012 11:17:20 +0200 Subject: [PATCH] added the router_http plugin --- buildconf/base.ini | 2 +- core/buffer.c | 68 +++++ core/logging.c | 252 +++++++++++++++++ core/protocol.c | 36 +-- core/utils.c | 418 ++++++++++------------------ core/uwsgi.c | 2 +- plugins/router_http/router_http.c | 93 +++++++ plugins/router_http/uwsgiplugin.py | 6 + plugins/router_uwsgi/router_uwsgi.c | 30 +- proto/http.c | 85 ++++++ proto/zeromq.c | 16 +- uwsgi.h | 25 +- uwsgiconfig.py | 2 +- 13 files changed, 700 insertions(+), 335 deletions(-) create mode 100644 core/buffer.c create mode 100644 plugins/router_http/router_http.c create mode 100644 plugins/router_http/uwsgiplugin.py diff --git a/buildconf/base.ini b/buildconf/base.ini index 1e8235c6..95cc4738 100644 --- a/buildconf/base.ini +++ b/buildconf/base.ini @@ -29,7 +29,7 @@ plugins = bin_name = uwsgi append_version = plugin_dir = . -embedded_plugins = %(main_plugin)s, ping, cache, nagios, rrdtool, carbon, rpc, corerouter, fastrouter, http, ugreen, signal, syslog, rsyslog, logsocket, router_uwsgi, router_redirect, router_basicauth, zergpool, redislog, rawrouter +embedded_plugins = %(main_plugin)s, ping, cache, nagios, rrdtool, carbon, rpc, corerouter, fastrouter, http, ugreen, signal, syslog, rsyslog, logsocket, router_uwsgi, router_redirect, router_basicauth, zergpool, redislog, router_rewrite, router_http as_shared_library = false locking = auto diff --git a/core/buffer.c b/core/buffer.c new file mode 100644 index 00000000..4a629be0 --- /dev/null +++ b/core/buffer.c @@ -0,0 +1,68 @@ +#include "uwsgi.h" + +extern struct uwsgi_server uwsgi; + +struct uwsgi_buffer *uwsgi_buffer_new(size_t len) { + struct uwsgi_buffer *ub = uwsgi_calloc(sizeof(struct uwsgi_buffer)); + + if (len) { + ub->buf = uwsgi_malloc(len); + ub->len = len; + } + return ub; + +} + +int uwsgi_buffer_append(struct uwsgi_buffer *ub, char *buf, size_t len) { + + size_t remains = ub->len - ub->pos; + + if (len > remains) { + char *new_buf = realloc(ub->buf, ub->len + UMAX(len, (size_t) uwsgi.page_size)); + if (!new_buf) { + uwsgi_error("realloc()"); + return -1; + } + } + + memcpy(ub->buf + ub->pos, buf, len); + ub->pos += len; + return 0; +} + +void uwsgi_buffer_destroy(struct uwsgi_buffer *ub) { + if (ub->buf) free(ub->buf); + free(ub); +} + +int uwsgi_buffer_send(struct uwsgi_buffer *ub, int fd) { + size_t remains = ub->pos; + char *ptr = ub->buf; + + while(remains > 0) { + int ret = uwsgi_waitfd_write(fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + if (ret > 0) { + ssize_t len = write(fd, ptr, remains); + if (len > 0) { + ptr += len; + remains -= len; + } + else if (len == 0) { + return -1; + } + else { + uwsgi_error("write()"); + return -1; + } + } + else if (ret == 0) { + uwsgi_log("timeout while sending buffer !!!\n"); + return -1; + } + else { + return -1; + } + } + + return 0; +} diff --git a/core/logging.c b/core/logging.c index 5e10cb75..75ccbdad 100644 --- a/core/logging.c +++ b/core/logging.c @@ -19,6 +19,258 @@ extern struct uwsgi_server uwsgi; +//use this instead of fprintf to avoid buffering mess with udp logging +void uwsgi_log(const char *fmt, ...) { + va_list ap; + char logpkt[4096]; + int rlen = 0; + int ret; + + struct timeval tv; + char sftime[64]; + char ctime_storage[26]; + time_t now; + + if (uwsgi.logdate) { + if (uwsgi.log_strftime) { + now = uwsgi_now(); + rlen = strftime(sftime, 64, uwsgi.log_strftime, localtime(&now)); + memcpy(logpkt, sftime, rlen); + memcpy(logpkt + rlen, " - ", 3); + rlen += 3; + } + else { + gettimeofday(&tv, NULL); +#ifdef __sun__ + ctime_r((const time_t *) &tv.tv_sec, ctime_storage, 26); +#else + ctime_r((const time_t *) &tv.tv_sec, ctime_storage); +#endif + memcpy(logpkt, ctime_storage, 24); + memcpy(logpkt + 24, " - ", 3); + + rlen = 24 + 3; + } + } + + va_start(ap, fmt); + ret = vsnprintf(logpkt + rlen, 4096 - rlen, fmt, ap); + va_end(ap); + + if (ret >= 4096) { + char *tmp_buf = uwsgi_malloc(rlen + ret + 1); + memcpy(tmp_buf, logpkt, rlen); + va_start(ap, fmt); + ret = vsnprintf(tmp_buf + rlen, ret + 1, fmt, ap); + va_end(ap); + rlen = write(2, tmp_buf, rlen + ret); + free(tmp_buf); + return; + } + + rlen += ret; + // do not check for errors + rlen = write(2, logpkt, rlen); +} + +void uwsgi_log_verbose(const char *fmt, ...) { + + va_list ap; + char logpkt[4096]; + int rlen = 0; + + struct timeval tv; + char sftime[64]; + time_t now; + char ctime_storage[26]; + + if (uwsgi.log_strftime) { + now = uwsgi_now(); + rlen = strftime(sftime, 64, uwsgi.log_strftime, localtime(&now)); + memcpy(logpkt, sftime, rlen); + memcpy(logpkt + rlen, " - ", 3); + rlen += 3; + } + else { + gettimeofday(&tv, NULL); +#ifdef __sun__ + ctime_r((const time_t *) &tv.tv_sec, ctime_storage, 26); +#else + ctime_r((const time_t *) &tv.tv_sec, ctime_storage); +#endif + memcpy(logpkt, ctime_storage, 24); + memcpy(logpkt + 24, " - ", 3); + + rlen = 24 + 3; + } + + + + va_start(ap, fmt); + rlen += vsnprintf(logpkt + rlen, 4096 - rlen, fmt, ap); + va_end(ap); + + // do not check for errors + rlen = write(2, logpkt, rlen); +} + + + + +// create the logpipe +void create_logpipe(void) { + +#if defined(SOCK_SEQPACKET) && defined(__linux__) + if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, uwsgi.shared->worker_log_pipe)) { +#else + if (socketpair(AF_UNIX, SOCK_DGRAM, 0, uwsgi.shared->worker_log_pipe)) { +#endif + uwsgi_error("socketpair()\n"); + exit(1); + } + + uwsgi_socket_nb(uwsgi.shared->worker_log_pipe[0]); + uwsgi_socket_nb(uwsgi.shared->worker_log_pipe[1]); + + if (uwsgi.shared->worker_log_pipe[1] != 1) { + if (dup2(uwsgi.shared->worker_log_pipe[1], 1) < 0) { + uwsgi_error("dup2()"); + exit(1); + } + } + + if (dup2(1, 2) < 0) { + uwsgi_error("dup2()"); + exit(1); + } + +} + +#ifdef UWSGI_ZEROMQ +// the zeromq logger +ssize_t uwsgi_zeromq_logger(struct uwsgi_logger *ul, char *message, size_t len) { + + if (!ul->configured) { + + if (!ul->arg) { + uwsgi_log_safe("invalid zeromq syntax\n"); + exit(1); + } + + void *ctx = uwsgi_zeromq_init(); + + ul->data = zmq_socket(ctx, ZMQ_PUSH); + if (ul->data == NULL) { + uwsgi_error_safe("zmq_socket()"); + exit(1); + } + + if (zmq_connect(ul->data, ul->arg) < 0) { + uwsgi_error_safe("zmq_connect()"); + exit(1); + } + + ul->configured = 1; + } + + zmq_msg_t msg; + if (zmq_msg_init_size(&msg, len) == 0) { + memcpy(zmq_msg_data(&msg), message, len); +#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0) + zmq_sendmsg(ul->data, &msg, 0); +#else + zmq_send(ul->data, &msg, 0); +#endif + zmq_msg_close(&msg); + } + + return 0; +} +#endif + + +// log to the specified file or udp address +void logto(char *logfile) { + + int fd; + +#ifdef UWSGI_UDP + char *udp_port; + struct sockaddr_in udp_addr; + + udp_port = strchr(logfile, ':'); + if (udp_port) { + udp_port[0] = 0; + if (!udp_port[1] || !logfile[0]) { + uwsgi_log("invalid udp address\n"); + exit(1); + } + + fd = socket(AF_INET, SOCK_DGRAM, 0); + if (fd < 0) { + uwsgi_error("socket()"); + exit(1); + } + + memset(&udp_addr, 0, sizeof(struct sockaddr_in)); + + udp_addr.sin_family = AF_INET; + udp_addr.sin_port = htons(atoi(udp_port + 1)); + char *resolved = uwsgi_resolve_ip(logfile); + if (resolved) { + udp_addr.sin_addr.s_addr = inet_addr(resolved); + } + else { + udp_addr.sin_addr.s_addr = inet_addr(logfile); + } + + if (connect(fd, (const struct sockaddr *) &udp_addr, sizeof(struct sockaddr_in)) < 0) { + uwsgi_error("connect()"); + exit(1); + } + } + else { +#endif + if (uwsgi.log_truncate) { + fd = open(logfile, O_RDWR | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR | S_IRGRP); + } + else { + fd = open(logfile, O_RDWR | O_CREAT | O_APPEND, S_IRUSR | S_IWUSR | S_IRGRP); + } + if (fd < 0) { + uwsgi_error_open(logfile); + exit(1); + } + uwsgi.logfile = logfile; + + if (uwsgi.chmod_logfile_value) { + if (chmod(uwsgi.logfile, uwsgi.chmod_logfile_value)) { + uwsgi_error("chmod()"); + } + } +#ifdef UWSGI_UDP + } +#endif + + + /* stdout */ + if (fd != 1) { + if (dup2(fd, 1) < 0) { + uwsgi_error("dup2()"); + exit(1); + } + close(fd); + } + + /* stderr */ + if (dup2(1, 2) < 0) { + uwsgi_error("dup2()"); + exit(1); + } +} + + + void uwsgi_setup_log() { if (uwsgi.daemonize) { diff --git a/core/protocol.c b/core/protocol.c index d9a5f861..0ff7da01 100644 --- a/core/protocol.c +++ b/core/protocol.c @@ -412,12 +412,9 @@ int uwsgi_enqueue_message(char *host, int port, uint8_t modifier1, uint8_t modif ssize_t uwsgi_send_message(int fd, uint8_t modifier1, uint8_t modifier2, char *message, uint16_t size, int pfd, ssize_t plen, int timeout) { - struct pollfd uwsgi_mpoll; ssize_t cnt; struct uwsgi_header uh; - char buffer[4096]; ssize_t ret = 0; - int pret; struct msghdr msg; struct iovec iov [1]; union { @@ -478,37 +475,8 @@ ssize_t uwsgi_send_message(int fd, uint8_t modifier1, uint8_t modifier2, char *m // transfer data from one socket to another if (pfd >= 0 && plen > 0) { - uwsgi_mpoll.fd = pfd; - uwsgi_mpoll.events = POLLIN; - - while(plen > 0) { - pret = poll(&uwsgi_mpoll, 1, timeout*1000); - if (pret < 0) { - uwsgi_error("poll()"); - return -1; - } - else if (pret == 0) { - uwsgi_log("timeout waiting for socket data\n"); - return -1; - } - else { - cnt = read(pfd, buffer, UMIN(4096, plen)); - if (cnt < 0) { - uwsgi_error("read()"); - return -1; - } - else if (cnt == 0) { - return ret; - } - // send to peer - if (write(fd, buffer, cnt) != cnt) { - uwsgi_error("write()"); - return -1; - } - ret += cnt; - plen -= cnt; - } - } + ret = uwsgi_pipe_sized(pfd, fd, timeout, plen); + if (ret < 0) return -1; } diff --git a/core/utils.c b/core/utils.c index 32756ac1..3d2f4182 100644 --- a/core/utils.c +++ b/core/utils.c @@ -76,6 +76,7 @@ uint64_t uwsgi_swap64(uint64_t x) { #endif +// check if a string is a valid hex number int check_hex(char *str, int len) { int i; for (i = 0; i < len; i++) { @@ -89,6 +90,7 @@ int check_hex(char *str, int len) { } +// increase worker harakiri void inc_harakiri(int sec) { if (uwsgi.master_process) { uwsgi.workers[uwsgi.mywid].harakiri += sec; @@ -98,6 +100,7 @@ void inc_harakiri(int sec) { } } +// set worker harakiri void set_harakiri(int sec) { if (sec == 0) { uwsgi.workers[uwsgi.mywid].harakiri = 0; @@ -110,6 +113,7 @@ void set_harakiri(int sec) { } } +// set user harakiri void set_user_harakiri(int sec) { if (!uwsgi.master_process) { uwsgi_log("!!! unable to set user harakiri without the master process !!!\n"); @@ -123,6 +127,7 @@ void set_user_harakiri(int sec) { } } +// set mule harakiri void set_mule_harakiri(int sec) { if (sec == 0) { uwsgi.mules[uwsgi.muleid - 1].harakiri = 0; @@ -136,6 +141,7 @@ void set_mule_harakiri(int sec) { } #ifdef UWSGI_SPOOLER +// set spooler harakiri void set_spooler_harakiri(int sec) { if (sec == 0) { uwsgi.i_am_a_spooler->harakiri = 0; @@ -150,6 +156,7 @@ void set_spooler_harakiri(int sec) { #endif +// daemonize to the specified logfile void daemonize(char *logfile) { pid_t pid; int fdin; @@ -212,158 +219,7 @@ void daemonize(char *logfile) { logto(logfile); } -void logto(char *logfile) { - - int fd; - -#ifdef UWSGI_UDP - char *udp_port; - struct sockaddr_in udp_addr; - - udp_port = strchr(logfile, ':'); - if (udp_port) { - udp_port[0] = 0; - if (!udp_port[1] || !logfile[0]) { - uwsgi_log("invalid udp address\n"); - exit(1); - } - - fd = socket(AF_INET, SOCK_DGRAM, 0); - if (fd < 0) { - uwsgi_error("socket()"); - exit(1); - } - - memset(&udp_addr, 0, sizeof(struct sockaddr_in)); - - udp_addr.sin_family = AF_INET; - udp_addr.sin_port = htons(atoi(udp_port + 1)); - char *resolved = uwsgi_resolve_ip(logfile); - if (resolved) { - udp_addr.sin_addr.s_addr = inet_addr(resolved); - } - else { - udp_addr.sin_addr.s_addr = inet_addr(logfile); - } - - if (connect(fd, (const struct sockaddr *) &udp_addr, sizeof(struct sockaddr_in)) < 0) { - uwsgi_error("connect()"); - exit(1); - } - } - else { -#endif - if (uwsgi.log_truncate) { - fd = open(logfile, O_RDWR | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR | S_IRGRP); - } - else { - fd = open(logfile, O_RDWR | O_CREAT | O_APPEND, S_IRUSR | S_IWUSR | S_IRGRP); - } - if (fd < 0) { - uwsgi_error_open(logfile); - exit(1); - } - uwsgi.logfile = logfile; - - if (uwsgi.chmod_logfile_value) { - if (chmod(uwsgi.logfile, uwsgi.chmod_logfile_value)) { - uwsgi_error("chmod()"); - } - } -#ifdef UWSGI_UDP - } -#endif - - - /* stdout */ - if (fd != 1) { - if (dup2(fd, 1) < 0) { - uwsgi_error("dup2()"); - exit(1); - } - close(fd); - } - - /* stderr */ - if (dup2(1, 2) < 0) { - uwsgi_error("dup2()"); - exit(1); - } -} - -#ifdef UWSGI_ZEROMQ -ssize_t uwsgi_zeromq_logger(struct uwsgi_logger *ul, char *message, size_t len) { - - if (!ul->configured) { - - if (!ul->arg) { - uwsgi_log_safe("invalid zeromq syntax\n"); - exit(1); - } - - void *ctx = zmq_init(1); - if (ctx == NULL) { - uwsgi_error_safe("zmq_init()"); - exit(1); - } - - ul->data = zmq_socket(ctx, ZMQ_PUSH); - if (ul->data == NULL) { - uwsgi_error_safe("zmq_socket()"); - exit(1); - } - - if (zmq_connect(ul->data, ul->arg) < 0) { - uwsgi_error_safe("zmq_connect()"); - exit(1); - } - - ul->configured = 1; - } - - zmq_msg_t msg; - if (zmq_msg_init_size(&msg, len) == 0) { - memcpy(zmq_msg_data(&msg), message, len); -#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0) - zmq_sendmsg(ul->data, &msg, 0); -#else - zmq_send(ul->data, &msg, 0); -#endif - zmq_msg_close(&msg); - } - - return 0; -} -#endif - -void create_logpipe(void) { - -#if defined(SOCK_SEQPACKET) && defined(__linux__) - if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, uwsgi.shared->worker_log_pipe)) { -#else - if (socketpair(AF_UNIX, SOCK_DGRAM, 0, uwsgi.shared->worker_log_pipe)) { -#endif - uwsgi_error("socketpair()\n"); - exit(1); - } - - uwsgi_socket_nb(uwsgi.shared->worker_log_pipe[0]); - uwsgi_socket_nb(uwsgi.shared->worker_log_pipe[1]); - - if (uwsgi.shared->worker_log_pipe[1] != 1) { - if (dup2(uwsgi.shared->worker_log_pipe[1], 1) < 0) { - uwsgi_error("dup2()"); - exit(1); - } - } - - if (dup2(1, 2) < 0) { - uwsgi_error("dup2()"); - exit(1); - } - -} - +// get current working directory char *uwsgi_get_cwd() { // set this to static to avoid useless reallocations in stats mode @@ -386,6 +242,7 @@ char *uwsgi_get_cwd() { } +// generate internal server error message void internal_server_error(struct wsgi_request *wsgi_req, char *message) { if (uwsgi.wsgi_req->headers_size == 0) { @@ -402,6 +259,7 @@ void internal_server_error(struct wsgi_request *wsgi_req, char *message) { uwsgi.wsgi_req->response_size += wsgi_req->socket->proto_write(wsgi_req, message, strlen(message)); } +// check if a string_list containes an item struct uwsgi_string_list *uwsgi_string_list_has_item(struct uwsgi_string_list *list, char *key, size_t keylen) { struct uwsgi_string_list *usl = list; while (usl) { @@ -491,6 +349,7 @@ void uwsgi_set_cgroup() { } #endif +// drop privileges (as root) void uwsgi_as_root() { @@ -766,6 +625,7 @@ void uwsgi_as_root() { } } +// destroy a request void uwsgi_destroy_request(struct wsgi_request *wsgi_req) { wsgi_req->socket->proto_close(wsgi_req); @@ -783,6 +643,7 @@ void uwsgi_destroy_request(struct wsgi_request *wsgi_req) { } +// finalize/close/free a request void uwsgi_close_request(struct wsgi_request *wsgi_req) { int waitpid_status; @@ -979,6 +840,7 @@ void uwsgi_linux_ksm_map(void) { #endif #endif +// setup for a new request void wsgi_req_setup(struct wsgi_request *wsgi_req, int async_id, struct uwsgi_socket *uwsgi_sock) { wsgi_req->poll.events = POLLIN; @@ -1044,6 +906,7 @@ int wsgi_req_async_recv(struct wsgi_request *wsgi_req) { } #endif +// receive a new request int wsgi_req_recv(struct wsgi_request *wsgi_req) { uwsgi.workers[uwsgi.mywid].cores[wsgi_req->async_id].in_request = 1; @@ -1075,6 +938,7 @@ int wsgi_req_recv(struct wsgi_request *wsgi_req) { } +// accept a new request int wsgi_req_simple_accept(struct wsgi_request *wsgi_req, int fd) { wsgi_req->poll.fd = wsgi_req->socket->proto_accept(wsgi_req, fd); @@ -1091,6 +955,7 @@ int wsgi_req_simple_accept(struct wsgi_request *wsgi_req, int fd) { return 0; } +// send heartbeat to the emperor void uwsgi_heartbeat() { if (!uwsgi.has_emperor) return; @@ -1106,6 +971,7 @@ void uwsgi_heartbeat() { } +// accept a request int wsgi_req_accept(int queue, struct wsgi_request *wsgi_req) { int ret; @@ -1217,6 +1083,7 @@ int wsgi_req_accept(int queue, struct wsgi_request *wsgi_req) { return -1; } +// fix related options void sanitize_args() { if (uwsgi.async > 1) { @@ -1256,6 +1123,7 @@ void sanitize_args() { } } +// translate a OS env to a uWSGI option void env_to_arg(char *src, char *dst) { int i; int val = 0; @@ -1278,6 +1146,7 @@ void env_to_arg(char *src, char *dst) { dst[strlen(src)] = 0; } +// lower a string char *uwsgi_lower(char *str, size_t size) { size_t i; for (i = 0; i < size; i++) { @@ -1287,6 +1156,7 @@ char *uwsgi_lower(char *str, size_t size) { return str; } +// parse OS envs void parse_sys_envs(char **envs) { char **uenvs = envs; @@ -1309,101 +1179,7 @@ void parse_sys_envs(char **envs) { } -//use this instead of fprintf to avoid buffering mess with udp logging -void uwsgi_log(const char *fmt, ...) { - va_list ap; - char logpkt[4096]; - int rlen = 0; - int ret; - - struct timeval tv; - char sftime[64]; - char ctime_storage[26]; - time_t now; - - if (uwsgi.logdate) { - if (uwsgi.log_strftime) { - now = uwsgi_now(); - rlen = strftime(sftime, 64, uwsgi.log_strftime, localtime(&now)); - memcpy(logpkt, sftime, rlen); - memcpy(logpkt + rlen, " - ", 3); - rlen += 3; - } - else { - gettimeofday(&tv, NULL); -#ifdef __sun__ - ctime_r((const time_t *) &tv.tv_sec, ctime_storage, 26); -#else - ctime_r((const time_t *) &tv.tv_sec, ctime_storage); -#endif - memcpy(logpkt, ctime_storage, 24); - memcpy(logpkt + 24, " - ", 3); - - rlen = 24 + 3; - } - } - - va_start(ap, fmt); - ret = vsnprintf(logpkt + rlen, 4096 - rlen, fmt, ap); - va_end(ap); - - if (ret >= 4096) { - char *tmp_buf = uwsgi_malloc(rlen + ret + 1); - memcpy(tmp_buf, logpkt, rlen); - va_start(ap, fmt); - ret = vsnprintf(tmp_buf + rlen, ret + 1, fmt, ap); - va_end(ap); - rlen = write(2, tmp_buf, rlen + ret); - free(tmp_buf); - return; - } - - rlen += ret; - // do not check for errors - rlen = write(2, logpkt, rlen); -} - -void uwsgi_log_verbose(const char *fmt, ...) { - - va_list ap; - char logpkt[4096]; - int rlen = 0; - - struct timeval tv; - char sftime[64]; - time_t now; - char ctime_storage[26]; - - if (uwsgi.log_strftime) { - now = uwsgi_now(); - rlen = strftime(sftime, 64, uwsgi.log_strftime, localtime(&now)); - memcpy(logpkt, sftime, rlen); - memcpy(logpkt + rlen, " - ", 3); - rlen += 3; - } - else { - gettimeofday(&tv, NULL); -#ifdef __sun__ - ctime_r((const time_t *) &tv.tv_sec, ctime_storage, 26); -#else - ctime_r((const time_t *) &tv.tv_sec, ctime_storage); -#endif - memcpy(logpkt, ctime_storage, 24); - memcpy(logpkt + 24, " - ", 3); - - rlen = 24 + 3; - } - - - - va_start(ap, fmt); - rlen += vsnprintf(logpkt + rlen, 4096 - rlen, fmt, ap); - va_end(ap); - - // do not check for errors - rlen = write(2, logpkt, rlen); -} - +// check if a string is contained in another one char *uwsgi_str_contains(char *str, int slen, char what) { int i; @@ -1415,6 +1191,7 @@ char *uwsgi_str_contains(char *str, int slen, char what) { return NULL; } +// fast compare 2 sized strings inline int uwsgi_strncmp(char *src, int slen, char *dst, int dlen) { if (slen != dlen) @@ -1424,6 +1201,7 @@ inline int uwsgi_strncmp(char *src, int slen, char *dst, int dlen) { } +// fast sized check of initial part of a string inline int uwsgi_starts_with(char *src, int slen, char *dst, int dlen) { if (slen < dlen) @@ -1432,6 +1210,7 @@ inline int uwsgi_starts_with(char *src, int slen, char *dst, int dlen) { return memcmp(src, dst, dlen); } +// unsized check inline int uwsgi_startswith(char *src, char *what, int wlen) { int i; @@ -1444,6 +1223,7 @@ inline int uwsgi_startswith(char *src, char *what, int wlen) { return 0; } +// concatenate strings char *uwsgi_concatn(int c, ...) { va_list s; @@ -1612,6 +1392,7 @@ char *uwsgi_concat4n(char *one, int s1, char *two, int s2, char *three, int s3, +// concat unsized strings char *uwsgi_concat(int c, ...) { va_list s; @@ -1670,6 +1451,7 @@ char *uwsgi_strncopy(char *s, int len) { } +// get the application id int uwsgi_get_app_id(char *app_name, int app_name_len, int modifier1) { int i; @@ -2336,13 +2118,10 @@ add: } -int uwsgi_waitfd(int fd, int timeout) { +int uwsgi_waitfd_event(int fd, int timeout, int event) { int ret; - struct pollfd upoll[1]; - char oob; - ssize_t rlen; - + struct pollfd upoll; if (!timeout) timeout = uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]; @@ -2351,34 +2130,24 @@ int uwsgi_waitfd(int fd, int timeout) { if (timeout < 0) timeout = -1; - upoll[0].fd = fd; - upoll[0].events = POLLIN | POLLPRI; - upoll[0].revents = 0; - ret = poll(upoll, 1, timeout); + upoll.fd = fd; + upoll.events = event; + upoll.revents = 0; + ret = poll(&upoll, 1, timeout); if (ret < 0) { uwsgi_error("poll()"); } else if (ret > 0) { - if (upoll[0].revents & POLLIN) { + if (upoll.revents & event) { return ret; } - - if (upoll[0].revents & POLLPRI) { - uwsgi_log("DETECTED PRI DATA\n"); - rlen = recv(fd, &oob, 1, MSG_OOB); - uwsgi_log("RECEIVE OOB DATA %d !!!\n", rlen); - if (rlen < 0) { - return -1; - } - return 0; - } + return -1; } return ret; } - inline void *uwsgi_malloc(size_t size) { char *ptr = malloc(size); @@ -4775,6 +4544,119 @@ char *uwsgi_sanitize_cert_filename(char *base, char *key, uint16_t keylen) { #endif +ssize_t uwsgi_pipe(int src, int dst, int timeout) { + char buf[8192]; + size_t written = -1; + ssize_t len; + + for(;;) { + int ret = uwsgi_waitfd(src, timeout); + if (ret > 0) { + len = read(src, buf, 8192); + if (len == 0) { + return written; + } + else if (len < 0) { + uwsgi_error("read()"); + return -1; + } + + size_t remains = len; + while(remains > 0) { + int ret = uwsgi_waitfd_write(dst, timeout); + if (ret > 0) { + len = write(dst, buf, remains); + if (len > 0) { + remains-=len; + written+=len; + } + else if (len == 0) { + return written; + } + else { + uwsgi_error("write()"); + return -1; + } + } + else if (ret == 0) { + goto timeout; + } + else { + return -1; + } + } + } + else if (ret == 0) { + goto timeout; + } + else { + return -1; + } + } + + return written; +timeout: + uwsgi_log("timeout while piping from %d to %d !!!\n", src, dst); + return -1; +} + +ssize_t uwsgi_pipe_sized(int src, int dst, size_t required, int timeout) { + char buf[8192]; + size_t written = -1; + ssize_t len; + + while(written < required) { + int ret = uwsgi_waitfd(src, timeout); + if (ret > 0) { + len = read(src, buf, UMIN(8192, required-written)); + if (len == 0) { + return written; + } + else if (len < 0) { + uwsgi_error("read()"); + return -1; + } + + size_t remains = len; + while(remains > 0) { + int ret = uwsgi_waitfd_write(dst, timeout); + if (ret > 0) { + len = write(dst, buf, remains); + if (len > 0) { + remains-=len; + written+=len; + } + else if (len == 0) { + return written; + } + else { + uwsgi_error("write()"); + return -1; + } + } + else if (ret == 0) { + goto timeout; + } + else { + return -1; + } + } + } + else if (ret == 0) { + goto timeout; + } + else { + return -1; + } + } + + return written; +timeout: + uwsgi_log("timeout while piping from %d to %d !!!\n", src, dst); + return -1; +} + + void uwsgi_set_cpu_affinity() { char buf[4096]; int ret; diff --git a/core/uwsgi.c b/core/uwsgi.c index 63fb234b..16bb19b2 100644 --- a/core/uwsgi.c +++ b/core/uwsgi.c @@ -2594,7 +2594,7 @@ next: #ifdef UWSGI_ZEROMQ // setup zeromq context (if required) one per-worker if (uwsgi.zeromq) { - uwsgi_zeromq_init(); + uwsgi_zeromq_init_sockets(); } #endif diff --git a/plugins/router_http/router_http.c b/plugins/router_http/router_http.c new file mode 100644 index 00000000..ef79e106 --- /dev/null +++ b/plugins/router_http/router_http.c @@ -0,0 +1,93 @@ +#include "../../uwsgi.h" + +#ifdef UWSGI_ROUTING + +extern struct uwsgi_server uwsgi; + +int uwsgi_routing_func_http(struct wsgi_request *wsgi_req, struct uwsgi_route *ur) { + + // mark a route request + wsgi_req->status = -17; + + // get the http address from the route + char *addr = ur->data; + + // connect to the http server + int http_fd = uwsgi_connect(addr, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT], 0); + if (http_fd < 0) { + uwsgi_log("unable to connect to host %s\n", addr); + return UWSGI_ROUTE_NEXT; + } + + // convert the wsgi_request to an http proxy request + struct uwsgi_buffer *ub = uwsgi_to_http(wsgi_req); + if (!ub) { + uwsgi_log("unable to generate http request for %s\n", addr); + close(http_fd); + return UWSGI_ROUTE_NEXT; + } + + // send the request + if (uwsgi_buffer_send(ub, http_fd)) { + uwsgi_log("error routing request to http server %s\n", addr); + close(http_fd); + uwsgi_buffer_destroy(ub); + return UWSGI_ROUTE_NEXT; + } + + ssize_t ret; + + // pipe the body + if (wsgi_req->post_cl > 0) { + int post_fd = wsgi_req->poll.fd; + if (wsgi_req->async_post) { + post_fd = fileno(wsgi_req->async_post); + } + ret = uwsgi_pipe_sized(post_fd, http_fd, wsgi_req->post_cl, 0); + if (ret < 0) { + uwsgi_log("error routing request body (%llu bytes) to http server %s\n", (unsigned long long) wsgi_req->post_cl, addr); + close(http_fd); + uwsgi_buffer_destroy(ub); + return UWSGI_ROUTE_BREAK; + } + } + + // pipe the response + ret = uwsgi_pipe(http_fd, wsgi_req->poll.fd, 0); + if (ret > 0) { + wsgi_req->response_size += ret; + } + else { + uwsgi_log("error routing request to http server %s\n", addr); + } + + close(http_fd); + uwsgi_buffer_destroy(ub); + + return UWSGI_ROUTE_BREAK; + +} + +int uwsgi_router_http(struct uwsgi_route *ur, char *args) { + + ur->func = uwsgi_routing_func_http; + ur->data = (void *) args; + ur->data_len = strlen(args); + return 0; +} + + +void router_http_register(void) { + + uwsgi_register_router("http", uwsgi_router_http); +} + +struct uwsgi_plugin router_http_plugin = { + .name = "router_http", + .on_load = router_http_register, +}; +#else +struct uwsgi_plugin router_http_plugin = { + .name = "router_http", +}; +#endif diff --git a/plugins/router_http/uwsgiplugin.py b/plugins/router_http/uwsgiplugin.py new file mode 100644 index 00000000..da4f9629 --- /dev/null +++ b/plugins/router_http/uwsgiplugin.py @@ -0,0 +1,6 @@ +NAME='router_http' + +CFLAGS = [] +LDFLAGS = [] +LIBS = [] +GCC_LIST = ['router_http'] diff --git a/plugins/router_uwsgi/router_uwsgi.c b/plugins/router_uwsgi/router_uwsgi.c index d4303635..bb4ba1e8 100644 --- a/plugins/router_uwsgi/router_uwsgi.c +++ b/plugins/router_uwsgi/router_uwsgi.c @@ -33,8 +33,6 @@ int uwsgi_routing_func_uwsgi_simple(struct wsgi_request *wsgi_req, struct uwsgi_ int uwsgi_routing_func_uwsgi_remote(struct wsgi_request *wsgi_req, struct uwsgi_route *ur) { - char buf[8192]; - ssize_t len; struct uwsgi_header *uh = (struct uwsgi_header *) ur->data; char *addr = ur->data + sizeof(struct uwsgi_header); @@ -62,29 +60,13 @@ int uwsgi_routing_func_uwsgi_remote(struct wsgi_request *wsgi_req, struct uwsgi_ return UWSGI_ROUTE_NEXT; } - for(;;) { - int ret = uwsgi_waitfd(uwsgi_fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); - if (ret > 0) { - len = read(uwsgi_fd, buf, 8192); - if (len == 0) { - break; - } - else if (len < 0) { - uwsgi_error("read()"); - break; - } - - if (write(wsgi_req->poll.fd, buf, len) != len) { - uwsgi_error("write()"); - break; - } - } - else { - uwsgi_log("timeout !!!\n"); - break; - } + ssize_t ret = uwsgi_pipe(uwsgi_fd, wsgi_req->poll.fd, 0); + if (ret > 0) { + wsgi_req->response_size += ret; + } + else { + uwsgi_log("unable to manage uwsgi route response for %s\n", addr); } - close(uwsgi_fd); return UWSGI_ROUTE_BREAK; diff --git a/proto/http.c b/proto/http.c index ed39d8a6..d7daa7c6 100644 --- a/proto/http.c +++ b/proto/http.c @@ -369,3 +369,88 @@ int uwsgi_proto_http_parser(struct wsgi_request *wsgi_req) { return UWSGI_AGAIN; } + +void uwsgi_httpize_var(char *buf, size_t len) { + size_t i; + int upper = 1; + for(i=0;imethod, wsgi_req->method_len)) goto clear; + if (uwsgi_buffer_append(ub, " ", 1)) goto clear; + + if (uwsgi_buffer_append(ub, wsgi_req->uri, wsgi_req->uri_len)) goto clear; + if (uwsgi_buffer_append(ub, " ", 1)) goto clear; + + if (uwsgi_buffer_append(ub, wsgi_req->protocol, wsgi_req->protocol_len)) goto clear; + if (uwsgi_buffer_append(ub, "\r\n", 2)) goto clear; + + int i; + char *x_forwarded_for = NULL; + size_t x_forwarded_for_len = 0; + + // starting adding headers + for(i=0;ivar_cnt;i++) { + if (!uwsgi_starts_with(wsgi_req->hvec[i].iov_base, wsgi_req->hvec[i].iov_len, "HTTP_", 5)) { + + char *header = wsgi_req->hvec[i].iov_base+5; + size_t header_len = wsgi_req->hvec[i].iov_len-5; + + if (!uwsgi_strncmp(header, header_len, "CONNECTION", 10)) goto next; + if (!uwsgi_strncmp(header, header_len, "KEEP_ALIVE", 10)) goto next; + if (!uwsgi_strncmp(header, header_len, "X_FORWARDED_FOR", 15)) { + x_forwarded_for = wsgi_req->hvec[i+1].iov_base; + x_forwarded_for_len = wsgi_req->hvec[i+1].iov_len; + goto next; + } + + if (uwsgi_buffer_append(ub, header, header_len)) goto clear; + + // transofmr uwsgi var to http header + uwsgi_httpize_var((ub->buf+ub->pos) - header_len, header_len); + + if (uwsgi_buffer_append(ub, ": ", 2)) goto clear; + if (uwsgi_buffer_append(ub, wsgi_req->hvec[i+1].iov_base, wsgi_req->hvec[i+1].iov_len)) goto clear; + if (uwsgi_buffer_append(ub, "\r\n", 2)) goto clear; + + } +next: + i++; + } + + // append required headers + if (uwsgi_buffer_append(ub, "Connection: close\r\n", 19)) goto clear; + if (uwsgi_buffer_append(ub, "X-Forwarded-For: ", 17)) goto clear; + + if (x_forwarded_for_len > 0) { + if (uwsgi_buffer_append(ub, x_forwarded_for, x_forwarded_for_len)) goto clear; + if (uwsgi_buffer_append(ub, ", ", 2)) goto clear; + } + + if (uwsgi_buffer_append(ub, wsgi_req->remote_addr, wsgi_req->remote_addr_len)) goto clear; + + if (uwsgi_buffer_append(ub, "\r\n\r\n", 4)) goto clear; + + return ub; +clear: + uwsgi_buffer_destroy(ub); + return NULL; +} + diff --git a/proto/zeromq.c b/proto/zeromq.c index 7427b9ca..e9efb72b 100644 --- a/proto/zeromq.c +++ b/proto/zeromq.c @@ -2,15 +2,23 @@ extern struct uwsgi_server uwsgi; -void uwsgi_zeromq_init() { - uwsgi.zmq_context = zmq_init(1); +void *uwsgi_zeromq_init() { + if (!uwsgi.zmq_context) { + uwsgi.zmq_context = zmq_init(1); if (uwsgi.zmq_context == NULL) { uwsgi_error("zmq_init()"); exit(1); } + } + return uwsgi.zmq_context; +} - struct uwsgi_socket *uwsgi_sock = uwsgi.sockets; - while(uwsgi_sock) { +void uwsgi_zeromq_init_sockets() { + + uwsgi_zeromq_init(); + + struct uwsgi_socket *uwsgi_sock = uwsgi.sockets; + while(uwsgi_sock) { if (!uwsgi_sock->proto_name || strcmp(uwsgi_sock->proto_name, "zmq")) { goto zmq_next; } diff --git a/uwsgi.h b/uwsgi.h index 3cafc4a5..c81c76e4 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -289,6 +289,11 @@ extern int pivot_root(const char *new_root, const char *put_old); #endif +struct uwsgi_buffer { + char *buf; + off_t pos; + size_t len; +}; struct uwsgi_string_list { @@ -2265,7 +2270,9 @@ void add_exported_option(char *, char *, int); ssize_t uwsgi_send_empty_pkt(int, char *, uint8_t, uint8_t); -int uwsgi_waitfd(int, int); +int uwsgi_waitfd_event(int, int, int); +#define uwsgi_waitfd(a, b) uwsgi_waitfd_event(a, b, POLLIN) +#define uwsgi_waitfd_write(a, b) uwsgi_waitfd_event(a, b, POLLOUT) int uwsgi_hooked_parse_dict_dgram(int, char *, size_t, uint8_t, uint8_t, void (*)(char *, uint16_t, char *, uint16_t, void *), void *); int uwsgi_hooked_parse(char *, size_t, void (*)(char *, uint16_t, char *, uint16_t, void *), void *); @@ -2283,6 +2290,7 @@ int uwsgi_cluster_add_me(void); char *generate_socket_name(char *); #define UMIN(a,b) ((a)>(b)?(b):(a)) +#define UMAX(a,b) ((a)<(b)?(b):(a)) ssize_t uwsgi_send_message(int, uint8_t, uint8_t, char *, uint16_t, int, ssize_t, int); @@ -2627,7 +2635,8 @@ ssize_t uwsgi_proto_zeromq_write(struct wsgi_request *, char *, size_t); ssize_t uwsgi_proto_zeromq_write_header(struct wsgi_request *, char *, size_t); ssize_t uwsgi_proto_zeromq_sendfile(struct wsgi_request *); int uwsgi_proto_zeromq_parser(struct wsgi_request *); -void uwsgi_zeromq_init(void); +void *uwsgi_zeromq_init(void); +void uwsgi_zeromq_init_sockets(void); #endif int uwsgi_num2str2(int, char *); @@ -3202,6 +3211,18 @@ void uwsgi_emperor_start(void); void uwsgi_bind_sockets(void); void uwsgi_set_sockets_protocols(void); +struct uwsgi_buffer *uwsgi_buffer_new(size_t); +int uwsgi_buffer_append(struct uwsgi_buffer *, char *, size_t); +void uwsgi_buffer_destroy(struct uwsgi_buffer *); + +void uwsgi_httpize_var(char *, size_t); +struct uwsgi_buffer *uwsgi_to_http(struct wsgi_request *); + +ssize_t uwsgi_pipe(int, int, int); +ssize_t uwsgi_pipe_sized(int, int, size_t, int); + +int uwsgi_buffer_send(struct uwsgi_buffer *, int); + void uwsgi_check_emperor(void); #ifdef UWSGI_AS_SHARED_LIBRARY int uwsgi_init(int, char **, char **); diff --git a/uwsgiconfig.py b/uwsgiconfig.py index 9b11ebb7..3e07f9ff 100644 --- a/uwsgiconfig.py +++ b/uwsgiconfig.py @@ -357,7 +357,7 @@ class uConf(object): self.config.read(filename) self.gcc_list = ['core/utils', 'core/protocol', 'core/socket', 'core/logging', 'core/master', 'core/master_utils', 'core/emperor', 'core/notify', 'core/mule', 'core/subscription', 'core/stats', - 'core/setup_utils', 'core/clock', 'core/init', + 'core/setup_utils', 'core/clock', 'core/init', 'core/buffer', 'core/plugins', 'core/lock', 'core/cache', 'core/queue', 'core/event', 'core/signal', 'core/cluster', 'core/rpc', 'core/gateway', 'core/loop', 'lib/rbtree', 'core/rb_timers', 'core/uwsgi']