From 0288313524105cadebd516de28b21aaf54fe03f9 Mon Sep 17 00:00:00 2001 From: "roberto@maverick64" Date: Mon, 7 Feb 2011 10:28:11 +0100 Subject: [PATCH] updated async stack --- async.c | 449 +++++++-------------------------- buildconf/default.ini | 2 +- loop.c | 34 ++- plugins/python/python_plugin.c | 3 +- tests/iobound_async.py | 4 + uwsgi.c | 16 +- uwsgi.h | 46 +--- 7 files changed, 150 insertions(+), 404 deletions(-) diff --git a/async.c b/async.c index 4bb35b0c..c2bce951 100644 --- a/async.c +++ b/async.c @@ -1,361 +1,103 @@ -#ifdef UWSGI_ASYNC - #include "uwsgi.h" extern struct uwsgi_server uwsgi; - -#ifdef __linux__ - -#include - -int async_queue_init(int serverfd) { - int epfd; - struct epoll_event ee; - - epfd = epoll_create(256); - - if (epfd < 0) { - uwsgi_error("epoll_create()"); - return -1; - } - - memset(&ee, 0, sizeof(struct epoll_event)); - ee.events = EPOLLIN; - ee.data.fd = serverfd; - - if (epoll_ctl(epfd, EPOLL_CTL_ADD, serverfd, &ee)) { - uwsgi_error("epoll_ctl()"); - close(epfd); - return -1; - } - - return epfd; -} - -int async_wait(int queuefd, void *events, int nevents, int block, int timeout) { - - int ret; - - if (timeout <= 0) { - timeout = block; - } - else { - timeout = timeout*1000; - } - - //uwsgi_log("waiting with timeout %d nevents %d\n", timeout, nevents); - ret = epoll_wait(queuefd, (struct epoll_event *) events, nevents, timeout); - if (ret < 0) { - uwsgi_error("epoll_wait()"); - } - return ret; -} - -int async_add(int queuefd, int fd, int etype) { - struct epoll_event ee; - - memset(&ee, 0, sizeof(struct epoll_event)); - ee.events = etype; - ee.data.fd = fd; - - if (epoll_ctl(queuefd, EPOLL_CTL_ADD, fd, &ee)) { - uwsgi_error("epoll_ctl()"); - return -1; - } - - return 0; -} - -int async_mod(int queuefd, int fd, int etype) { - struct epoll_event ee; - - memset(&ee, 0, sizeof(struct epoll_event)); - ee.events = etype; - ee.data.fd = fd; - - if (epoll_ctl(queuefd, EPOLL_CTL_MOD, fd, &ee)) { - uwsgi_error("epoll_ctl()"); - return -1; - } - - return 0; -} - -int async_del(int queuefd, int fd, int etype) { - struct epoll_event ee; - - memset(&ee, 0, sizeof(struct epoll_event)); - ee.events = etype; - ee.data.fd = fd; - - if (epoll_ctl(queuefd, EPOLL_CTL_DEL, fd, &ee)) { - uwsgi_error("epoll_ctl()"); - return -1; - } - - return 0; -} - -#elif defined(__sun__) - -int async_queue_init(int serverfd) { - int dpfd; - struct pollfd dpev; - - - dpfd = open("/dev/poll", O_RDWR); - - if (dpfd < 0) { - uwsgi_error("open()"); - return -1; - } - - - dpev.fd = serverfd; - dpev.events = POLLIN; - dpev.revents = 0; - - if (write(dpfd, &dpev, sizeof(struct pollfd)) < 0) { - uwsgi_error("write()"); - return -1; - } - - - return dpfd; -} - -int async_wait(int queuefd, void *events, int nevents, int block, int timeout) { - - int ret; - struct dvpoll dv; - - if (timeout <= 0) { - timeout = block; - } - else { - timeout = timeout*1000; - } - - dv.dp_fds = (struct pollfd *) events; - dv.dp_nfds = nevents; - dv.dp_timeout = timeout; - - //uwsgi_log("waiting with timeout %d nevents %d\n", timeout, nevents); - ret = ioctl(queuefd, DP_POLL, &dv); - if (ret < 0) { - uwsgi_error("ioctl()"); - } - return ret; -} - -int async_add(int queuefd, int fd, int etype) { - struct pollfd pl; - - pl.fd = fd; - pl.events = etype; - pl.revents = 0; - - if (write(queuefd, &pl, sizeof(struct pollfd)) < 0) { - uwsgi_error("write()"); - return -1; - } - - return 0; -} - -int async_mod(int queuefd, int fd, int etype) { - // using the same fd will overwrite existing rule - return async_add(queuefd, fd, etype); -} - -int async_del(int queuefd, int fd, int etype) { - // use POLLREMOVE to remove an fd - return async_add(queuefd, fd, POLLREMOVE); -} - -#else -int async_queue_init(int serverfd) { - - int eqfd = event_queue_init(); - if (eqfd < 0) { - exit(1); - } - - if (event_queue_add_fd_read(eqfd, serverfd) < 0) { - exit(1); - } - - return eqfd ; -} - -int async_wait(int queuefd, void *events, int nevents, int block, int timeout) { - - int ret; - struct timespec ts; - - - if (timeout <= 0) { - if (!block) { - memset(&ts, 0, sizeof(struct timespec)); - ts.tv_sec = timeout; - ret = kevent(queuefd, NULL, 0, events, nevents, &ts); - } - else { - ret = kevent(queuefd, NULL, 0, events, nevents, NULL); - } - } - else { - memset(&ts, 0, sizeof(struct timespec)); - ts.tv_sec = timeout; - ret = kevent(queuefd, NULL, 0, events, nevents, &ts); - } - - if (ret < 0) { - uwsgi_error("kevent()"); - } - - return ret; -} - -int async_add(int queuefd, int fd, int etype) { - struct kevent kev; - - - EV_SET(&kev, fd, etype, EV_ADD, 0, 0, 0); - if (kevent(queuefd, &kev, 1, NULL, 0, NULL) < 0) { - uwsgi_error("kevent()"); - return -1; - } - return 0; -} - -int async_mod(int queuefd, int fd, int etype) { - struct kevent kev; - - EV_SET(&kev, fd, ASYNC_OUT, EV_DISABLE, 0, 0, 0); - if (kevent(queuefd, &kev, 1, NULL, 0, NULL) < 0) { - uwsgi_error("kevent()"); - return -1; - } - - EV_SET(&kev, fd, etype, EV_ADD, 0, 0, 0); - if (kevent(queuefd, &kev, 1, NULL, 0, NULL) < 0) { - uwsgi_error("kevent()"); - return -1; - } - return 0; -} - -int async_del(int queuefd, int fd, int etype) { - struct kevent kev; - - EV_SET(&kev, fd, etype, EV_DELETE, 0, 0, 0); - if (kevent(queuefd, &kev, 1, NULL, 0, NULL) < 0) { - uwsgi_error("kevent()"); - return -1; - } - - return 0; -} - -#endif - int async_get_timeout() { - - struct wsgi_request* wsgi_req; - int i; - time_t curtime, tdelta = 0; - int ret = 0; + struct wsgi_request* wsgi_req ; + int i ; + time_t curtime, tdelta = 0 ; + int ret = 0 ; if (!uwsgi.async_running) return 0; - for(i=0;iasync_status == UWSGI_AGAIN) { + if (wsgi_req->async_status == UWSGI_AGAIN) { if (wsgi_req->async_timeout_expired) { return 0; } if (wsgi_req->async_timeout > 0) { if (tdelta <= 0 || tdelta > wsgi_req->async_timeout) { - tdelta = wsgi_req->async_timeout; + tdelta = wsgi_req->async_timeout ; } } - } - } + } + } curtime = time(NULL); - ret = tdelta - curtime; + ret = tdelta - curtime ; if (ret > 0) { return ret; } - + return 0; } void async_expire_timeouts() { - struct wsgi_request* wsgi_req; - int i; + struct wsgi_request* wsgi_req ; + int i ; time_t deadline = time(NULL); - for(i=0;iasync_status == UWSGI_AGAIN && wsgi_req->async_timeout > 0) { + for(i=0;iasync_status == UWSGI_AGAIN && wsgi_req->async_timeout > 0) { if (wsgi_req->async_timeout <= deadline) { - wsgi_req->async_timeout = 0; - wsgi_req->async_timeout_expired = 1; + wsgi_req->async_timeout = 0 ; + wsgi_req->async_timeout_expired = 1 ; if (wsgi_req->async_waiting_fd != -1) { - async_del(uwsgi.async_queue, wsgi_req->async_waiting_fd, wsgi_req->async_waiting_fd_type); - wsgi_req->async_waiting_fd = -1; + event_queue_del_fd(uwsgi.async_queue, wsgi_req->async_waiting_fd); + uwsgi.async_waiting_fd_table[wsgi_req->async_waiting_fd] = -1; + wsgi_req->async_waiting_fd = -1; wsgi_req->async_waiting_fd_monitored = 0; - } - } - } - } + } + } + } + } } struct wsgi_request *find_first_available_wsgi_req() { - struct wsgi_request* wsgi_req = uwsgi.wsgi_requests[0]; - int i; + struct wsgi_request* wsgi_req; + int i ; - for(i=0;iasync_status == UWSGI_OK) { - return wsgi_req; + // optimization + if (uwsgi.async_current_max > 1) { + if (uwsgi.wsgi_requests[uwsgi.async_current_max-1]->async_status == UWSGI_OK) { + uwsgi_log("decreasing current max cores\n"); + uwsgi.async_current_max--; } - wsgi_req = uwsgi.wsgi_requests[i+1]; } - return NULL; + for(i=0;iasync_status == UWSGI_OK) { + // optimization + if (i > uwsgi.async_current_max-1) uwsgi.async_current_max = i+1; + wsgi_req->async_id = i; + return wsgi_req ; + } + } + + return NULL ; } -struct wsgi_request *find_wsgi_req_by_fd(int fd, int etype) { +struct wsgi_request *find_wsgi_req_by_fd(int fd) { - struct wsgi_request* wsgi_req = uwsgi.wsgi_requests[0]; - int i; + struct wsgi_request* wsgi_req = NULL ; + int core_id = uwsgi.async_waiting_fd_table[fd]; - if (etype != -1) { - for(i=0;iasync_waiting_fd == fd && wsgi_req->async_waiting_fd_type == etype) { - return wsgi_req; - } - wsgi_req = uwsgi.wsgi_requests[i+1]; - } - } - else { - for(i=0;iasync_waiting_fd == fd) { - return wsgi_req; - } - wsgi_req = uwsgi.wsgi_requests[i+1]; - } - } + if (core_id == -1) return NULL; - return NULL; + wsgi_req = uwsgi.wsgi_requests[core_id]; + //if (wsgi_req->async_waiting_fd_type == etype) return wsgi_req ; + return wsgi_req; + + return NULL ; } @@ -363,96 +105,89 @@ void async_set_timeout(struct wsgi_request *wsgi_req, time_t timeout) { wsgi_req->async_timeout = time(NULL); wsgi_req->async_timeout += timeout; - wsgi_req->async_timeout_expired = 0; - + wsgi_req->async_timeout_expired = 0 ; + } void async_write_all(char *data, size_t len) { - - struct wsgi_request *wsgi_req = uwsgi.wsgi_requests[0]; + + struct wsgi_request *wsgi_req; int i; - ssize_t rlen; + ssize_t rlen ; - for(i=0;iasync_status == UWSGI_PAUSED) { + for(i=0;iasync_status == UWSGI_PAUSED) { rlen = write(wsgi_req->poll.fd, data, len); if (rlen < 0) { uwsgi_error("write()"); } else { - wsgi_req->response_size += rlen; + wsgi_req->response_size += rlen ; } } - wsgi_req = uwsgi.wsgi_requests[i+1]; } } void async_unpause_all() { - - struct wsgi_request *wsgi_req = uwsgi.wsgi_requests[0]; + + struct wsgi_request *wsgi_req ; int i; - for(i=0;iasync_status == UWSGI_PAUSED) { + for(i=0;iasync_status == UWSGI_PAUSED) { wsgi_req->async_status = UWSGI_AGAIN; } - wsgi_req = uwsgi.wsgi_requests[i+1]; } } -struct wsgi_request *find_first_accepting_wsgi_req() { - - struct wsgi_request* wsgi_req; - int i; - - for(i=0;iasync_status == UWSGI_ACCEPTING) { - return wsgi_req; - } - } - - return NULL; -} - - struct wsgi_request * async_loop() { - struct wsgi_request *wsgi_req; - int i; + struct wsgi_request *wsgi_req ; + int i ; + int ret; - uwsgi.async_running = -1; - wsgi_req = uwsgi.wsgi_requests[0]; + uwsgi.async_running = -1 ; - - for(i=0;iasync_status == UWSGI_AGAIN) { - if (wsgi_req->async_waiting_fd != -1 && !wsgi_req->async_waiting_fd_monitored) { + for(i=0;iasync_status == UWSGI_AGAIN) { + if (wsgi_req->async_waiting_fd != -1 && !wsgi_req->async_waiting_fd_monitored) { // add fd to monitoring - if (async_add(uwsgi.async_queue, wsgi_req->async_waiting_fd, wsgi_req->async_waiting_fd_type)) { + ret = -1; + if (wsgi_req->async_waiting_fd_type == ASYNC_IN) { + ret = event_queue_add_fd_read(uwsgi.async_queue, wsgi_req->async_waiting_fd); + } + else if (wsgi_req->async_waiting_fd_type == ASYNC_OUT) { + ret = event_queue_add_fd_write(uwsgi.async_queue, wsgi_req->async_waiting_fd); + } + + if (ret < 0) { // error adding fd to the async queue, better to close it... close(wsgi_req->async_waiting_fd); - wsgi_req->async_status = UWSGI_OK; + wsgi_req->async_status = UWSGI_OK ; return wsgi_req; } + uwsgi.async_waiting_fd_table[wsgi_req->async_waiting_fd] = wsgi_req->async_id; wsgi_req->async_waiting_fd_monitored = 1; wsgi_req->async_status = UWSGI_AGAIN; } else if (wsgi_req->async_waiting_fd == -1 && wsgi_req->async_timeout <= 0) { - uwsgi.async_running = 0; - // st global wsgi_req for python functions - uwsgi.wsgi_req = wsgi_req; - wsgi_req->async_status = uwsgi.p[wsgi_req->uh.modifier1]->request(wsgi_req); + uwsgi.async_running = 0 ; + // st global wsgi_req + uwsgi.wsgi_req = wsgi_req ; + wsgi_req->async_status = uwsgi.p[wsgi_req->uh.modifier1]->request(wsgi_req);; + + wsgi_req->switches++; if (wsgi_req->async_status < UWSGI_AGAIN) { return wsgi_req; } } } - wsgi_req = uwsgi.wsgi_requests[i+1]; } return NULL; } -#endif diff --git a/buildconf/default.ini b/buildconf/default.ini index 9e1a6dde..e4f6e5ba 100644 --- a/buildconf/default.ini +++ b/buildconf/default.ini @@ -23,7 +23,7 @@ xml_implementation = libxml2 plugins = bin_name = uwsgi plugin_dir = . -embedded_plugins = python, ping, proxy, nagios, rpc, fastrouter +embedded_plugins = python, ping, nagios, rpc, fastrouter locking = auto event = auto diff --git a/loop.c b/loop.c index ab1c6bef..adc679df 100644 --- a/loop.c +++ b/loop.c @@ -92,11 +92,14 @@ void *simple_loop(void *arg1) { void complex_loop() { int current_async_timeout = 0; int i; + int interesting_fd; while (uwsgi.workers[uwsgi.mywid].manage_next_request) { current_async_timeout = async_get_timeout(); - uwsgi.async_nevents = async_wait(uwsgi.async_queue, uwsgi.async_events, uwsgi.async, uwsgi.async_running, current_async_timeout); + + current_async_timeout = 0; + uwsgi.async_nevents = event_queue_wait_multi(uwsgi.async_queue, current_async_timeout, uwsgi.async_events, 64); async_expire_timeouts(); if (uwsgi.async_nevents < 0) { @@ -107,7 +110,9 @@ void complex_loop() { for(i=0; iasync_id ); - if (wsgi_req_accept(uwsgi.wsgi_req)) { - continue; - } + if (wsgi_req_simple_accept(uwsgi.wsgi_req, interesting_fd)) { + continue; + } + +// on linux we do not need to reset the socket to blocking state +#ifndef __linux__ + if (uwsgi.numproc > 1) { + /* re-set blocking socket */ + if (fcntl(uwsgi.wsgi_req->poll.fd, F_SETFL, uwsgi.fcntl_arg) < 0) { + uwsgi_error("fcntl()"); + return -1; + } + } +#endif if (wsgi_req_recv(uwsgi.wsgi_req)) { continue; @@ -130,7 +146,7 @@ void complex_loop() { } } - else if ( (int) uwsgi.async_events[i].ASYNC_FD == uwsgi.sockets[uwsgi.sockets_cnt].fd) { + else if ( interesting_fd == uwsgi.sockets[uwsgi.sockets_cnt].fd) { // wake up cores waiting for signal char byte; if (read(uwsgi.sockets[uwsgi.sockets_cnt].fd, &byte, 1) == 1) { @@ -138,7 +154,7 @@ void complex_loop() { } } else { - uwsgi.wsgi_req = find_wsgi_req_by_fd(uwsgi.async_events[i].ASYNC_FD, uwsgi.async_events[i].ASYNC_EV); + uwsgi.wsgi_req = find_wsgi_req_by_fd(interesting_fd); if (uwsgi.wsgi_req) { uwsgi.wsgi_req->async_status = UWSGI_AGAIN; uwsgi.wsgi_req->async_waiting_fd = -1; @@ -147,7 +163,7 @@ void complex_loop() { uwsgi.wsgi_req->async_timeout = 0; } - async_del(uwsgi.async_queue, uwsgi.async_events[i].ASYNC_FD, uwsgi.async_events[i].ASYNC_EV); + event_queue_del_fd(uwsgi.async_queue, interesting_fd); } } diff --git a/plugins/python/python_plugin.c b/plugins/python/python_plugin.c index df854066..e8539e97 100644 --- a/plugins/python/python_plugin.c +++ b/plugins/python/python_plugin.c @@ -51,7 +51,8 @@ PyMethodDef uwsgi_write_method[] = { {"uwsgi_write", py_uwsgi_write, METH_VARARG int uwsgi_python_init() { - uwsgi_log("Python version: %s\n", Py_GetVersion()); + char *pyversion = strchr(Py_GetVersion(), '\n'); + uwsgi_log("Python version: %.*s %s\n", pyversion-Py_GetVersion(), Py_GetVersion(), Py_GetCompiler()+1); if (up.home != NULL) { uwsgi_log("Setting PythonHome to %s...\n", up.home); diff --git a/tests/iobound_async.py b/tests/iobound_async.py index e4b0af56..e41690f5 100644 --- a/tests/iobound_async.py +++ b/tests/iobound_async.py @@ -28,9 +28,13 @@ def application(env, start_response): #yield "" + print "opening socket" + c = s.connect_ex(('www.google.it', 80)) if c == errno.EINPROGRESS: + print "yielding" yield env['x-wsgiorg.fdevent.writable'](s.fileno(), 10) + print "waiting for fd write" for r in send_request(env, s): yield r elif c == errno.EISCONN: diff --git a/uwsgi.c b/uwsgi.c index 9a06f393..f751f2af 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -1053,6 +1053,14 @@ int uwsgi_start(void *v_argv) { uwsgi.async_buf = uwsgi_malloc(sizeof(char *) * uwsgi.cores); + if (uwsgi.async > 1) { + uwsgi.async_waiting_fd_table = malloc( sizeof(int) * uwsgi.max_fd); + if (!uwsgi.async_waiting_fd_table) { + uwsgi_error("malloc()"); + exit(1); + } + } + if (uwsgi.post_buffering > 0) { uwsgi.async_post_buf = uwsgi_malloc(sizeof(char *) * uwsgi.cores); if (!uwsgi.post_buffering_bufsize) { @@ -1650,10 +1658,14 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100 //do not pass kfd after fork() #ifdef UWSGI_ASYNC if (uwsgi.async > 1) { - uwsgi.async_queue = async_queue_init(uwsgi.sockets[0].fd); + uwsgi.async_queue = event_queue_init(); if (uwsgi.async_queue < 0) { exit(1); } + + for(i=0;ihooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100 uwsgi.sockets_poll[uwsgi.sockets_cnt].events = POLLIN; #ifdef UWSGI_ASYNC if (uwsgi.async > 1) { - async_add(uwsgi.async_queue, uwsgi.sockets_poll[uwsgi.sockets_cnt].fd, ASYNC_IN); + event_queue_add_fd_read(uwsgi.async_queue, uwsgi.sockets_poll[uwsgi.sockets_cnt].fd); } #endif } diff --git a/uwsgi.h b/uwsgi.h index 4e5bcbe5..9a88d257 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -727,6 +727,9 @@ struct uwsgi_server { char **async_buf; char **async_post_buf; + int *async_waiting_fd_table; + int async_current_max; + #ifdef UWSGI_ROUTING int **async_ovector; #endif @@ -1178,10 +1181,13 @@ struct http_status_codes { }; #ifdef UWSGI_ASYNC + +#define ASYNC_IN 1 +#define ASYNC_OUT 2 struct wsgi_request *async_loop(void); struct wsgi_request *find_first_available_wsgi_req(void); struct wsgi_request *find_first_accepting_wsgi_req(void); -struct wsgi_request *find_wsgi_req_by_fd(int, int); +struct wsgi_request *find_wsgi_req_by_fd(int); struct wsgi_request *find_wsgi_req_by_id(int); #ifdef __clang__ @@ -1191,40 +1197,12 @@ inline struct wsgi_request *next_wsgi_req(struct wsgi_request *); #endif -int async_add(int, int, int); -int async_mod(int, int, int); -int async_wait(int, void *, int, int, int); -int async_del(int, int, int); -int async_queue_init(int); +int async_get_timeout(void); +void async_set_timeout(struct wsgi_request*, time_t); +void async_expire_timeouts(void); +void async_write_all(char *, size_t); +void async_unpause_all(void); -int async_get_timeout(void); -void async_set_timeout(struct wsgi_request *, time_t); -void async_expire_timeouts(void); -void async_write_all(char *, size_t); -void async_unpause_all(void); - -#ifdef __linux__ -#define ASYNC_FD data.fd -#define ASYNC_EV events -#define ASYNC_IN EPOLLIN -#define ASYNC_OUT EPOLLOUT -#define ASYNC_IS_IN ASYNC_EV & ASYNC_IN -#define ASYNC_IS_OUT ASYNC_EV & ASYNC_OUT -#elif defined(__sun__) -#define ASYNC_FD fd -#define ASYNC_EV revents -#define ASYNC_IN POLLIN -#define ASYNC_OUT POLLOUT -#define ASYNC_IS_IN ASYNC_EV & ASYNC_IN -#define ASYNC_IS_OUT ASYNC_EV & ASYNC_OUT -#else -#define ASYNC_FD ident -#define ASYNC_EV filter -#define ASYNC_IN EVFILT_READ -#define ASYNC_OUT EVFILT_WRITE -#define ASYNC_IS_IN ASYNC_EV == ASYNC_IN -#define ASYNC_IS_OUT ASYNC_EV == ASYNC_OUT -#endif #endif