From 0ba1b982d126af0e5c73211d5ee3065a78d48ccf Mon Sep 17 00:00:00 2001 From: "roberto@maverick64" Date: Sun, 3 Jul 2011 12:43:30 +0200 Subject: [PATCH] implemented target='workers' for signal infrastructure --- decoratortest.py | 7 +++++++ loop.c | 29 ++++++++++++++++++++++++++-- master_utils.c | 15 +++++++++++++-- plugins/python/uwsgi_pymodule.c | 34 +++++++++++++++++++++++++-------- signal.c | 9 ++++++--- sigwait.py | 10 +++++----- utils.c | 2 +- uwsgi.c | 9 +++++++++ uwsgi.h | 3 +++ 9 files changed, 97 insertions(+), 21 deletions(-) diff --git a/decoratortest.py b/decoratortest.py index 7f7398f4..4c9c39e4 100644 --- a/decoratortest.py +++ b/decoratortest.py @@ -17,6 +17,11 @@ def hello_world(): def what_time_is_it(num): print(time.asctime()) +# register signal 100 +@signal(100, target='workers') +def what_time_is_it(num): + print("*** I AM THE WORKER %d AT %s ***" % (uwsgi.worker_id(), time.asctime())) + # a 3 seconds timer @timer(3) @@ -53,6 +58,8 @@ def an_infinite_task(args): @spool def delayed_task(args): print("*** I am a delayed spool job. It is %s [%s]***" % (time.asctime(), str(args))) + # send a signal to all workers + uwsgi.signal(100) # run a task every hour @cron(59, -1, -1, -1, -1) diff --git a/loop.c b/loop.c index 4293a451..67520918 100644 --- a/loop.c +++ b/loop.c @@ -72,6 +72,7 @@ void *simple_loop(void *arg1) { if (uwsgi.signal_socket > -1) { event_queue_add_fd_read(main_queue, uwsgi.signal_socket); + event_queue_add_fd_read(main_queue, uwsgi.my_signal_socket); } while (uwsgi.workers[uwsgi.mywid].manage_next_request) { @@ -115,7 +116,7 @@ void *zeromq_loop(void *arg1) { struct wsgi_request *wsgi_req = uwsgi.wsgi_requests[core_id]; uwsgi.zeromq_recv_flag = 0; - zmq_pollitem_t zmq_poll_items[2]; + zmq_pollitem_t zmq_poll_items[3]; char uwsgi_signal; if (uwsgi.threads > 1) { @@ -159,6 +160,10 @@ void *zeromq_loop(void *arg1) { zmq_poll_items[1].socket = NULL; zmq_poll_items[1].fd = uwsgi.signal_socket; zmq_poll_items[1].events = ZMQ_POLLIN; + + zmq_poll_items[2].socket = NULL; + zmq_poll_items[2].fd = uwsgi.my_signal_socket; + zmq_poll_items[2].events = ZMQ_POLLIN; } @@ -171,7 +176,7 @@ void *zeromq_loop(void *arg1) { if (uwsgi.signal_socket > -1) { - if (zmq_poll(zmq_poll_items, 2, -1) < 0) { + if (zmq_poll(zmq_poll_items, 3, -1) < 0) { uwsgi_error("zmq_poll()"); continue; } @@ -194,6 +199,26 @@ void *zeromq_loop(void *arg1) { continue; } + if (zmq_poll_items[2].revents & ZMQ_POLLIN) { + if (read(uwsgi.my_signal_socket, &uwsgi_signal, 1) <= 0) { + if (uwsgi.no_orphans) { + uwsgi_log_verbose("uWSGI worker %d screams: UAAAAAAH my master died, i will follow him...\n", uwsgi.mywid); + end_me(0); + } + } + else { +#ifdef UWSGI_DEBUG + uwsgi_log_verbose("master sent signal %d to worker %d\n", uwsgi_signal, uwsgi.mywid); +#endif + if (uwsgi_signal_handler(uwsgi_signal)) { + uwsgi_log_verbose("error managing signal %d on worker %d\n", uwsgi_signal, uwsgi.mywid); + } + } + continue; + } + + + if (zmq_poll_items[0].revents & ZMQ_POLLIN) { wsgi_req->poll.fd = wsgi_req->socket->proto_accept(wsgi_req, uwsgi.zmq_socket->fd); } diff --git a/master_utils.c b/master_utils.c index 8e9affa8..9b39a9f2 100644 --- a/master_utils.c +++ b/master_utils.c @@ -37,12 +37,19 @@ int uwsgi_respawn_worker(int wid) { int respawns = uwsgi.workers[wid].respawn_count; int i; + if (uwsgi.master_process) { + if (uwsgi.workers[wid].signal_pipe[0] != -1) close(uwsgi.workers[wid].signal_pipe[0]); + + if (socketpair(AF_UNIX, SOCK_STREAM, 0, uwsgi.workers[wid].signal_pipe)) { + uwsgi_error("socketpair()\n"); + } + } + + pid_t pid = fork(); if (pid == 0) { uwsgi.mywid = wid; - // fix the communication pipe - close(uwsgi.shared->worker_signal_pipe[0]); uwsgi.mypid = getpid(); uwsgi.workers[uwsgi.mywid].pid = uwsgi.mypid; uwsgi.workers[uwsgi.mywid].id = uwsgi.mywid; @@ -54,6 +61,10 @@ int uwsgi_respawn_worker(int wid) { uwsgi.workers[uwsgi.mywid].manage_next_request = 1; if (uwsgi.master_process) { + // fix the communication pipe + close(uwsgi.shared->worker_signal_pipe[0]); + close(uwsgi.workers[wid].signal_pipe[0]); + uwsgi.my_signal_socket = uwsgi.workers[wid].signal_pipe[1]; if (uwsgi.shared->spooler_signal_pipe[0] != -1) close (uwsgi.shared->spooler_signal_pipe[0]); if ((uwsgi.workers[uwsgi.mywid].respawn_count || uwsgi.cheap)) { for (i = 0; i < 0xFF; i++) { diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index 744a96f3..4121baa9 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -14,6 +14,7 @@ PyObject *py_uwsgi_signal_wait(PyObject * self, PyObject * args) { uint8_t uwsgi_signal = 0; uint8_t received_signal; int ret; + struct pollfd pfd[2]; wsgi_req->signal_received = -1; @@ -30,18 +31,35 @@ PyObject *py_uwsgi_signal_wait(PyObject * self, PyObject * args) { } else { #endif + pfd[0].fd = uwsgi.signal_socket; + pfd[0].events = POLLIN; + pfd[1].fd = uwsgi.my_signal_socket; + pfd[1].events = POLLIN; cycle: - ret = uwsgi_waitfd(uwsgi.signal_socket, -1); + ret = poll(pfd, 2, -1); if (ret > 0) { - if (read(uwsgi.signal_socket, &received_signal, 1) != 1) { - uwsgi_error("read()"); - } - else { - wsgi_req->signal_received = received_signal; - if (wait_for_specific_signal) { - if (received_signal != uwsgi_signal) goto cycle; + if (pfd[0].revents == POLLIN) { + if (read(uwsgi.signal_socket, &received_signal, 1) != 1) { + uwsgi_error("read()"); + } + else { + wsgi_req->signal_received = received_signal; + if (wait_for_specific_signal) { + if (received_signal != uwsgi_signal) goto cycle; + } } } + if (pfd[1].revents == POLLIN) { + if (read(uwsgi.my_signal_socket, &received_signal, 1) != 1) { + uwsgi_error("read()"); + } + else { + wsgi_req->signal_received = received_signal; + if (wait_for_specific_signal) { + if (received_signal != uwsgi_signal) goto cycle; + } + } + } } #ifdef UWSGI_ASYNC diff --git a/signal.c b/signal.c index 60b48613..57624883 100644 --- a/signal.c +++ b/signal.c @@ -154,6 +154,7 @@ int uwsgi_signal_add_rb_timer(uint8_t sig, int secs, int iterations) { void uwsgi_route_signal(uint8_t sig) { + int i; struct uwsgi_signal_entry *use = &ushared->signal_table[sig]; // send to first available worker if (use->receiver[0] == 0 || !strcmp(use->receiver, "worker") || !strcmp(use->receiver, "worker0")) { @@ -164,9 +165,11 @@ void uwsgi_route_signal(uint8_t sig) { } // send to all workers else if (!strcmp(use->receiver, "workers")) { - 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); + for(i=1;i<=uwsgi.numproc;i++) { + if (write(uwsgi.workers[i].signal_pipe[0], &sig, 1) != 1) { + uwsgi_error("write()"); + uwsgi_log("could not deliver signal %d to worker %d\n", sig, i); + } } } // route to subscribed diff --git a/sigwait.py b/sigwait.py index f5a8cefe..940b1022 100644 --- a/sigwait.py +++ b/sigwait.py @@ -1,16 +1,16 @@ import uwsgi from uwsgidecorators import * +@signal(17, target='workers') +def hello(signum): + print("I AM THE WORKER %d" % uwsgi.worker_id()) + @postfork def wait_for_signal(): - if uwsgi.worker_id() == 2: + if uwsgi.worker_id() != 2: print("waiting for a signal...") uwsgi.signal_wait() print("signal %d received" % uwsgi.signal_received()) - elif uwsgi.worker_id() == 3: - print("waiting for signal 30...") - uwsgi.signal_wait(30) - print("signal %d received" % uwsgi.signal_received()) def application(e, s): diff --git a/utils.c b/utils.c index 2c2109b3..9677e194 100644 --- a/utils.c +++ b/utils.c @@ -778,7 +778,7 @@ int wsgi_req_accept(int queue, struct wsgi_request *wsgi_req) { if (uwsgi.threads > 1) pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &ret); #endif - if (uwsgi.signal_socket > -1 && interesting_fd == uwsgi.signal_socket) { + if (uwsgi.signal_socket > -1 && (interesting_fd == uwsgi.signal_socket || interesting_fd == uwsgi.my_signal_socket)) { if (read(interesting_fd, &uwsgi_signal, 1) <= 0) { if (uwsgi.no_orphans) { uwsgi_log_verbose("uWSGI worker %d screams: UAAAAAAH my master died, i will follow him...\n", uwsgi.mywid); diff --git a/uwsgi.c b/uwsgi.c index 33bb83ce..3eb85f18 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -812,6 +812,7 @@ int main(int argc, char *argv[], char *envp[]) { uwsgi.master_queue = -1; uwsgi.signal_socket = -1; + uwsgi.my_signal_socket = -1; uwsgi.emperor_fd_config = -1; uwsgi.emperor_pid = -1; @@ -1940,6 +1941,13 @@ int uwsgi_start(void *v_argv) { uwsgi.workers[0].pid = masterpid; + // fix the second signal socket + if (uwsgi.master_process) { + for(i=1;i<=uwsgi.numproc;i++) { + uwsgi.workers[i].signal_pipe[0] = -1; + } + } + /* uwsgi.shared->hooks[0] = uwsgi_request_wsgi; @@ -2188,6 +2196,7 @@ int uwsgi_start(void *v_argv) { #endif } + if (uwsgi.worker_exec) { char *w_argv[2]; w_argv[0] = uwsgi.worker_exec; diff --git a/uwsgi.h b/uwsgi.h index 55136bb4..a35e8b0a 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -1154,6 +1154,7 @@ struct uwsgi_server { char *protocol; int signal_socket; + int my_signal_socket; #ifdef UWSGI_ZEROMQ char *zeromq; @@ -1459,6 +1460,8 @@ struct uwsgi_worker { int destroy; + int signal_pipe[2]; + }; char *uwsgi_get_cwd(void);