diff --git a/plugins/gevent/gevent.c b/plugins/gevent/gevent.c index d215541c..ec622a93 100644 --- a/plugins/gevent/gevent.c +++ b/plugins/gevent/gevent.c @@ -10,6 +10,10 @@ struct option gevent_options[] = { #define GEVENT_SWITCH PyObject *gswitch = python_call(ugevent.greenlet_switch, ugevent.greenlet_switch_args, 0, NULL); Py_DECREF(gswitch) #define GET_CURRENT_GREENLET python_call(ugevent.get_current, ugevent.get_current_args, 0, NULL) #define free_req_queue uwsgi.async_queue_unused_ptr++; uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr] = uwsgi.wsgi_req +#define stop_the_watchers ret = PyObject_CallMethod(timer, "stop", NULL);\ + if (ret) Py_DECREF(ret);\ + ret = PyObject_CallMethod(watcher, "stop", NULL);\ + if (ret) Py_DECREF(ret); struct uwsgi_gevent { PyObject *greenlet_switch; @@ -20,11 +24,68 @@ struct uwsgi_gevent { PyObject *hub_loop; PyObject *spawn; PyObject *greenlet_args; + PyObject *signal_args; } ugevent; +PyObject *py_uwsgi_gevent_signal_handler(PyObject * self, PyObject * args) { -PyObject *py_uwsgi_gevent_callback(PyObject * self, PyObject * args) { + uint8_t uwsgi_signal; + int signal_socket; + + if (!PyArg_ParseTuple(args, "i:uwsgi_gevent_signal_handler", &signal_socket)) { + return NULL; + } + + if (read(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); + } + // close the socket to end the mess...from now on the worker is alone (no master) + else close(signal_socket); + } + 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); + } + } + + Py_INCREF(Py_None); + return Py_None; +} + +PyObject *py_uwsgi_gevent_signal(PyObject * self, PyObject * args) { + + PyTuple_SetItem(ugevent.signal_args, 1, PyInt_FromLong(uwsgi.signal_socket)); + + // spawn the signal_handler greenlet + PyObject *new_gl = python_call(ugevent.spawn, ugevent.signal_args, 0, NULL); + Py_DECREF(new_gl); + + Py_INCREF(Py_None); + return Py_None; + +} + +// yes copy&paste no-DRY for me :P +PyObject *py_uwsgi_gevent_my_signal(PyObject * self, PyObject * args) { + + PyTuple_SetItem(ugevent.signal_args, 1, PyInt_FromLong(uwsgi.my_signal_socket)); + + // spawn the signal_handler greenlet + PyObject *new_gl = python_call(ugevent.spawn, ugevent.signal_args, 0, NULL); + Py_DECREF(new_gl); + + Py_INCREF(Py_None); + return Py_None; +} + + +PyObject *py_uwsgi_gevent_main(PyObject * self, PyObject * args) { struct wsgi_request *wsgi_req = find_first_available_wsgi_req(); @@ -35,8 +96,10 @@ PyObject *py_uwsgi_gevent_callback(PyObject * self, PyObject * args) { } uwsgi.wsgi_req = wsgi_req; + // fill wsgi_request structure wsgi_req_setup(wsgi_req, wsgi_req->async_id, uwsgi.sockets ); + // mark core as used uwsgi.core[wsgi_req->async_id]->in_request = 1; gettimeofday(&wsgi_req->start_of_request, NULL); @@ -46,15 +109,17 @@ PyObject *py_uwsgi_gevent_callback(PyObject * self, PyObject * args) { set_harakiri(uwsgi.shared->options[UWSGI_OPTION_HARAKIRI]); } + // accept the connection if (wsgi_req_simple_accept(wsgi_req, uwsgi.sockets->fd)) { uwsgi_close_request(wsgi_req); free_req_queue; goto clear; } - + // hack to easily pass wsgi_req pointer to the greenlet PyTuple_SetItem(ugevent.greenlet_args, 1, PyLong_FromLong((long)wsgi_req)); + // spawn the request greenlet PyObject *new_gl = python_call(ugevent.spawn, ugevent.greenlet_args, 0, NULL); Py_DECREF(new_gl); @@ -81,7 +146,7 @@ PyObject *uwsgi_gevent_wait(PyObject *watcher, PyObject *timer, PyObject *curren return PyObject_CallMethod(ugevent.hub, "switch", NULL); } -PyObject *py_uwsgi_gevent_greenlet(PyObject * self, PyObject * args) { +PyObject *py_uwsgi_gevent_request(PyObject * self, PyObject * args) { PyObject *ret; PyObject *py_wsgi_req = PyTuple_GetItem(args, 0); @@ -93,20 +158,20 @@ PyObject *py_uwsgi_gevent_greenlet(PyObject * self, PyObject * args) { uwsgi.wsgi_req = wsgi_req; + // create a watcher for request socket PyObject *watcher = PyObject_CallMethod(ugevent.hub_loop, "io", "ii", wsgi_req->poll.fd, 1); - if (!watcher) { - goto clear1; - } + if (!watcher) goto clear1; + // a timer to implement timeoit (thanks Denis) PyObject *timer = PyObject_CallMethod(ugevent.hub_loop, "timer", "i", uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); - if (!timer) { - goto clear0; - } + if (!timer) goto clear0; for(;;) { + // wait for data in the socket PyObject *ret = uwsgi_gevent_wait(watcher, timer, greenlet_switch); if (!ret) goto clear_and_stop; + // do not forget to overwrite this pointer each time !!! uwsgi.wsgi_req = wsgi_req; // we can safely decref here as watcher and timer has got a +1 for start() method @@ -122,11 +187,7 @@ PyObject *py_uwsgi_gevent_greenlet(PyObject * self, PyObject * args) { goto clear_and_stop; } else if (status == 0) { - ret = PyObject_CallMethod(timer, "stop", NULL); - if (ret) Py_DECREF(ret); - - ret = PyObject_CallMethod(watcher, "stop", NULL); - if (ret) Py_DECREF(ret); + stop_the_watchers; break; } } @@ -135,12 +196,7 @@ PyObject *py_uwsgi_gevent_greenlet(PyObject * self, PyObject * args) { goto clear_and_stop; } - ret = PyObject_CallMethod(timer, "stop", NULL); - if (ret) Py_DECREF(ret); - - ret = PyObject_CallMethod(watcher, "stop", NULL); - if (ret) Py_DECREF(ret); - + stop_the_watchers; } for(;;) { @@ -149,6 +205,7 @@ PyObject *py_uwsgi_gevent_greenlet(PyObject * self, PyObject * args) { if (wsgi_req->async_status <= UWSGI_OK) { goto clear; } + // switch after each yield GEVENT_SWITCH; } @@ -156,11 +213,7 @@ PyObject *py_uwsgi_gevent_greenlet(PyObject * self, PyObject * args) { clear_and_stop: - ret = PyObject_CallMethod(timer, "stop", NULL); - if (ret) Py_DECREF(ret); - - ret = PyObject_CallMethod(watcher, "stop", NULL); - if (ret) Py_DECREF(ret); + stop_the_watchers; clear: Py_DECREF(timer); @@ -181,8 +234,11 @@ clear1: } -PyMethodDef uwsgi_gevent_callback_method[] = { {"uwsgi_gevent_callback", py_uwsgi_gevent_callback, METH_VARARGS, ""} }; -PyMethodDef uwsgi_gevent_greenlet_method[] = { {"uwsgi_gevent_greenlet", py_uwsgi_gevent_greenlet, METH_VARARGS, ""} }; +PyMethodDef uwsgi_gevent_main_def[] = { {"uwsgi_gevent_main", py_uwsgi_gevent_main, METH_VARARGS, ""} }; +PyMethodDef uwsgi_gevent_request_def[] = { {"uwsgi_gevent_request", py_uwsgi_gevent_request, METH_VARARGS, ""} }; +PyMethodDef uwsgi_gevent_signal_def[] = { {"uwsgi_gevent_signal", py_uwsgi_gevent_signal, METH_VARARGS, ""} }; +PyMethodDef uwsgi_gevent_my_signal_def[] = { {"uwsgi_gevent_my_signal", py_uwsgi_gevent_my_signal, METH_VARARGS, ""} }; +PyMethodDef uwsgi_gevent_signal_handler_def[] = { {"uwsgi_gevent_signal_handler", py_uwsgi_gevent_signal_handler, METH_VARARGS, ""} }; void gevent_loop() { @@ -196,16 +252,10 @@ void gevent_loop() { PyObject *gevent_dict = get_uwsgi_pydict("gevent"); - if (!gevent_dict) { - PyErr_Print(); - exit(1); - } + if (!gevent_dict) uwsgi_pyexit; PyObject *gevent_version = PyDict_GetItemString(gevent_dict, "version_info"); - if (!gevent_version) { - PyErr_Print(); - exit(1); - } + if (!gevent_version) uwsgi_pyexit; if (PyInt_AsLong(PyTuple_GetItem(gevent_version, 0)) < 1) { uwsgi_log("uWSGI requires at least gevent 1.x version\n"); @@ -213,16 +263,10 @@ void gevent_loop() { } ugevent.spawn = PyDict_GetItemString(gevent_dict, "spawn"); - if (!ugevent.spawn) { - PyErr_Print(); - exit(1); - } + if (!ugevent.spawn) uwsgi_pyexit; ugevent.greenlet_switch = PyDict_GetItemString(gevent_dict, "sleep"); - if (!ugevent.greenlet_switch) { - PyErr_Print(); - exit(1); - } + if (!ugevent.greenlet_switch) uwsgi_pyexit; ugevent.greenlet_switch_args = PyTuple_New(0); Py_INCREF(ugevent.greenlet_switch_args); @@ -231,47 +275,63 @@ void gevent_loop() { PyObject *gevent_get_hub = PyDict_GetItemString(gevent_dict, "get_hub"); ugevent.hub = python_call(gevent_get_hub, PyTuple_New(0), 0, NULL); - if (!ugevent.hub) { - PyErr_Print(); - exit(1); - } + if (!ugevent.hub) uwsgi_pyexit; ugevent.get_current = PyDict_GetItemString(gevent_dict, "getcurrent"); - if (!ugevent.get_current) { - PyErr_Print(); - exit(1); - } + if (!ugevent.get_current) uwsgi_pyexit; + ugevent.get_current_args = PyTuple_New(0); Py_INCREF(ugevent.get_current_args); ugevent.hub_loop = PyObject_GetAttrString(ugevent.hub, "loop"); - if (!ugevent.hub_loop) { - - PyErr_Print(); - exit(1); - } - + if (!ugevent.hub_loop) uwsgi_pyexit; + // this is the watcher for server socket PyObject *watcher = PyObject_CallMethod(ugevent.hub_loop, "io", "ii", uwsgi_sock->fd, 1); - if (!watcher) { - PyErr_Print(); - exit(1); - } + if (!watcher) uwsgi_pyexit; + // main greenlet waiting for connection + PyObject *uwsgi_gevent_main = PyCFunction_New(uwsgi_gevent_main_def, NULL); + Py_INCREF(uwsgi_gevent_main); - PyObject *uwsgi_gevent_callback = PyCFunction_New(uwsgi_gevent_callback_method, NULL); - Py_INCREF(uwsgi_gevent_callback); - - PyObject *uwsgi_gevent_greenlet = PyCFunction_New(uwsgi_gevent_greenlet_method, NULL); - Py_INCREF(uwsgi_gevent_greenlet); + // greenlet to run at each request + PyObject *uwsgi_request_greenlet = PyCFunction_New(uwsgi_gevent_request_def, NULL); + Py_INCREF(uwsgi_request_greenlet); + // pre-fill the greenlet args ugevent.greenlet_args = PyTuple_New(2); - PyTuple_SetItem(ugevent.greenlet_args, 0, uwsgi_gevent_greenlet); + PyTuple_SetItem(ugevent.greenlet_args, 0, uwsgi_request_greenlet); + if (uwsgi.signal_socket > -1) { + // and these are the watcher for signal sockets + PyObject *signal_watcher = PyObject_CallMethod(ugevent.hub_loop, "io", "ii", uwsgi.signal_socket, 1); + if (!signal_watcher) uwsgi_pyexit; - PyObject_CallMethod(watcher, "start", "O", uwsgi_gevent_callback); + PyObject *my_signal_watcher = PyObject_CallMethod(ugevent.hub_loop, "io", "ii", uwsgi.my_signal_socket, 1); + if (!my_signal_watcher) uwsgi_pyexit; + + PyObject *uwsgi_greenlet_signal = PyCFunction_New(uwsgi_gevent_signal_def, NULL); + Py_INCREF(uwsgi_greenlet_signal); + + PyObject *uwsgi_greenlet_my_signal = PyCFunction_New(uwsgi_gevent_my_signal_def, NULL); + Py_INCREF(uwsgi_greenlet_my_signal); + + PyObject *uwsgi_greenlet_signal_handler = PyCFunction_New(uwsgi_gevent_signal_handler_def, NULL); + Py_INCREF(uwsgi_greenlet_signal_handler); + + ugevent.signal_args = PyTuple_New(2); + PyTuple_SetItem(ugevent.signal_args, 0, uwsgi_greenlet_signal_handler); + + // start the two signal watchers + if (!PyObject_CallMethod(signal_watcher, "start", "O", uwsgi_greenlet_signal)) uwsgi_pyexit; + if (!PyObject_CallMethod(my_signal_watcher, "start", "O", uwsgi_greenlet_my_signal)) uwsgi_pyexit; + + } + + // start the main greenlet + PyObject_CallMethod(watcher, "start", "O", uwsgi_gevent_main); if (!PyObject_CallMethod(ugevent.hub, "join", NULL)) { PyErr_Print(); diff --git a/plugins/python/python_plugin.c b/plugins/python/python_plugin.c index 1c939728..3dd2ca51 100644 --- a/plugins/python/python_plugin.c +++ b/plugins/python/python_plugin.c @@ -1213,6 +1213,7 @@ int uwsgi_python_signal_handler(uint8_t sig, void *handler) { ret = python_call(handler, args, 0, NULL); Py_DECREF(args); if (ret) { + Py_DECREF(ret); UWSGI_RELEASE_GIL; return 0; } diff --git a/plugins/python/uwsgi_python.h b/plugins/python/uwsgi_python.h index 357a375a..b5236838 100644 --- a/plugins/python/uwsgi_python.h +++ b/plugins/python/uwsgi_python.h @@ -264,6 +264,8 @@ void uwsgi_python_reset_random_seed(void); char *uwsgi_pythonize(char *); +#define uwsgi_pyexit {PyErr_Print();exit(1);} + #ifdef __linux__ #ifndef PYTHREE int uwsgi_init_symbol_import(void); diff --git a/tests/ugevent.py b/tests/ugevent.py index cbbd3310..4e520e51 100644 --- a/tests/ugevent.py +++ b/tests/ugevent.py @@ -1,17 +1,39 @@ import gevent import gevent.socket import sys +import uwsgi +from uwsgidecorators import * if 'gettotalrefcount' in sys.__dict__: REFCNT = True else: REFCNT = False +@signal(17) +def hello(signum): + print "hello i am signal %d, i am here because the background job is finished" % signum + if REFCNT: + print sys.gettotalrefcount() + +@timer(10) +def ten_seconds(signum): + print "10 seconds elapsed, signal %d raised" % signum + if REFCNT: + print sys.gettotalrefcount() + +@filemon('/tmp') +def tmp_modified(signum): + print "/tmp has been touched, i am the greenlet %s running on worker %d" % (gevent.getcurrent(), uwsgi.worker_id()) + if REFCNT: + print sys.gettotalrefcount() def bg_task(): for i in range(1,10): print "background task", i - gevent.sleep(2) + gevent.sleep(1) + + # task ended raise a signal !!! + uwsgi.signal(17) def long_task(): for i in range(1,10): @@ -43,6 +65,7 @@ def application(e, sr): if REFCNT: print sys.gettotalrefcount() + yield "%d" % sys.gettotalrefcount() # this task will goes on after request end gevent.spawn(bg_task) diff --git a/uwsgi.c b/uwsgi.c index d09e067c..4a3388b5 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -2582,9 +2582,11 @@ void uwsgi_ignition() { uwsgi_log("unavailable loop engine !!!\n"); exit(1); } - uwsgi_log("running %s loop %p\n", uwsgi.loop, u_loop); + if (uwsgi.mywid == 1) { + uwsgi_log("*** running %s loop engine [addr:%p] ***\n", uwsgi.loop, u_loop); + } u_loop(); - uwsgi_log("done\n"); + uwsgi_log("your loop engine died. R.I.P.\n"); } else { #ifdef UWSGI_ZEROMQ