From 913493fd2ed7c5222c65fdc320138c41edffbe9f Mon Sep 17 00:00:00 2001 From: "roberto@natty32" Date: Tue, 19 Apr 2011 06:59:34 +0200 Subject: [PATCH] mongrel2/zeromq support added --- async.c | 2 +- cache.c | 16 ++++---- logging.c | 2 + loop.c | 24 ++++-------- plugins/python/wsgi_handlers.c | 8 ++-- plugins/python/wsgi_headers.c | 2 +- plugins/python/wsgi_subhandler.c | 4 +- plugins/rpc/rpc_plugin.c | 4 +- proto/zeromq.c | 63 ++++++++++++++---------------- sendfile.c | 4 +- utils.c | 66 +++++++++++++------------------- uwsgi.c | 10 +++-- uwsgi.h | 13 +++---- 13 files changed, 94 insertions(+), 124 deletions(-) diff --git a/async.c b/async.c index b6c517cc..42e313bb 100644 --- a/async.c +++ b/async.c @@ -307,7 +307,7 @@ void *async_loop(void *arg1) { // proto event uwsgi.wsgi_req = find_wsgi_req_proto_by_fd(interesting_fd); if (uwsgi.wsgi_req) { - proto_parser_status = uwsgi.wsgi_req->socket_proto(uwsgi.wsgi_req); + proto_parser_status = uwsgi.wsgi_req->socket->proto(uwsgi.wsgi_req); // reset timeout rb_erase(&uwsgi.wsgi_req->async_timeout->rbt, uwsgi.rb_async_timeouts); free(uwsgi.wsgi_req->async_timeout); diff --git a/cache.c b/cache.c index 03197719..1f0a9864 100644 --- a/cache.c +++ b/cache.c @@ -497,17 +497,17 @@ void cache_command(char *key, uint16_t keylen, char *val, uint16_t vallen, void if (!uwsgi_strncmp(key, keylen, "key", 3)) { val = uwsgi_cache_get(val, vallen, &tmp_vallen); if (val && tmp_vallen > 0) { - wsgi_req->response_size = wsgi_req->socket_proto_write(wsgi_req, val, tmp_vallen); + wsgi_req->response_size = wsgi_req->socket->proto_write(wsgi_req, val, tmp_vallen); } } else if (!uwsgi_strncmp(key, keylen, "get", 3)) { val = uwsgi_cache_get(val, vallen, &tmp_vallen); if (val && vallen > 0) { - wsgi_req->response_size = wsgi_req->socket_proto_write(wsgi_req, val, tmp_vallen); + wsgi_req->response_size = wsgi_req->socket->proto_write(wsgi_req, val, tmp_vallen); } else { - wsgi_req->response_size = wsgi_req->socket_proto_write(wsgi_req, "HTTP/1.0 404 Not Found\r\n\r\n

Not Found

", 44); + wsgi_req->response_size = wsgi_req->socket->proto_write(wsgi_req, "HTTP/1.0 404 Not Found\r\n\r\n

Not Found

", 44); } } } @@ -527,8 +527,8 @@ int uwsgi_cache_request(struct wsgi_request *wsgi_req) { value = uwsgi_cache_get(wsgi_req->buffer, wsgi_req->uh.pktsize, &vallen); if (value && vallen > 0) { wsgi_req->uh.pktsize = vallen; - wsgi_req->response_size = wsgi_req->socket_proto_write(wsgi_req, (char *)&wsgi_req->uh, 4); - wsgi_req->response_size += wsgi_req->socket_proto_write(wsgi_req, value, vallen); + wsgi_req->response_size = wsgi_req->socket->proto_write(wsgi_req, (char *)&wsgi_req->uh, 4); + wsgi_req->response_size += wsgi_req->socket->proto_write(wsgi_req, value, vallen); } } break; @@ -563,13 +563,13 @@ int uwsgi_cache_request(struct wsgi_request *wsgi_req) { if (value && vallen > 0) { wsgi_req->uh.pktsize = 0; wsgi_req->uh.modifier2 = 1; - wsgi_req->response_size = wsgi_req->socket_proto_write(wsgi_req, (char *)&wsgi_req->uh, 4); - wsgi_req->response_size += wsgi_req->socket_proto_write(wsgi_req, value, vallen); + wsgi_req->response_size = wsgi_req->socket->proto_write(wsgi_req, (char *)&wsgi_req->uh, 4); + wsgi_req->response_size += wsgi_req->socket->proto_write(wsgi_req, value, vallen); } else { wsgi_req->uh.pktsize = 0; wsgi_req->uh.modifier2 = 0; - wsgi_req->response_size = wsgi_req->socket_proto_write(wsgi_req, (char *)&wsgi_req->uh, 4); + wsgi_req->response_size = wsgi_req->socket->proto_write(wsgi_req, (char *)&wsgi_req->uh, 4); } } break; diff --git a/logging.c b/logging.c index 1f1863a4..61014600 100644 --- a/logging.c +++ b/logging.c @@ -47,6 +47,8 @@ void log_request(struct wsgi_request *wsgi_req) { struct uwsgi_app *wi; + if (wsgi_req->do_not_log) return ; + if (wsgi_req->app_id >= 0) { wi = &uwsgi.apps[wsgi_req->app_id]; if (wi->requests > 0) { diff --git a/loop.c b/loop.c index f790c764..572fe771 100644 --- a/loop.c +++ b/loop.c @@ -82,7 +82,9 @@ void *simple_loop(void *arg1) { uwsgi_close_request(wsgi_req); } +#ifdef UWSGI_THREADING pthread_exit(NULL); +#endif //never here return NULL; @@ -93,34 +95,22 @@ void *zeromq_loop(void *arg1) { long core_id = (long) arg1; struct wsgi_request *wsgi_req = uwsgi.wsgi_requests[core_id]; + uwsgi.zeromq_recv_flag = 0; while (uwsgi.workers[uwsgi.mywid].manage_next_request) { UWSGI_CLEAR_STATUS; - wsgi_req_setup(wsgi_req, core_id, -1); uwsgi.edge_triggered = 1; int socket_id = uwsgi.zmq_socket; - 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->socket = &uwsgi.sockets[socket_id]; + wsgi_req->poll.fd = wsgi_req->socket->proto_accept(wsgi_req, uwsgi.sockets[socket_id].fd); - wsgi_req->poll.fd = wsgi_req->socket_proto_accept(wsgi_req, uwsgi.sockets[socket_id].fd); - - if (wsgi_req->poll.fd < 0) { - continue; - } - - if (wsgi_req_recv(wsgi_req)) { - continue; + if (wsgi_req->poll.fd >= 0) { + wsgi_req_recv(wsgi_req); } uwsgi_close_request(wsgi_req); diff --git a/plugins/python/wsgi_handlers.c b/plugins/python/wsgi_handlers.c index 02ac0aac..205e5ecd 100644 --- a/plugins/python/wsgi_handlers.c +++ b/plugins/python/wsgi_handlers.c @@ -171,7 +171,7 @@ PyObject *py_uwsgi_write(PyObject * self, PyObject * args) { content = PyString_AsString(data); len = PyString_Size(data); UWSGI_RELEASE_GIL - wsgi_req->response_size = wsgi_req->socket_proto_write(wsgi_req, content, len); + wsgi_req->response_size = wsgi_req->socket->proto_write(wsgi_req, content, len); UWSGI_GET_GIL } @@ -492,9 +492,9 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { // LOCK THIS PART - wsgi_req->response_size += wsgi_req->socket_proto_write(wsgi_req, wsgi_req->protocol, wsgi_req->protocol_len); - wsgi_req->response_size += wsgi_req->socket_proto_write(wsgi_req, " 500 Internal Server Error\r\n", 28 ); - wsgi_req->response_size += wsgi_req->socket_proto_write(wsgi_req, "Content-type: text/plain\r\n\r\n", 28 ); + wsgi_req->response_size += wsgi_req->socket->proto_write(wsgi_req, wsgi_req->protocol, wsgi_req->protocol_len); + wsgi_req->response_size += wsgi_req->socket->proto_write(wsgi_req, " 500 Internal Server Error\r\n", 28 ); + wsgi_req->response_size += wsgi_req->socket->proto_write(wsgi_req, "Content-type: text/plain\r\n\r\n", 28 ); wsgi_req->header_cnt = 1; /* diff --git a/plugins/python/wsgi_headers.c b/plugins/python/wsgi_headers.c index fc6cba07..8b2cfba3 100644 --- a/plugins/python/wsgi_headers.c +++ b/plugins/python/wsgi_headers.c @@ -176,7 +176,7 @@ PyObject *py_uwsgi_spit(PyObject * self, PyObject * args) { uh.pktsize += wsgi_req->hvec[j].iov_len; UWSGI_RELEASE_GIL - wsgi_req->headers_size = wsgi_req->socket_proto_writev_header(wsgi_req, wsgi_req->hvec, j + 1); + wsgi_req->headers_size = wsgi_req->socket->proto_writev_header(wsgi_req, wsgi_req->hvec, j + 1); UWSGI_GET_GIL if (wsgi_req->headers_size < 0) { uwsgi_error("writev()"); diff --git a/plugins/python/wsgi_subhandler.c b/plugins/python/wsgi_subhandler.c index 2ae6d258..c2007b32 100644 --- a/plugins/python/wsgi_subhandler.c +++ b/plugins/python/wsgi_subhandler.c @@ -124,7 +124,7 @@ int uwsgi_response_subhandler_wsgi(struct wsgi_request *wsgi_req) { // return or yield ? if (PyString_Check((PyObject *)wsgi_req->async_result)) { - if ((wsize = wsgi_req->socket_proto_write(wsgi_req, PyString_AsString(wsgi_req->async_result), PyString_Size(wsgi_req->async_result))) < 0) { + if ((wsize = wsgi_req->socket->proto_write(wsgi_req, PyString_AsString(wsgi_req->async_result), PyString_Size(wsgi_req->async_result))) < 0) { uwsgi_error("write()"); goto clear; } @@ -185,7 +185,7 @@ int uwsgi_response_subhandler_wsgi(struct wsgi_request *wsgi_req) { if (PyString_Check(pychunk)) { - if ((wsize = wsgi_req->socket_proto_write(wsgi_req, PyString_AsString(pychunk), PyString_Size(pychunk))) < 0) { + if ((wsize = wsgi_req->socket->proto_write(wsgi_req, PyString_AsString(pychunk), PyString_Size(pychunk))) < 0) { uwsgi_error("write()"); Py_DECREF(pychunk); goto clear; diff --git a/plugins/rpc/rpc_plugin.c b/plugins/rpc/rpc_plugin.c index 0424dc25..fc920339 100644 --- a/plugins/rpc/rpc_plugin.c +++ b/plugins/rpc/rpc_plugin.c @@ -29,10 +29,10 @@ int uwsgi_rpc_request(struct wsgi_request *wsgi_req) { wsgi_req->uh.pktsize = uwsgi_rpc(argv[0], argc-1, argv+1, wsgi_req->buffer); if (wsgi_req->uh.modifier2 == 0) { - wsgi_req->headers_size = wsgi_req->socket_proto_write_header(wsgi_req, (char *)&wsgi_req->uh, 4); + wsgi_req->headers_size = wsgi_req->socket->proto_write_header(wsgi_req, (char *)&wsgi_req->uh, 4); } - wsgi_req->response_size = wsgi_req->socket_proto_write(wsgi_req, wsgi_req->buffer, wsgi_req->uh.pktsize); + wsgi_req->response_size = wsgi_req->socket->proto_write(wsgi_req, wsgi_req->buffer, wsgi_req->uh.pktsize); wsgi_req->status = 0; return 0; diff --git a/proto/zeromq.c b/proto/zeromq.c index 53899919..6311bdd3 100644 --- a/proto/zeromq.c +++ b/proto/zeromq.c @@ -71,7 +71,7 @@ int uwsgi_proto_zeromq_accept(struct wsgi_request *wsgi_req, int fd) { wsgi_req->do_not_add_to_async_queue = 1; wsgi_req->proto_parser_status = 0; zmq_msg_init(&message); - if (zmq_recv(uwsgi.zmq_pull, &message, 0) < 0) { + if (zmq_recv(uwsgi.zmq_pull, &message, uwsgi.zeromq_recv_flag) < 0) { if (errno == EAGAIN) { uwsgi.edge_triggered = 0; } @@ -83,7 +83,7 @@ int uwsgi_proto_zeromq_accept(struct wsgi_request *wsgi_req, int fd) { } uwsgi.edge_triggered = 1; wsgi_req->proto_parser_pos = zmq_msg_size(&message); - uwsgi_log("%.*s\n", (int) wsgi_req->proto_parser_pos, zmq_msg_data(&message)); + //uwsgi_log("%.*s\n", (int) wsgi_req->proto_parser_pos, zmq_msg_data(&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); @@ -91,7 +91,6 @@ int uwsgi_proto_zeromq_accept(struct wsgi_request *wsgi_req, int fd) { } wsgi_req->proto_parser_buf = uwsgi_malloc(wsgi_req->proto_parser_pos); - uwsgi_log("proto_parser_pos %d\n", (int) wsgi_req->proto_parser_pos); ptr = zmq_msg_data(&message); for(i=0;i<(int)wsgi_req->proto_parser_pos;i++) { if (ptr[i] == ' ') { @@ -129,8 +128,6 @@ int uwsgi_proto_zeromq_accept(struct wsgi_request *wsgi_req, int fd) { } - uwsgi_log("STATUS: %.*s %.*s\n", req_uuid_len, req_uuid, req_body_size_len, req_body_size); - if (wsgi_req->proto_parser_status >= 4) { // ok ready to parse json data and build uwsgi request uwsgi_log("JSON: %s\n", wsgi_req->proto_parser_buf); @@ -139,8 +136,24 @@ int uwsgi_proto_zeromq_accept(struct wsgi_request *wsgi_req, int fd) { uwsgi_log("error parsing JSON data: line %d %s\n", error.line, error.text); zmq_msg_close(&message); free(wsgi_req->proto_parser_buf); + wsgi_req->proto_parser_pos = 0; + wsgi_req->do_not_log = 1; return -1; } + + 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")) { + json_decref(root); + zmq_msg_close(&message); + free(wsgi_req->proto_parser_buf); + wsgi_req->proto_parser_pos = 0; + wsgi_req->do_not_log = 1; + 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, "VERSION"); if (json_is_string(json_value)) { @@ -155,18 +168,6 @@ int uwsgi_proto_zeromq_accept(struct wsgi_request *wsgi_req, int fd) { 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")) { - uwsgi_log("refcnt: %d\n", root->refcount); - json_decref(root); - 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)) { @@ -218,7 +219,6 @@ int uwsgi_proto_zeromq_accept(struct wsgi_request *wsgi_req, int fd) { json_iter = json_object_iter_next(root, json_iter); } - uwsgi_log("refcnt: %d\n", root->refcount); json_decref(root); memcpy(wsgi_req->proto_parser_buf, req_uuid, req_uuid_len); @@ -234,36 +234,30 @@ int uwsgi_proto_zeromq_accept(struct wsgi_request *wsgi_req, int fd) { } - // 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; - } - } - */ - return 0; } return -1; } +static void uwsgi_proto_zeromq_free(void *data, void *hint) { + free(data); +} + 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); + // check for already freed wsgi_req->proto_parser_buf/wsgi_req->proto_parser_pos + if (!wsgi_req->proto_parser_pos) return; + + zmq_msg_init_data(&reply, wsgi_req->proto_parser_buf, wsgi_req->proto_parser_pos, uwsgi_proto_zeromq_free, NULL); if (zmq_send(uwsgi.zmq_pub, &reply, 0)) { uwsgi_error("zmq_send()"); } zmq_msg_close(&reply); - free(wsgi_req->proto_parser_buf); } @@ -294,18 +288,17 @@ ssize_t uwsgi_proto_zeromq_write(struct wsgi_request *wsgi_req, char *buf, size_ if (len == 0) return 0; - zmq_body = uwsgi_concat2n(wsgi_req->proto_parser_buf, (int) wsgi_req->proto_parser_pos, buf, len); + zmq_body = uwsgi_concat2n(wsgi_req->proto_parser_buf, (int) wsgi_req->proto_parser_pos, buf, (int) 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); + zmq_msg_init_data(&reply, zmq_body, wsgi_req->proto_parser_pos+len, uwsgi_proto_zeromq_free, 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; } diff --git a/sendfile.c b/sendfile.c index 2315b9dd..fdbef7b5 100644 --- a/sendfile.c +++ b/sendfile.c @@ -26,8 +26,8 @@ ssize_t uwsgi_sendfile(struct wsgi_request *wsgi_req) { if (!wsgi_req->sendfile_fd_chunk) wsgi_req->sendfile_fd_chunk = 4096; - if (wsgi_req->socket_proto_sendfile) { - sst = wsgi_req->socket_proto_sendfile(wsgi_req); + if (wsgi_req->socket->proto_sendfile) { + sst = wsgi_req->socket->proto_sendfile(wsgi_req); } else { sst = uwsgi_do_sendfile(sockfd, wsgi_req->sendfile_fd, wsgi_req->sendfile_fd_size, wsgi_req->sendfile_fd_chunk, &wsgi_req->sendfile_fd_pos, uwsgi.async); diff --git a/utils.c b/utils.c index 52ff4192..06bb98f8 100644 --- a/utils.c +++ b/utils.c @@ -415,7 +415,7 @@ void uwsgi_close_request(struct wsgi_request *wsgi_req) { // close the connection with the webserver 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); + wsgi_req->socket->proto_close(wsgi_req); } uwsgi.workers[0].requests++; uwsgi.workers[uwsgi.mywid].requests++; @@ -484,14 +484,7 @@ void wsgi_req_setup(struct wsgi_request *wsgi_req, int async_id, int socket_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; + wsgi_req->socket = &uwsgi.sockets[socket_id]; } } @@ -524,9 +517,8 @@ int wsgi_req_recv(struct wsgi_request *wsgi_req) { gettimeofday(&wsgi_req->start_of_request, NULL); - - if (!uwsgi.edge_triggered) { - if (!uwsgi_parse_response(&wsgi_req->poll, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT], (struct uwsgi_header *) wsgi_req, wsgi_req->buffer, wsgi_req->socket_proto)) { + if (!wsgi_req->socket->edge_trigger) { + if (!uwsgi_parse_response(&wsgi_req->poll, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT], (struct uwsgi_header *) wsgi_req, wsgi_req->buffer, wsgi_req->socket->proto)) { return -1; } } @@ -544,13 +536,13 @@ 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 = wsgi_req->socket_proto_accept(wsgi_req, fd); + wsgi_req->poll.fd = wsgi_req->socket->proto_accept(wsgi_req, fd); if (wsgi_req->poll.fd < 0) { return -1; } - if (uwsgi.close_on_exec) { + if (wsgi_req->socket->edge_trigger && uwsgi.close_on_exec) { fcntl(wsgi_req->poll.fd, F_SETFD, FD_CLOEXEC); } @@ -563,8 +555,8 @@ 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;isocket_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->socket = &uwsgi.sockets[socket_id]; - wsgi_req->poll.fd = wsgi_req->socket_proto_accept(wsgi_req, uwsgi.sockets_poll[i].fd); + wsgi_req->poll.fd = wsgi_req->socket->proto_accept(wsgi_req, uwsgi.sockets_poll[i].fd); if (wsgi_req->poll.fd < 0) { return -1; if (uwsgi.sockets[i].edge_trigger) { return -1 ;} - if (errno == EWOULDBLOCK) { uwsgi_log("GOTO polling\n"); goto polling;} + if (errno == EWOULDBLOCK) { goto polling;} uwsgi_error("accept()"); return -1; } + if (!uwsgi.sockets[socket_id].edge_trigger) { // in Linux, new sockets do not inherit attributes #ifndef __linux__ - /* re-set blocking socket */ - int arg = uwsgi.sockets[i].arg ; - arg &= (~O_NONBLOCK); - if (fcntl(wsgi_req->poll.fd, F_SETFL, arg) < 0) { - uwsgi_error("fcntl()"); - return -1; - } + /* re-set blocking socket */ + int arg = uwsgi.sockets[i].arg ; + arg &= (~O_NONBLOCK); + if (fcntl(wsgi_req->poll.fd, F_SETFL, arg) < 0) { + uwsgi_error("fcntl()"); + return -1; + } #endif - if (uwsgi.close_on_exec) { - fcntl(wsgi_req->poll.fd, F_SETFD, FD_CLOEXEC); - } + if (uwsgi.close_on_exec) { + fcntl(wsgi_req->poll.fd, F_SETFD, FD_CLOEXEC); + } - // set socket protocol + } return 0; } diff --git a/uwsgi.c b/uwsgi.c index 0341a618..7606d004 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -2239,6 +2239,7 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100 uuid_generate(uuid_zmq); uuid_unparse(uuid_zmq, uuid_zmq_str); + uwsgi_log("%.*s\n", 36, uuid_zmq_str); if (zmq_setsockopt(uwsgi.zmq_pub, ZMQ_IDENTITY, uuid_zmq_str, 36) < 0) { uwsgi_error("zmq_setsockopt()"); exit(1); @@ -2274,6 +2275,8 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100 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; + + uwsgi.zeromq_recv_flag = ZMQ_NOBLOCK; } #endif @@ -2383,11 +2386,11 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100 uwsgi_log("done\n"); goto end; } else { - if (uwsgi.zeromq) { + if (uwsgi.zeromq && uwsgi.cores < 2 && uwsgi.sockets_cnt == 1) { long y = 0; zeromq_loop((void *) y); } - if (uwsgi.threads > 1) { + else if (uwsgi.threads > 1) { pthread_attr_t pa; pthread_t *a_thread; int ret; @@ -2412,8 +2415,7 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100 pthread_create(a_thread, &pa, simple_loop, (void *) j); } } - - if (uwsgi.async < 2) { + else if (uwsgi.async < 2) { long y = 0; simple_loop((void *) y); } else { diff --git a/uwsgi.h b/uwsgi.h index 86b1c0df..b24c91df 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -723,6 +723,8 @@ struct wsgi_request { uint16_t var_cnt; uint16_t header_cnt; + int do_not_log; + int do_not_add_to_async_queue; int status; @@ -757,14 +759,8 @@ struct wsgi_request { char *post_buffering_buf; 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); - ssize_t (*socket_proto_writev_header)(struct wsgi_request *, struct iovec *, size_t); - ssize_t (*socket_proto_sendfile)(struct wsgi_request *); - void (*socket_proto_close)(struct wsgi_request *); + // current socket mapped to request + struct uwsgi_socket *socket; int body_as_file; //for generic use @@ -1069,6 +1065,7 @@ struct uwsgi_server { void *zmq_context; void *zmq_pull; void *zmq_pub; + int zeromq_recv_flag; #endif struct uwsgi_socket sockets[MAX_SOCKETS]; // leave a slot for no-orphan mode