diff --git a/core/async.c b/core/async.c index 729d19b2..f7374e4d 100644 --- a/core/async.c +++ b/core/async.c @@ -386,6 +386,12 @@ void async_loop() { event_queue_del_fd(uwsgi.async_queue, interesting_fd, event_queue_read()); // put request in the runqueue runqueue_push(uwsgi.wsgi_req); +#ifdef UWSGI_ROUTING + if (uwsgi_apply_routes(uwsgi.wsgi_req) == UWSGI_ROUTE_BREAK) { + uwsgi.async_proto_fd_table[interesting_fd] = NULL; + close(interesting_fd); + } +#endif continue; } else if (proto_parser_status < 0) { diff --git a/core/io.c b/core/io.c index 4f40bbc8..93843c68 100644 --- a/core/io.c +++ b/core/io.c @@ -703,6 +703,8 @@ written: } + + // like uwsgi_pipe but with fixed size ssize_t uwsgi_pipe_sized(int src, int dst, size_t required, int timeout) { char buf[8192]; @@ -820,6 +822,35 @@ int uwsgi_read_nb(int fd, char *buf, size_t remains, int timeout) { return 0; } +/* + this is like uwsgi_read_nb() but with fast initial read and hooked wait (use it in request plugin) +*/ +ssize_t uwsgi_read_true_nb(int fd, char *buf, size_t len, int timeout) { + int ret; + + ssize_t rlen = read(fd, buf, len); + if (rlen > 0) { + return rlen; + } + if (rlen == 0) return -1; + if (rlen < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS) goto wait; + } + return -1; +wait: + ret = uwsgi.wait_read_hook(fd, timeout); + if (ret > 0) { + rlen = read(fd, buf, len); + if (rlen > 0) { + return rlen; + } + return -1; + } + return ret; +} + + + /* this is a pretty magic function used for read a full uwsgi response it is true non blocking, so you can use it in request plugins @@ -895,3 +926,64 @@ readok2: return 0; } + +/* + + this is a commodity (big) function to send a buffer and wsgi_req body to a socket + and to receive back data (and send them to the client) + +*/ + +int uwsgi_proxy_nb(struct wsgi_request *wsgi_req, char *addr, struct uwsgi_buffer *ub, size_t remains, int timeout) { + + int fd = uwsgi_connect(addr, 0, 1); + if (fd < 0) { + return -1; + } + + int ret = uwsgi.wait_write_hook(fd, timeout); + if (ret <= 0) { + goto end; + } + + // send the request (+ remaining data) + if (ub) { + if (uwsgi_write_true_nb(fd, ub->buf, ub->pos, timeout)) { + goto end; + } + } + + // send the body + while(remains > 0) { + ssize_t rlen = 0; + char *buf = uwsgi_request_body_read(wsgi_req, 8192, &rlen); + if (!buf) { + goto end; + } + if (buf == uwsgi.empty) break; + // write data to the node + if (uwsgi_write_true_nb(fd, buf, rlen, timeout)) { + goto end; + } + remains -= rlen; + } + + // read the response + for(;;) { + char buf[8192]; + ssize_t rlen = uwsgi_read_true_nb(fd, buf, 8192, timeout); + if (rlen > 0) { + if (uwsgi_response_write_body_do(wsgi_req, buf, rlen)) { + break; + } + continue; + } + break; + } + + close(fd); + return 0; +end: + close(fd); + return -1; +} diff --git a/core/reader.c b/core/reader.c index ba01e639..6dd1a241 100644 --- a/core/reader.c +++ b/core/reader.c @@ -3,7 +3,25 @@ extern struct uwsgi_server uwsgi; int uwsgi_simple_wait_read_hook(int fd, int timeout) { - return uwsgi_waitfd(fd, timeout); + struct pollfd upoll; + timeout = timeout * 1000; + + upoll.fd = fd; + upoll.events = POLLIN; + upoll.revents = 0; + int ret = poll(&upoll, 1, timeout); + + if (ret > 0) { + if (upoll.revents & POLLIN) { + return 1; + } + return -1; + } + if (ret < 0) { + uwsgi_error("uwsgi_simple_wait_read_hook()/poll()"); + } + + return ret; } /* diff --git a/core/utils.c b/core/utils.c index 82720745..70458e3b 100644 --- a/core/utils.c +++ b/core/utils.c @@ -890,8 +890,6 @@ int wsgi_req_async_recv(struct wsgi_request *wsgi_req) { uwsgi.async_proto_fd_table[wsgi_req->fd] = wsgi_req; } - - // enter harakiri mode if (uwsgi.shared->options[UWSGI_OPTION_HARAKIRI] > 0) { set_harakiri(uwsgi.shared->options[UWSGI_OPTION_HARAKIRI]); diff --git a/core/writer.c b/core/writer.c index 0324bd46..c003380c 100644 --- a/core/writer.c +++ b/core/writer.c @@ -277,5 +277,23 @@ sendfile: int uwsgi_simple_wait_write_hook(int fd, int timeout) { - return uwsgi_waitfd_write(fd, timeout); + struct pollfd upoll; + timeout = timeout * 1000; + + upoll.fd = fd; + upoll.events = POLLOUT; + upoll.revents = 0; + int ret = poll(&upoll, 1, timeout); + + if (ret > 0) { + if (upoll.revents & POLLOUT) { + return 1; + } + return -1; + } + if (ret < 0) { + uwsgi_error("uwsgi_simple_wait_write_hook()/poll()"); + } + + return ret; } diff --git a/plugins/gevent/gevent.c b/plugins/gevent/gevent.c index c468525a..602d947f 100644 --- a/plugins/gevent/gevent.c +++ b/plugins/gevent/gevent.c @@ -226,6 +226,12 @@ PyObject *py_uwsgi_gevent_request(PyObject * self, PyObject * args) { request: +#ifdef UWSGI_ROUTING + if (uwsgi_apply_routes(wsgi_req) == UWSGI_ROUTE_BREAK) { + goto end; + } +#endif + for(;;) { wsgi_req->async_status = uwsgi.p[wsgi_req->uh->modifier1]->request(wsgi_req); if (wsgi_req->async_status <= UWSGI_OK) { diff --git a/plugins/router_http/router_http.c b/plugins/router_http/router_http.c index c3521718..0719cb16 100644 --- a/plugins/router_http/router_http.c +++ b/plugins/router_http/router_http.c @@ -30,6 +30,17 @@ int uwsgi_routing_func_http(struct wsgi_request *wsgi_req, struct uwsgi_route *u if (uri) free(uri); + // amount of body to send + size_t remains = wsgi_req->post_cl - wsgi_req->proto_parser_remains; + // append remaining body... + if (wsgi_req->proto_parser_remains > 0) { + if (uwsgi_buffer_append(ub, wsgi_req->proto_parser_remains_buf, wsgi_req->proto_parser_remains)) { + uwsgi_log("unable to generate http request for %s\n", addr); + return UWSGI_ROUTE_NEXT; + } + wsgi_req->proto_parser_remains = 0; + } + // ok now if have offload threads, directly use them if (wsgi_req->socket->can_offload) { if (!uwsgi_offload_request_net_do(wsgi_req, addr, ub)) { @@ -38,49 +49,10 @@ int uwsgi_routing_func_http(struct wsgi_request *wsgi_req, struct uwsgi_route *u } } - // 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); - free(ub); - 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->fd; - if (wsgi_req->post_file) { - post_fd = fileno((FILE *)wsgi_req->post_file); - } - 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->fd, 0); - if (ret > 0) { - wsgi_req->response_size += ret; - } - else { + if (uwsgi_proxy_nb(wsgi_req, addr, ub, remains, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT])) { uwsgi_log("error routing request to http server %s\n", addr); } - close(http_fd); uwsgi_buffer_destroy(ub); return UWSGI_ROUTE_BREAK; diff --git a/plugins/ugreen/ugreen.c b/plugins/ugreen/ugreen.c index 3850d4cc..fcdb9242 100644 --- a/plugins/ugreen/ugreen.c +++ b/plugins/ugreen/ugreen.c @@ -64,7 +64,9 @@ static void u_green_schedule_to_main(struct wsgi_request *wsgi_req) { uwsgi.p[wsgi_req->uh->modifier1]->suspend(wsgi_req); } + uwsgi_log("pippo %d %p %p\n", wsgi_req->async_id, &ug.contexts[wsgi_req->async_id], &ug.main); swapcontext(&ug.contexts[wsgi_req->async_id], &ug.main); + uwsgi_log("pluto\n"); if (uwsgi.p[wsgi_req->uh->modifier1]->resume) { uwsgi.p[wsgi_req->uh->modifier1]->resume(wsgi_req); diff --git a/uwsgi.h b/uwsgi.h index 08845246..43afd708 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -2943,7 +2943,9 @@ void uwsgi_socket_nb(int); void uwsgi_socket_b(int); int uwsgi_write_nb(int, char *, size_t, int); int uwsgi_read_nb(int, char *, size_t, int); +ssize_t uwsgi_read_true_nb(int, char *, size_t, int); int uwsgi_read_uh(int fd, struct uwsgi_header *, int); +int uwsgi_proxy_nb(struct wsgi_request *, char *, struct uwsgi_buffer *, size_t, int); int uwsgi_read_with_realloc(int, char **, size_t *, int); int uwsgi_write_true_nb(int, char *, size_t, int);