more improvements on the non-blocking api

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) {

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;
+}

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;
 }
 
 /*

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]);

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;
 }

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) {

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;

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);

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);