From bcff6e0173a717f85b8a6815e09cdeef5291afaf Mon Sep 17 00:00:00 2001 From: "roberto@goyle" Date: Wed, 16 Feb 2011 15:44:53 +0100 Subject: [PATCH] various fixes --- async.c | 12 ++++++++++-- event.c | 12 +++++++++--- loop.c | 29 +++++++++++++++++++++++------ plugins/python/uwsgi_pymodule.c | 25 +++++++++++++++++++++++++ plugins/python/wsgi_handlers.c | 1 + signal.c | 9 +++++++++ uwsgi.c | 8 ++------ uwsgi.h | 3 ++- 8 files changed, 81 insertions(+), 18 deletions(-) diff --git a/async.c b/async.c index a3d0cd32..ddf86e76 100644 --- a/async.c +++ b/async.c @@ -9,12 +9,14 @@ int async_get_timeout() { time_t curtime, tdelta = 0 ; int ret = 0 ; + // do not wait if there are cores running if (!uwsgi.async_running) return 0; for(i=0;iasync_status == UWSGI_AGAIN) { if (wsgi_req->async_timeout_expired) { + // do not wait if there are timeout expired return 0; } if (wsgi_req->async_timeout > 0) { @@ -32,7 +34,7 @@ int async_get_timeout() { return ret; } - return 0; + return -1; } void async_expire_timeouts() { @@ -153,7 +155,11 @@ struct wsgi_request * async_loop() { for(i=0;iasync_status == UWSGI_AGAIN) { - if (wsgi_req->async_waiting_fd != -1 && !wsgi_req->async_waiting_fd_monitored) { + if (wsgi_req->sigwait) { + uwsgi_log("waiting for signal\n"); + continue; + } + else if (wsgi_req->async_waiting_fd != -1 && !wsgi_req->async_waiting_fd_monitored) { // add fd to monitoring ret = -1; if (wsgi_req->async_waiting_fd_type == ASYNC_IN) { @@ -177,6 +183,8 @@ struct wsgi_request * async_loop() { uwsgi.async_running = 0 ; // st global wsgi_req uwsgi.wsgi_req = wsgi_req ; + + uwsgi_log("!!! getting new part\n"); wsgi_req->async_status = uwsgi.p[wsgi_req->uh.modifier1]->request(wsgi_req);; wsgi_req->switches++; diff --git a/event.c b/event.c index 1c817831..5e8b4c92 100644 --- a/event.c +++ b/event.c @@ -314,6 +314,12 @@ int event_queue_fd_write_to_read(int eq, int fd) { struct kevent kev; + EV_SET(&kev, fd, EVFILT_WRITE, EV_DISABLE, 0, 0, 0); + if (kevent(eq, &kev, 1, NULL, 0, NULL) < 0) { + uwsgi_error("kevent()"); + return -1; + } + EV_SET(&kev, fd, EVFILT_READ, EV_ADD, 0, 0, 0); if (kevent(eq, &kev, 1, NULL, 0, NULL) < 0) { uwsgi_error("kevent()"); @@ -327,7 +333,7 @@ int event_queue_del_fd(int eq, int fd) { struct kevent kev; - EV_SET(&kev, fd, EVFILT_WRITE, EV_DELETE, 0, 0, 0); + EV_SET(&kev, fd, EVFILT_READ, EV_DELETE, 0, 0, 0); if (kevent(eq, &kev, 1, NULL, 0, NULL) < 0) { uwsgi_error("kevent()"); return -1; @@ -372,7 +378,7 @@ int event_queue_wait_multi(int eq, int timeout, void *events, int nevents) { int ret; struct timespec ts; - if (timeout <= 0) { + if (timeout < 0) { ret = kevent(eq, NULL, 0, events, nevents, NULL); } else { @@ -398,7 +404,7 @@ int event_queue_interesting_fd(void *events, int id) { int event_queue_interesting_fd_has_error(void *events, int id) { struct kevent *ev = (struct kevent *) events; - if (ev[id].flags & EV_ERROR || ev[id].flags & EV_EOF) { + if (ev[id].flags == EV_ERROR || ev[id].flags == EV_EOF) { return 1; } return 0; diff --git a/loop.c b/loop.c index 7e6b851d..6c80ef4c 100644 --- a/loop.c +++ b/loop.c @@ -91,15 +91,18 @@ void *simple_loop(void *arg1) { #ifdef UWSGI_ASYNC void complex_loop() { int current_async_timeout = 0; - int i; + int i, j; int interesting_fd; + struct wsgi_request *wsgi_req; + while (uwsgi.workers[uwsgi.mywid].manage_next_request) { current_async_timeout = async_get_timeout(); - current_async_timeout = 0; + uwsgi_log("waiting multi %d\n", current_async_timeout); uwsgi.async_nevents = event_queue_wait_multi(uwsgi.async_queue, current_async_timeout, uwsgi.async_events, 64); + uwsgi_log("waiting done\n"); async_expire_timeouts(); if (uwsgi.async_nevents < 0) { @@ -112,6 +115,8 @@ void complex_loop() { interesting_fd = event_queue_interesting_fd(uwsgi.async_events, i); + uwsgi_log("interesting_fd: %d\n", interesting_fd); + if ( interesting_fd == uwsgi.sockets[0].fd) { uwsgi.wsgi_req = find_first_available_wsgi_req(); @@ -146,11 +151,22 @@ void complex_loop() { } } - else if ( interesting_fd == uwsgi.sockets[uwsgi.sockets_cnt].fd) { + else if ( interesting_fd == uwsgi.sockets_poll[uwsgi.sockets_cnt].fd) { // wake up cores waiting for signal char byte; - if (read(uwsgi.sockets[uwsgi.sockets_cnt].fd, &byte, 1) == 1) { - uwsgi_log("signal %c received\n", byte); + uwsgi_log("*** READING SIGNAL ***\n"); + if (read(uwsgi.sockets_poll[uwsgi.sockets_cnt].fd, &byte, 1) == 1) { + uwsgi_log("signal %d received\n", byte); + } + else { + uwsgi_error("read()"); + } + for(j=0;jsigwait) { + wsgi_req->signal_received = byte; + wsgi_req->sigwait = 0; + } } } else { @@ -158,7 +174,6 @@ void complex_loop() { if (uwsgi.wsgi_req) { uwsgi.wsgi_req->async_status = UWSGI_AGAIN; uwsgi.wsgi_req->async_waiting_fd = -1; - uwsgi.wsgi_req->async_waiting_signal = -1; uwsgi.wsgi_req->async_waiting_fd_monitored = 0; uwsgi.wsgi_req->async_timeout = 0; } @@ -168,6 +183,8 @@ void complex_loop() { } cycle: + + uwsgi_log("async_loop\n"); uwsgi.wsgi_req = async_loop(); if (uwsgi.wsgi_req == NULL) diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index 11354273..caeea170 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -7,6 +7,23 @@ char *spool_buffer = NULL; extern struct uwsgi_server uwsgi; extern struct uwsgi_python up; +PyObject *py_uwsgi_signal_wait(PyObject * self, PyObject * args) { + + struct wsgi_request *wsgi_req = current_wsgi_req(); + + wsgi_req->sigwait = 1; + + return PyString_FromString(""); +} + +PyObject *py_uwsgi_signal_received(PyObject * self, PyObject * args) { + + struct wsgi_request *wsgi_req = current_wsgi_req(); + + return PyInt_FromLong(wsgi_req->signal_received); +} + + char *uwsgi_encode_pydict(PyObject * pydict, uint16_t * size) { int i; @@ -2294,6 +2311,8 @@ static PyMethodDef uwsgi_advanced_methods[] = { {"register_signal", py_uwsgi_register_signal, METH_VARARGS, ""}, {"signal", py_uwsgi_signal, METH_VARARGS, ""}, + {"signal_wait", py_uwsgi_signal_wait, METH_VARARGS, ""}, + {"signal_received", py_uwsgi_signal_received, METH_VARARGS, ""}, {"register_file_monitor", py_uwsgi_register_file_monitor, METH_VARARGS, ""}, {"register_timer", py_uwsgi_register_timer, METH_VARARGS, ""}, @@ -2481,6 +2500,11 @@ PyObject *py_uwsgi_queue_push(PyObject * self, PyObject * args) { } +PyObject *py_uwsgi_queue_slot(PyObject * self, PyObject * args) { + + return PyInt_FromLong(uwsgi.shared->queue_pos); +} + PyObject *py_uwsgi_queue_pull(PyObject * self, PyObject * args) { char *message; @@ -2601,6 +2625,7 @@ static PyMethodDef uwsgi_queue_methods[] = { {"queue_get", py_uwsgi_queue_get, METH_VARARGS, ""}, {"queue_push", py_uwsgi_queue_push, METH_VARARGS, ""}, {"queue_pull", py_uwsgi_queue_pull, METH_VARARGS, ""}, + {"queue_slot", py_uwsgi_queue_slot, METH_VARARGS, ""}, {NULL, NULL}, }; diff --git a/plugins/python/wsgi_handlers.c b/plugins/python/wsgi_handlers.c index 4940967c..6c37c44f 100644 --- a/plugins/python/wsgi_handlers.c +++ b/plugins/python/wsgi_handlers.c @@ -185,6 +185,7 @@ PyObject *py_uwsgi_write(PyObject * self, PyObject * args) { #ifdef UWSGI_ASYNC + PyObject *py_eventfd_read(PyObject * self, PyObject * args) { int fd, timeout = 0; diff --git a/signal.c b/signal.c index 362d7f2e..3be8c279 100644 --- a/signal.c +++ b/signal.c @@ -114,4 +114,13 @@ void uwsgi_route_signal(uint8_t sig) { // route to subscribed else if (!strcmp(use->receiver, "subscribed")) { } + else { + // unregistered signal, sending it to all the workers + uwsgi_log("^^^ ROUTING UNREGISTERED SIGNAL ^^^\n"); + if (write(ushared->worker_signal_pipe[0], &sig, 1) != 1) { + uwsgi_error("write()"); + uwsgi_log("could not deliver signal %d to workers pool\n", sig); + } + + } } diff --git a/uwsgi.c b/uwsgi.c index 08e0deeb..c22bc0b1 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -970,7 +970,7 @@ int uwsgi_start(void *v_argv) { } if (!getrlimit(RLIMIT_NOFILE, &uwsgi.rl)) { - uwsgi.max_fd = uwsgi.rl.rlim_max; + uwsgi.max_fd = uwsgi.rl.rlim_cur; } uwsgi.wsgi_requests = uwsgi_malloc(sizeof(struct wsgi_request *) * uwsgi.cores); @@ -983,6 +983,7 @@ int uwsgi_start(void *v_argv) { uwsgi.async_buf = uwsgi_malloc(sizeof(char *) * uwsgi.cores); if (uwsgi.async > 1) { + uwsgi_log("%d\n", uwsgi.max_fd); uwsgi.async_waiting_fd_table = malloc( sizeof(int) * uwsgi.max_fd); if (!uwsgi.async_waiting_fd_table) { uwsgi_error("malloc()"); @@ -1680,11 +1681,6 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100 } } -#ifdef UWSGI_ASYNC - uwsgi.async_running = -1; -#endif - - //re - initialize wsgi_req(can be full of init_uwsgi_app data) for (i = 0; i < uwsgi.cores; i++) { diff --git a/uwsgi.h b/uwsgi.h index f3abee4a..dc290569 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -619,7 +619,6 @@ struct wsgi_request { int async_waiting_fd; int async_waiting_fd_type; int async_waiting_fd_monitored; - int async_waiting_signal; int switches; @@ -651,6 +650,8 @@ struct wsgi_request { int log_this; + int sigwait; + uint8_t signal_received; };