From 4a571153369457e6e2eb90e77cbc8e70124d9e62 Mon Sep 17 00:00:00 2001 From: "roberto@natty32" Date: Mon, 18 Apr 2011 17:07:13 +0200 Subject: [PATCH] edge triggered events support --- async.c | 14 +- buildconf/default.ini | 1 + loop.c | 2 +- mongrel2-uwsgi.conf | 22 ++ plugins/python/wsgi_subhandler.c | 2 +- proto/base.c | 12 ++ proto/fastcgi.c | 2 +- proto/http.c | 2 +- proto/zeromq.c | 339 +++++++++++++++++++++++++++++++ utils.c | 71 ++++--- uwsgi.c | 91 +++++++++ uwsgi.h | 43 +++- uwsgiconfig.py | 17 ++ 13 files changed, 584 insertions(+), 34 deletions(-) create mode 100644 mongrel2-uwsgi.conf create mode 100644 proto/base.c create mode 100644 proto/zeromq.c diff --git a/async.c b/async.c index f72e9be7..b6c517cc 100644 --- a/async.c +++ b/async.c @@ -263,11 +263,13 @@ void *async_loop(void *arg1) { break;; } - wsgi_req_setup(uwsgi.wsgi_req, uwsgi.wsgi_req->async_id ); + wsgi_req_setup(uwsgi.wsgi_req, uwsgi.wsgi_req->async_id, j ); if (wsgi_req_simple_accept(uwsgi.wsgi_req, interesting_fd)) { #ifdef UWSGI_EVENT_USE_PORT event_queue_add_fd_read(uwsgi.async_queue, interesting_fd); #endif + uwsgi.async_queue_unused_ptr++; + uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr] = uwsgi.wsgi_req; break; } #ifdef UWSGI_EVENT_USE_PORT @@ -280,15 +282,23 @@ void *async_loop(void *arg1) { /* re-set blocking socket */ if (fcntl(uwsgi.wsgi_req->poll.fd, F_SETFL, uwsgi.sockets[j].arg) < 0) { uwsgi_error("fcntl()"); + uwsgi.async_queue_unused_ptr++; + uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr] = uwsgi.wsgi_req; break; } } #endif - if (wsgi_req_async_recv(uwsgi.wsgi_req, j)) { + if (wsgi_req_async_recv(uwsgi.wsgi_req)) { + uwsgi.async_queue_unused_ptr++; + uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr] = uwsgi.wsgi_req; break; } + if (uwsgi.wsgi_req->do_not_add_to_async_queue) { + runqueue_push(uwsgi.wsgi_req); + } + break; } } diff --git a/buildconf/default.ini b/buildconf/default.ini index 9220ecbe..00a3de2d 100644 --- a/buildconf/default.ini +++ b/buildconf/default.ini @@ -3,6 +3,7 @@ xml = true ini = true yaml = true json = auto +zeromq = auto snmp = true sctp = false spooler = true diff --git a/loop.c b/loop.c index 77a7b38b..610ad4b4 100644 --- a/loop.c +++ b/loop.c @@ -69,7 +69,7 @@ void *simple_loop(void *arg1) { UWSGI_CLEAR_STATUS; - wsgi_req_setup(wsgi_req, core_id); + wsgi_req_setup(wsgi_req, core_id, -1); if (wsgi_req_accept(wsgi_req)) { continue; diff --git a/mongrel2-uwsgi.conf b/mongrel2-uwsgi.conf new file mode 100644 index 00000000..6376b7ca --- /dev/null +++ b/mongrel2-uwsgi.conf @@ -0,0 +1,22 @@ +main = Server( + uuid="f400bf85-4538-4f7a-8908-67e313d515c2", + access_log="/logs/access.log", + error_log="/logs/error.log", + chroot="./", + default_host="192.168.173.32", + name="test", + pid_file="/run/mongrel2.pid", + port=6767, + hosts = [ + Host(name="192.168.173.32", routes={ + '/': Handler(send_spec='tcp://127.0.0.1:9999', + send_ident='54c6755b-9628-40a4-9a2d-cc82a816345e', + recv_spec='tcp://127.0.0.1:9998', recv_ident='') + }) + ] +) + +settings = {'upload.temp_store':'/tmp/mongrel2.upload.XXXXXX'} +servers = [main] + + diff --git a/plugins/python/wsgi_subhandler.c b/plugins/python/wsgi_subhandler.c index a95f0670..2ae6d258 100644 --- a/plugins/python/wsgi_subhandler.c +++ b/plugins/python/wsgi_subhandler.c @@ -216,7 +216,7 @@ clear: } if (wsgi_req->async_post && !wsgi_req->fd_closed) { fclose(wsgi_req->async_post); - if ( (!uwsgi.post_buffering || wsgi_req->post_cl <= (size_t) uwsgi.post_buffering) && !wsgi_req->body_as_file) { + if ( !uwsgi.post_buffering || wsgi_req->post_cl <= (size_t) uwsgi.post_buffering) { wsgi_req->fd_closed = 1; } } diff --git a/proto/base.c b/proto/base.c new file mode 100644 index 00000000..266ccce1 --- /dev/null +++ b/proto/base.c @@ -0,0 +1,12 @@ + +#include "../uwsgi.h" + +int uwsgi_proto_base_accept(struct wsgi_request *wsgi_req, int fd) { + + int client_fd = accept(fd, (struct sockaddr *) &wsgi_req->c_addr, (socklen_t *) &wsgi_req->c_len); + if (client_fd < 0) { + uwsgi_error("accept()"); + } + + return client_fd; +} diff --git a/proto/fastcgi.c b/proto/fastcgi.c index 945d52ce..f03ce5d8 100644 --- a/proto/fastcgi.c +++ b/proto/fastcgi.c @@ -50,6 +50,7 @@ int uwsgi_proto_fastcgi_parser(struct wsgi_request *wsgi_req) { // allocate space for a fastcgi record if (!wsgi_req->proto_parser_buf) { wsgi_req->proto_parser_buf = uwsgi_malloc(8 + 65536); + wsgi_req->body_as_file = 1; } if (wsgi_req->proto_parser_status == PROTO_STATUS_RECV_HDR) { @@ -142,7 +143,6 @@ int uwsgi_proto_fastcgi_parser(struct wsgi_request *wsgi_req) { } // stdin else if (fr->type == 5) { - wsgi_req->body_as_file = 1; if (!wsgi_req->async_post) { wsgi_req->async_post = tmpfile(); if (!wsgi_req->async_post) { diff --git a/proto/http.c b/proto/http.c index 7713c615..9690fc33 100644 --- a/proto/http.c +++ b/proto/http.c @@ -218,6 +218,7 @@ int uwsgi_proto_http_parser(struct wsgi_request *wsgi_req) { // first round ? this memory area will be freed by async_loop if (!wsgi_req->proto_parser_buf) { wsgi_req->proto_parser_buf = uwsgi_malloc(uwsgi.buffer_size); + wsgi_req->body_as_file = 1; } if (wsgi_req->post_cl) { @@ -273,7 +274,6 @@ int uwsgi_proto_http_parser(struct wsgi_request *wsgi_req) { ret = http_parse(wsgi_req, ptr); //is there a Content_Length ? if (wsgi_req->post_cl) { - wsgi_req->body_as_file = 1; wsgi_req->async_post = tmpfile(); if (!wsgi_req->async_post) { free(wsgi_req->proto_parser_buf); diff --git a/proto/zeromq.c b/proto/zeromq.c new file mode 100644 index 00000000..2ae5ed17 --- /dev/null +++ b/proto/zeromq.c @@ -0,0 +1,339 @@ +#include "../uwsgi.h" + +extern struct uwsgi_server uwsgi; + +#include + +static uint16_t http_add_uwsgi_var(struct wsgi_request *wsgi_req, char *key, uint16_t keylen, char *val, uint16_t vallen) { + + + char *buffer = wsgi_req->buffer+wsgi_req->uh.pktsize; + char *watermark = wsgi_req->buffer+uwsgi.buffer_size; + char *ptr = buffer; + + if (buffer+keylen+vallen+2+2 >= watermark) { + uwsgi_log("[WARNING] unable to add %.*s=%.*s to uwsgi packet, consider increasing buffer size\n", keylen, key, vallen, val); + return 0; + } + + + *ptr++= (uint8_t) (keylen & 0xff); + *ptr++= (uint8_t) ((keylen >> 8) & 0xff); + memcpy(ptr, key, keylen); ptr+=keylen; + + *ptr++= (uint8_t) (vallen & 0xff); + *ptr++= (uint8_t) ((vallen >> 8) & 0xff); + memcpy(ptr, val, vallen); + +#ifdef UWSGI_DEBUG + uwsgi_log("add uwsgi var: %.*s = %.*s\n", keylen, key, vallen, val); +#endif + + return keylen+vallen+2+2; +} + +int uwsgi_proto_zeromq_parser(struct wsgi_request *wsgi_req) { + + return UWSGI_OK; +} + +int uwsgi_proto_zeromq_accept(struct wsgi_request *wsgi_req, int fd) { + + zmq_msg_t message; + char *ptr; + char *req_uuid = NULL; int req_uuid_len = 0; + char *req_id = NULL; int req_id_len = 0; + char *req_path = NULL; int req_path_len = 0; + char *req_body_size = NULL; int req_body_size_len = 0; + json_t *root; + json_error_t error; + void *json_iter; + json_t *json_value; + char *json_key, *json_val; + char *query_string = NULL; int query_string_len = 0; + int script_name_len = 0; + int i; + int resp_id_len; + uint32_t events; + size_t events_len = sizeof(uint32_t); + + + if (zmq_getsockopt(uwsgi.zmq_pull, ZMQ_EVENTS, &events, &events_len) < 0) { + uwsgi_error("zmq_getsockopt()"); + uwsgi.edge_triggered = 0; + return -1; + } + + if (events & ZMQ_POLLIN || uwsgi.edge_triggered) { + wsgi_req->body_as_file = 1; + wsgi_req->do_not_add_to_async_queue = 1; + zmq_msg_init(&message); + if (zmq_recv(uwsgi.zmq_pull, &message, ZMQ_NOBLOCK) < 0) { + if (errno == EAGAIN) { + uwsgi.edge_triggered = 0; + } + else { + uwsgi_error("zmq_recv()"); + } + zmq_msg_close(&message); + return -1; + } + wsgi_req->proto_parser_pos = zmq_msg_size(&message); + if (wsgi_req->proto_parser_pos > 65536) { + uwsgi_log("too much big message %d\n", wsgi_req->proto_parser_pos); + zmq_msg_close(&message); + return -1; + } + wsgi_req->proto_parser_buf = uwsgi_malloc(wsgi_req->proto_parser_pos); + + ptr = zmq_msg_data(&message); + for(i=0;i<(int)wsgi_req->proto_parser_pos;i++) { + if (ptr[i] == ' ') { + // get uuid + if (wsgi_req->proto_parser_status == 0) { + req_uuid = ptr; + req_uuid_len = i; + wsgi_req->proto_parser_status = 1; + } + // get req_id + else if (wsgi_req->proto_parser_status == 1) { + req_id = req_uuid+req_uuid_len+1; + req_id_len = (ptr+i) - req_id; + wsgi_req->proto_parser_status = 2; + } + // get path + else if (wsgi_req->proto_parser_status == 2) { + req_path = req_id+req_id_len+1; + req_path_len = (ptr+i) - req_path; + wsgi_req->proto_parser_status = 3; + } + } + // get body size; + else if (ptr[i] == ':') { + if (wsgi_req->proto_parser_status == 3) { + req_body_size = req_path+req_path_len+1; + req_body_size_len = (ptr+i) - req_body_size; + // check if remaining bytes < wsgi_req->proto_parser_pos + memcpy(wsgi_req->proto_parser_buf, ptr+i+1, uwsgi_str_num(req_body_size, req_body_size_len)); + ((char *)wsgi_req->proto_parser_buf)[uwsgi_str_num(req_body_size, req_body_size_len)] = 0; + wsgi_req->proto_parser_status = 4; + break; + } + } + } + + + if (wsgi_req->proto_parser_status >= 4) { + // ok ready to parse json data and build uwsgi request + root = json_loads(wsgi_req->proto_parser_buf, 0, &error); + if (!root) { + uwsgi_log("error parsing JSON data: line %d %s\n", error.line, error.text); + zmq_msg_close(&message); + free(wsgi_req->proto_parser_buf); + return -1; + } + + json_value = json_object_get(root, "VERSION"); + if (json_is_string(json_value)) { + json_val = (char *)json_string_value(json_value); + wsgi_req->uh.pktsize += http_add_uwsgi_var(wsgi_req, "SERVER_PROTOCOL", 15, json_val, strlen(json_val)); + } + + json_value = json_object_get(root, "QUERY"); + if (json_is_string(json_value)) { + query_string = (char *)json_string_value(json_value); + query_string_len = strlen(query_string); + wsgi_req->uh.pktsize += http_add_uwsgi_var(wsgi_req, "QUERY_STRING", 12, query_string, query_string_len); + } + + json_value = json_object_get(root, "METHOD"); + if (json_is_string(json_value)) { + json_val = (char *)json_string_value(json_value); + if (!strcmp(json_val, "JSON")) { + zmq_msg_close(&message); + free(wsgi_req->proto_parser_buf); + return -1; + } + wsgi_req->uh.pktsize += http_add_uwsgi_var(wsgi_req, "REQUEST_METHOD", 14, json_val, strlen(json_val)); + } + + json_value = json_object_get(root, "PATTERN"); + if (json_is_string(json_value)) { + json_val = (char *)json_string_value(json_value); + wsgi_req->uh.pktsize += http_add_uwsgi_var(wsgi_req, "SCRIPT_NAME", 11, json_val, strlen(json_val)-1); + script_name_len = strlen(json_val)-1; + } + + json_value = json_object_get(root, "PATH"); + if (json_is_string(json_value)) { + json_val = (char *)json_string_value(json_value); + wsgi_req->uh.pktsize += http_add_uwsgi_var(wsgi_req, "PATH_INFO", 9, json_val+script_name_len, strlen(json_val+script_name_len)); + if (query_string_len) { + char *request_uri = uwsgi_concat3n(json_val, strlen(json_val), "?", 1, query_string, query_string_len); + wsgi_req->uh.pktsize += http_add_uwsgi_var(wsgi_req, "REQUEST_URI", 11 , request_uri, strlen(json_val)+1+query_string_len ); + free(request_uri); + } + else { + wsgi_req->uh.pktsize += http_add_uwsgi_var(wsgi_req, "REQUEST_URI", 11, json_val, strlen(json_val)); + } + } + + json_value = json_object_get(root, "host"); + if (json_is_string(json_value)) { + json_val = (char *)json_string_value(json_value); + char *colon = strchr(json_val, ':'); + if (colon) { + wsgi_req->uh.pktsize += http_add_uwsgi_var(wsgi_req, "SERVER_PORT", 11, colon+1, strlen(colon+1)); + } + else { + wsgi_req->uh.pktsize += http_add_uwsgi_var(wsgi_req, "SERVER_PORT", 11, "80", 2); + } + } + + wsgi_req->uh.pktsize += http_add_uwsgi_var(wsgi_req, "SERVER_NAME", 11, uwsgi.hostname, uwsgi.hostname_len); + + json_iter = json_object_iter(root); + + while(json_iter) { + json_key = (char *)json_object_iter_key(json_iter); + // is it a header ? + if (json_key[0] >= 97) { + json_value = json_object_iter_value(json_iter); + if (json_is_string(json_value)) { + json_val = (char *)json_string_value(json_value); + wsgi_req->uh.pktsize += http_add_uwsgi_var(wsgi_req, json_key, strlen(json_key), json_val, strlen(json_val)); + } + } + json_iter = json_object_iter_next(root, json_iter); + } + + memcpy(wsgi_req->proto_parser_buf, req_uuid, req_uuid_len); + ((char *)wsgi_req->proto_parser_buf)[req_uuid_len] = ' '; + resp_id_len = uwsgi_num2str2(req_id_len, wsgi_req->proto_parser_buf+req_uuid_len+1); + ((char *)wsgi_req->proto_parser_buf)[req_uuid_len+1+resp_id_len] = ':'; + + memcpy((char *)wsgi_req->proto_parser_buf+req_uuid_len+1+resp_id_len+1, req_id, req_id_len); + + ((char *)wsgi_req->proto_parser_buf)[req_uuid_len+1+resp_id_len+1+req_id_len] = ','; + ((char *)wsgi_req->proto_parser_buf)[req_uuid_len+1+resp_id_len+1+req_id_len+1] = ' '; + wsgi_req->proto_parser_pos = (uint64_t) req_uuid_len+1+resp_id_len+1+req_id_len+1+1; + + } + + // get req_id + zmq_msg_close(&message); + + + /* + while(events & ZMQ_POLLIN) { + if (zmq_getsockopt(uwsgi.zmq_pull, ZMQ_EVENTS, &events, &events_len) < 0) { + uwsgi_error("zmq_getsockopt()"); + free(wsgi_req->proto_parser_buf); + return -1; + } + } + */ + + // zeromq is edge-triggered + uwsgi.edge_triggered = 1; + + return 0; + } + + return -1; +} + +void uwsgi_proto_zeromq_close(struct wsgi_request *wsgi_req) { + zmq_msg_t reply; + + //uwsgi_log("CLOSING |%.*s|\n", (int)wsgi_req->proto_parser_pos, wsgi_req->proto_parser_buf); + zmq_msg_init_data(&reply, wsgi_req->proto_parser_buf, wsgi_req->proto_parser_pos, NULL, NULL); + if (zmq_send(uwsgi.zmq_pub, &reply, 0)) { + uwsgi_error("zmq_send()"); + } + zmq_msg_close(&reply); + free(wsgi_req->proto_parser_buf); + +} + + +ssize_t uwsgi_proto_zeromq_writev_header(struct wsgi_request *wsgi_req, struct iovec *iovec, size_t iov_len) { + int i; + ssize_t len; + ssize_t ret = 0; + + for(i=0;i<(int)iov_len;i++) { + len = uwsgi_proto_zeromq_write(wsgi_req, iovec[i].iov_base, iovec[i].iov_len); + if (len <= 0) { + return len; + } + ret += len; + } + + return ret; +} + +ssize_t uwsgi_proto_zeromq_writev(struct wsgi_request *wsgi_req, struct iovec *iovec, size_t iov_len) { + return uwsgi_proto_zeromq_writev_header(wsgi_req, iovec, iov_len); +} + +ssize_t uwsgi_proto_zeromq_write(struct wsgi_request *wsgi_req, char *buf, size_t len) { + zmq_msg_t reply; + char *zmq_body; + + if (len == 0) return 0; + + zmq_body = uwsgi_concat2n(wsgi_req->proto_parser_buf, (int) wsgi_req->proto_parser_pos, buf, len); + + //uwsgi_log("|%.*s|\n", (int)wsgi_req->proto_parser_pos+len, zmq_body); + + zmq_msg_init_data(&reply, zmq_body, wsgi_req->proto_parser_pos+len, NULL, NULL); + if (zmq_send(uwsgi.zmq_pub, &reply, 0)) { + uwsgi_error("zmq_send()"); + zmq_msg_close(&reply); + return -1; + } + zmq_msg_close(&reply); + free(zmq_body); + + return len; +} + +ssize_t uwsgi_proto_zeromq_write_header(struct wsgi_request *wsgi_req, char *buf, size_t len) { + return uwsgi_proto_zeromq_write(wsgi_req, buf, len); +} + +ssize_t uwsgi_proto_zeromq_sendfile(struct wsgi_request *wsgi_req) { + + ssize_t len; + char buf[65536]; + size_t remains = wsgi_req->sendfile_fd_size-wsgi_req->sendfile_fd_pos; + + wsgi_req->sendfile_fd_chunk = 65536; + + + if (uwsgi.async > 1) { + len = read(wsgi_req->sendfile_fd, buf, UMIN(remains, wsgi_req->sendfile_fd_chunk)); + if (len != (int)UMIN(remains, wsgi_req->sendfile_fd_chunk)) { + uwsgi_error("read()"); + return -1; + } + wsgi_req->sendfile_fd_pos+=len; + return uwsgi_proto_zeromq_write(wsgi_req, buf, len); + } + + while(remains) { + len = read(wsgi_req->sendfile_fd, buf, UMIN(remains, wsgi_req->sendfile_fd_chunk)); + if (len != (int)UMIN(remains, wsgi_req->sendfile_fd_chunk)) { + uwsgi_error("read()"); + return -1; + } + wsgi_req->sendfile_fd_pos+=len; + len = uwsgi_proto_zeromq_write(wsgi_req, buf, len); + remains = wsgi_req->sendfile_fd_size-wsgi_req->sendfile_fd_pos; + } + + return wsgi_req->sendfile_fd_pos; + +} + diff --git a/utils.c b/utils.c index 9cc68b30..0585fac8 100644 --- a/utils.c +++ b/utils.c @@ -413,7 +413,7 @@ void uwsgi_close_request(struct wsgi_request *wsgi_req) { // close the connection with the webserver - if (!wsgi_req->fd_closed) { + if (!wsgi_req->fd_closed || wsgi_req->body_as_file) { // NOTE, if we close the socket before receiving eventually sent data, socket layer will send a RST wsgi_req->socket_proto_close(wsgi_req); } @@ -463,7 +463,7 @@ void uwsgi_close_request(struct wsgi_request *wsgi_req) { } -void wsgi_req_setup(struct wsgi_request *wsgi_req, int async_id) { +void wsgi_req_setup(struct wsgi_request *wsgi_req, int async_id, int socket_id) { wsgi_req->poll.events = POLLIN; wsgi_req->app_id = uwsgi.default_app; @@ -483,28 +483,32 @@ void wsgi_req_setup(struct wsgi_request *wsgi_req, int async_id) { wsgi_req->post_buffering_buf = uwsgi.async_post_buf[wsgi_req->async_id]; } + if (socket_id > -1) { + wsgi_req->socket_proto = uwsgi.sockets[socket_id].proto; + wsgi_req->socket_proto_accept = uwsgi.sockets[socket_id].proto_accept; + wsgi_req->socket_proto_write = uwsgi.sockets[socket_id].proto_write; + wsgi_req->socket_proto_writev = uwsgi.sockets[socket_id].proto_writev; + wsgi_req->socket_proto_write_header = uwsgi.sockets[socket_id].proto_write_header; + wsgi_req->socket_proto_writev_header = uwsgi.sockets[socket_id].proto_writev_header; + wsgi_req->socket_proto_sendfile = uwsgi.sockets[socket_id].proto_sendfile; + wsgi_req->socket_proto_close = uwsgi.sockets[socket_id].proto_close; + } } -int wsgi_req_async_recv(struct wsgi_request *wsgi_req, int socket_id) { +int wsgi_req_async_recv(struct wsgi_request *wsgi_req) { UWSGI_SET_IN_REQUEST; gettimeofday(&wsgi_req->start_of_request, NULL); - if (event_queue_add_fd_read( uwsgi.async_queue, wsgi_req->poll.fd ) < 0) return -1; + if (!wsgi_req->do_not_add_to_async_queue) { + if (event_queue_add_fd_read( uwsgi.async_queue, wsgi_req->poll.fd ) < 0) return -1; - async_add_timeout(wsgi_req, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + async_add_timeout(wsgi_req, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + uwsgi.async_proto_fd_table[wsgi_req->poll.fd] = wsgi_req; + } - uwsgi.async_proto_fd_table[wsgi_req->poll.fd] = wsgi_req; - - wsgi_req->socket_proto = uwsgi.sockets[socket_id].proto; - wsgi_req->socket_proto_write = uwsgi.sockets[socket_id].proto_write; - wsgi_req->socket_proto_writev = uwsgi.sockets[socket_id].proto_writev; - wsgi_req->socket_proto_write_header = uwsgi.sockets[socket_id].proto_write_header; - wsgi_req->socket_proto_writev_header = uwsgi.sockets[socket_id].proto_writev_header; - wsgi_req->socket_proto_sendfile = uwsgi.sockets[socket_id].proto_sendfile; - wsgi_req->socket_proto_close = uwsgi.sockets[socket_id].proto_close; // enter harakiri mode if (uwsgi.shared->options[UWSGI_OPTION_HARAKIRI] > 0) { @@ -538,10 +542,9 @@ int wsgi_req_recv(struct wsgi_request *wsgi_req) { int wsgi_req_simple_accept(struct wsgi_request *wsgi_req, int fd) { - wsgi_req->poll.fd = accept(fd, (struct sockaddr *) &wsgi_req->c_addr, (socklen_t *) &wsgi_req->c_len); + wsgi_req->poll.fd = wsgi_req->socket_proto_accept(wsgi_req, fd); if (wsgi_req->poll.fd < 0) { - uwsgi_error("accept()"); return -1; } @@ -558,6 +561,13 @@ int wsgi_req_accept(struct wsgi_request *wsgi_req) { int ret; char uwsgi_signal; + if (uwsgi.edge_triggered) { + uwsgi_log("EDGE TRIGGERED\n"); + for(i=0;ipoll.fd = accept(uwsgi.sockets_poll[i].fd, (struct sockaddr *) &wsgi_req->c_addr, (socklen_t *) &wsgi_req->c_len); + if (uwsgi.sockets_poll[i].revents & POLLIN || (uwsgi.edge_triggered && uwsgi.sockets[i].edge_trigger)) { + int socket_id = i; + wsgi_req->socket_proto = uwsgi.sockets[socket_id].proto; + wsgi_req->socket_proto_accept = uwsgi.sockets[socket_id].proto_accept; + wsgi_req->socket_proto_write = uwsgi.sockets[socket_id].proto_write; + wsgi_req->socket_proto_writev = uwsgi.sockets[socket_id].proto_writev; + wsgi_req->socket_proto_write_header = uwsgi.sockets[socket_id].proto_write_header; + wsgi_req->socket_proto_writev_header = uwsgi.sockets[socket_id].proto_writev_header; + wsgi_req->socket_proto_sendfile = uwsgi.sockets[socket_id].proto_sendfile; + wsgi_req->socket_proto_close = uwsgi.sockets[socket_id].proto_close; + wsgi_req->poll.fd = wsgi_req->socket_proto_accept(wsgi_req, uwsgi.sockets_poll[i].fd); if (wsgi_req->poll.fd < 0) { - if (errno == EWOULDBLOCK) goto polling; + if (errno == EWOULDBLOCK) { uwsgi_log("GOTO polling\n"); goto polling;} uwsgi_error("accept()"); return -1; } @@ -612,13 +632,7 @@ polling: } // set socket protocol - wsgi_req->socket_proto = uwsgi.sockets[i].proto; - wsgi_req->socket_proto_write = uwsgi.sockets[i].proto_write; - wsgi_req->socket_proto_writev = uwsgi.sockets[i].proto_writev; - wsgi_req->socket_proto_write_header = uwsgi.sockets[i].proto_write_header; - wsgi_req->socket_proto_writev_header = uwsgi.sockets[i].proto_writev_header; - wsgi_req->socket_proto_sendfile = uwsgi.sockets[i].proto_sendfile; - wsgi_req->socket_proto_close = uwsgi.sockets[i].proto_close; + return 0; } } @@ -1694,6 +1708,11 @@ char *uwsgi_num2str(int num) { return str; } +int uwsgi_num2str2(int num, char *ptr) { + + return snprintf(ptr, 11, "%d", num); +} + int is_unix(char *socket_name, int len) { int i; for(i=0;ioptions[UWSGI_OPTION_CGI_MODE] = 1; uwsgi.sockets[i].proto = uwsgi_proto_fastcgi_parser; + uwsgi.sockets[i].proto_accept = uwsgi_proto_base_accept; uwsgi.sockets[i].proto_write = uwsgi_proto_fastcgi_write; uwsgi.sockets[i].proto_writev = uwsgi_proto_fastcgi_writev; uwsgi.sockets[i].proto_write_header = uwsgi_proto_fastcgi_write_header; @@ -1787,6 +1793,7 @@ int uwsgi_start(void *v_argv) { } else { uwsgi.sockets[i].proto = uwsgi_proto_uwsgi_parser; + uwsgi.sockets[i].proto_accept = uwsgi_proto_base_accept; uwsgi.sockets[i].proto_write = uwsgi_proto_uwsgi_write; uwsgi.sockets[i].proto_writev = uwsgi_proto_uwsgi_writev; uwsgi.sockets[i].proto_write_header = uwsgi_proto_uwsgi_write_header; @@ -1795,6 +1802,25 @@ int uwsgi_start(void *v_argv) { uwsgi.sockets[i].proto_close = uwsgi_proto_uwsgi_close; } } + +#ifdef UWSGI_ZEROMQ + if (uwsgi.zeromq) { + if (uwsgi.sockets_cnt+1 > MAX_SOCKETS) { + uwsgi_log("too much sockets defined, cannot add a zeromq receiver\n"); + exit(1); + } + uwsgi.zmq_responder = strchr(uwsgi.zeromq,','); + if (!uwsgi.zmq_responder) { + uwsgi_log("invalid zeromq address\n"); + exit(1); + } + uwsgi.zmq_receiver = uwsgi_concat2n(uwsgi.zeromq, uwsgi.zmq_responder-uwsgi.zeromq, "", 0); + uwsgi.zmq_responder++; + uwsgi_log("zmq receiver: %s\n", uwsgi.zmq_receiver); + uwsgi_log("zmq responder: %s\n", uwsgi.zmq_responder); + uwsgi.zmq_socket = uwsgi.sockets_cnt++; + } +#endif } @@ -2185,6 +2211,66 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100 } } +#ifdef UWSGI_ZEROMQ + if (uwsgi.zmq_receiver && uwsgi.zmq_responder) { + uwsgi.zmq_context = zmq_init(1); + if (uwsgi.zmq_context == NULL) { + uwsgi_error("zmq_init()"); + exit(1); + } + uwsgi.zmq_pull = zmq_socket(uwsgi.zmq_context, ZMQ_PULL); + if (uwsgi.zmq_pull == NULL) { + uwsgi_error("zmq_socket()"); + exit(1); + } + if (zmq_connect(uwsgi.zmq_pull, uwsgi.zmq_receiver) < 0) { + uwsgi_error("zmq_connect()"); + exit(1); + } + + uwsgi.zmq_pub = zmq_socket(uwsgi.zmq_context, ZMQ_PUB); + if (uwsgi.zmq_pub == NULL) { + uwsgi_error("zmq_socket()"); + exit(1); + } + if (zmq_setsockopt(uwsgi.zmq_pub, ZMQ_IDENTITY, "550e8400-e29b-41d4-a716-446655440000", strlen("550e8400-e29b-41d4-a716-446655440000")) < 0) { + uwsgi_error("zmq_setsockopt()"); + exit(1); + } + + if (zmq_connect(uwsgi.zmq_pub, uwsgi.zmq_responder) < 0) { + uwsgi_error("zmq_connect()"); + exit(1); + } + + uwsgi.sockets[uwsgi.zmq_socket].name = uwsgi.zmq_receiver; + uwsgi.sockets[uwsgi.zmq_socket].name_len = strlen(uwsgi.zmq_receiver); + + uwsgi.sockets[uwsgi.zmq_socket].proto = uwsgi_proto_zeromq_parser; + + uwsgi.sockets[uwsgi.zmq_socket].proto_accept = uwsgi_proto_zeromq_accept; + uwsgi.sockets[uwsgi.zmq_socket].proto_close = uwsgi_proto_zeromq_close; + + uwsgi.sockets[uwsgi.zmq_socket].proto_write = uwsgi_proto_zeromq_write; + uwsgi.sockets[uwsgi.zmq_socket].proto_writev = uwsgi_proto_zeromq_writev; + uwsgi.sockets[uwsgi.zmq_socket].proto_write_header = uwsgi_proto_zeromq_write_header; + uwsgi.sockets[uwsgi.zmq_socket].proto_writev_header = uwsgi_proto_zeromq_writev_header; + + uwsgi.sockets[uwsgi.zmq_socket].proto_sendfile = uwsgi_proto_zeromq_sendfile; + + uwsgi.sockets[uwsgi.zmq_socket].edge_trigger = 1; + size_t zmq_socket_len = sizeof(int); + if (zmq_getsockopt(uwsgi.zmq_pull, ZMQ_FD, &uwsgi.sockets[uwsgi.zmq_socket].fd, &zmq_socket_len) < 0) { + uwsgi_error("zmq_getsockopt()"); + exit(1); + } + + uwsgi.sockets_poll[uwsgi.zmq_socket].fd = uwsgi.sockets[uwsgi.zmq_socket].fd; + uwsgi.sockets_poll[uwsgi.zmq_socket].events = POLLIN; + uwsgi.sockets[uwsgi.zmq_socket].bound = 1; + } +#endif + //postpone the queue initialization as kevent //do not pass kfd after fork() #ifdef UWSGI_ASYNC @@ -2364,6 +2450,11 @@ end: case LONG_ARGS_WORKER_EXEC: uwsgi.worker_exec = optarg; return 1; +#ifdef UWSGI_ZEROMQ + case LONG_ARGS_ZEROMQ: + uwsgi.zeromq = optarg; + return 1; +#endif case LONG_ARGS_REMAP_MODIFIER: uwsgi.remap_modifier = optarg; return 1; diff --git a/uwsgi.h b/uwsgi.h index 2d03316f..2d03f575 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -199,6 +199,10 @@ extern int pivot_root(const char * new_root, const char * put_old); #undef __EXTENSIONS__ #endif +#ifdef UWSGI_ZEROMQ +#include +#endif + /* this value are taken from nginx */ #if defined(__APPLE__) || defined(__freebsd__) #define UWSGI_LISTEN_QUEUE -1 @@ -211,6 +215,7 @@ extern int pivot_root(const char * new_root, const char * put_old); #define uwsgi_cache_update_start(x, y, z) uwsgi_cache_set(x, y, "", 0, CACHE_FLAG_UNGETTABLE) + struct uwsgi_help_item { char *key; @@ -404,6 +409,7 @@ struct uwsgi_opt { #define LONG_ARGS_EMPEROR_AMQP_USERNAME 17097 #define LONG_ARGS_EMPEROR_AMQP_PASSWORD 17098 #define LONG_ARGS_PROTOCOL 17099 +#define LONG_ARGS_ZEROMQ 17100 #define UWSGI_OK 0 @@ -499,12 +505,14 @@ struct uwsgi_socket { void *ctx; int (*proto)(struct wsgi_request *); + int (*proto_accept)(struct wsgi_request *, int); ssize_t (*proto_write)(struct wsgi_request *, char *, size_t); ssize_t (*proto_writev)(struct wsgi_request *, struct iovec *, size_t); ssize_t (*proto_write_header)(struct wsgi_request *, char *, size_t); ssize_t (*proto_writev_header)(struct wsgi_request *, struct iovec *, size_t); ssize_t (*proto_sendfile)(struct wsgi_request *); void (*proto_close)(struct wsgi_request *); + int edge_trigger; }; struct uwsgi_server; @@ -713,6 +721,8 @@ struct wsgi_request { uint16_t var_cnt; uint16_t header_cnt; + int do_not_add_to_async_queue; + int status; size_t response_size; ssize_t headers_size; @@ -746,6 +756,7 @@ struct wsgi_request { uint64_t post_buffering_read; int (*socket_proto)(struct wsgi_request *); + int (*socket_proto_accept)(struct wsgi_request *, int); ssize_t (*socket_proto_write)(struct wsgi_request *, char *, size_t); ssize_t (*socket_proto_writev)(struct wsgi_request *, struct iovec *, size_t); ssize_t (*socket_proto_write_header)(struct wsgi_request *, char *, size_t); @@ -956,6 +967,8 @@ struct uwsgi_server { int async_queue; int async_nevents; + int edge_triggered; + int max_vars; int vec_size; @@ -1045,6 +1058,16 @@ struct uwsgi_server { char *protocol; int sockets_cnt; + +#ifdef UWSGI_ZEROMQ + char *zeromq; + char *zmq_receiver; + char *zmq_responder; + int zmq_socket; + void *zmq_context; + void *zmq_pull; + void *zmq_pub; +#endif struct uwsgi_socket sockets[MAX_SOCKETS]; // leave a slot for no-orphan mode struct pollfd sockets_poll[9]; @@ -1433,9 +1456,9 @@ void nagios(void); void uwsgi_close_request(struct wsgi_request *); -void wsgi_req_setup(struct wsgi_request *, int); +void wsgi_req_setup(struct wsgi_request *, int, int); int wsgi_req_recv(struct wsgi_request *); -int wsgi_req_async_recv(struct wsgi_request *, int); +int wsgi_req_async_recv(struct wsgi_request *); int wsgi_req_accept(struct wsgi_request *); int wsgi_req_simple_accept(struct wsgi_request *, int); @@ -1828,3 +1851,19 @@ ssize_t uwsgi_proto_fastcgi_write_header(struct wsgi_request *, char *, size_t); ssize_t uwsgi_proto_fastcgi_sendfile(struct wsgi_request *); void uwsgi_proto_fastcgi_close(struct wsgi_request *); + +int uwsgi_proto_base_accept(struct wsgi_request *, int); + +#ifdef UWSGI_ZEROMQ +int uwsgi_proto_zeromq_accept(struct wsgi_request *, int); +void uwsgi_proto_zeromq_close(struct wsgi_request *); +ssize_t uwsgi_proto_zeromq_writev_header(struct wsgi_request *, struct iovec *, size_t); +ssize_t uwsgi_proto_zeromq_writev(struct wsgi_request *, struct iovec *, size_t); +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 *); +ssize_t uwsgi_proto_zeromq_parser(struct wsgi_request *); +int uwsgi_proto_zeromq_parser(struct wsgi_request *); +#endif + +int uwsgi_num2str2(int, char *); diff --git a/uwsgiconfig.py b/uwsgiconfig.py index 95a1934d..45087d02 100644 --- a/uwsgiconfig.py +++ b/uwsgiconfig.py @@ -159,6 +159,7 @@ class uConf(object): self.gcc_list = ['utils', 'protocol', 'socket', 'logging', 'master', 'emperor', 'plugins', 'lock', 'cache', 'queue', 'event', 'signal', 'rpc', 'gateway', 'loop', 'lib/rbtree', 'lib/amqp', 'rb_timers', 'uwsgi'] # add protocols + self.gcc_list.append('proto/base') self.gcc_list.append('proto/uwsgi') self.gcc_list.append('proto/http') self.gcc_list.append('proto/fastcgi') @@ -377,6 +378,8 @@ class uConf(object): self.gcc_list.append('regexp') self.cflags.append("-DUWSGI_PCRE") + has_json = False + if self.get('async'): self.cflags.append("-DUWSGI_ASYNC") self.gcc_list.append('async') @@ -404,14 +407,17 @@ class uConf(object): self.cflags.append("-DUWSGI_JSON") self.gcc_list.append('json') self.libs.append(spcall("pkg-config --libs jansson")) + has_json = True elif os.path.exists('/usr/include/jansson.h') or os.path.exists('/usr/local/include/jansson.h'): self.cflags.append("-DUWSGI_JSON") self.gcc_list.append('json') self.libs.append('-ljansson') + has_json = True else: self.cflags.append("-DUWSGI_JSON") self.gcc_list.append('json') self.libs.append('-ljansson') + has_json = True if self.get('ldap'): if self.get('ldap') == 'auto': @@ -424,6 +430,17 @@ class uConf(object): self.gcc_list.append('ldap') self.libs.append('-lldap') + if has_json and self.get('zeromq'): + if self.get('zeromq') == 'auto': + if os.path.exists('/usr/include/zmq.h') or os.path.exists('/usr/local/include/zmq.h'): + self.cflags.append("-DUWSGI_ZEROMQ") + self.gcc_list.append('proto/zeromq') + self.libs.append('-lzmq') + else: + self.cflags.append("-DUWSGI_ZEROMQ") + self.gcc_list.append('proto/zeromq') + self.libs.append('-lzmq') + if self.get('evdis'): self.cflags.append("-DUWSGI_EVDIS") self.gcc_list.append('evdis')