diff --git a/plugins/gevent/gevent.c b/plugins/gevent/gevent.c index b7b7e287..6847e86e 100644 --- a/plugins/gevent/gevent.c +++ b/plugins/gevent/gevent.c @@ -32,6 +32,9 @@ void uwsgi_opt_setup_gevent(char *opt, char *value, void *null) { // set async mode uwsgi_opt_set_int(opt, value, &uwsgi.async); + if (uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT] < 30) { + uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT] = 30; + } // set loop engine uwsgi.loop = "gevent"; @@ -173,9 +176,13 @@ edge: goto edge; } goto clear; - } +// on linux we need to set the socket in non-blocking as it is not inherited +#ifdef __linux__ + uwsgi_socket_nb(wsgi_req->poll.fd); +#endif + // hack to easily pass wsgi_req pointer to the greenlet PyTuple_SetItem(ugevent.greenlet_args, 1, PyLong_FromLong((long)wsgi_req)); @@ -195,6 +202,103 @@ clear: return Py_None; } +void uwsgi_gevent_nb_write(struct wsgi_request *wsgi_req, PyObject *str) { + PyObject *ret; + char *content = PyString_AsString(str); + size_t content_len = PyString_Size(str); + /// create a watcher for writes + PyObject *watcher = PyObject_CallMethod(ugevent.hub_loop, "io", "ii", wsgi_req->poll.fd, 2); + if (!watcher) goto error; + + PyObject *timer = PyObject_CallMethod(ugevent.hub_loop, "timer", "i", uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + if (!timer) { + Py_DECREF(watcher); + goto error; + } + + PyObject *current_greenlet = GET_CURRENT_GREENLET; + PyObject *current = PyObject_GetAttrString(current_greenlet, "switch"); + + char *ptr = content; + size_t remains = content_len; + + // this is the main writing cycle, wait for writability and send... + for(;;) { + ret = PyObject_CallMethod(watcher, "start", "OO", current, watcher); + if (!ret) { + stop_the_watchers + Py_DECREF(current); Py_DECREF(current_greenlet); + Py_DECREF(watcher); + Py_DECREF(timer); + goto error; + } + Py_DECREF(ret); + + ret = PyObject_CallMethod(timer, "start", "OO", current, timer); + if (!ret) { + stop_the_watchers + Py_DECREF(current); Py_DECREF(current_greenlet); + Py_DECREF(watcher); + Py_DECREF(timer); + goto error; + } + Py_DECREF(ret); + + ret = PyObject_CallMethod(ugevent.hub, "switch", NULL); + if (!ret) { + stop_the_watchers + Py_DECREF(current); Py_DECREF(current_greenlet); + Py_DECREF(watcher); + Py_DECREF(timer); + goto error; + } + Py_DECREF(ret); + + if (ret == timer) { + goto fail; + } + + // ok we can write a chunk to the socket + UWSGI_RELEASE_GIL + ssize_t len = write(wsgi_req->poll.fd, ptr, remains); + UWSGI_GET_GIL + if (len > 0) { + ptr += len; + remains -= len; + wsgi_req->response_size += len; + if (remains == 0) { + break; + } + stop_the_watchers + continue; + } + else if (len < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS) { + stop_the_watchers + continue; + } + } + +fail: + stop_the_watchers + Py_DECREF(current); Py_DECREF(current_greenlet); + Py_DECREF(watcher); + Py_DECREF(timer); + goto error; + } + + stop_the_watchers + Py_DECREF(current); Py_DECREF(current_greenlet); + Py_DECREF(watcher); + Py_DECREF(timer); + return ; + +error: + if (PyErr_Occurred()) + PyErr_Print(); + wsgi_req->write_errors++; +} + PyObject *uwsgi_gevent_wait(PyObject *watcher, PyObject *timer, PyObject *current) { PyObject *ret; @@ -237,7 +341,7 @@ PyObject *py_uwsgi_gevent_request(PyObject * self, PyObject * args) { watcher = PyObject_CallMethod(ugevent.hub_loop, "io", "ii", wsgi_req->poll.fd, 1); if (!watcher) goto clear1; - // a timer to implement timeoit (thanks Denis) + // a timer to implement timeout (thanks Denis) timer = PyObject_CallMethod(ugevent.hub_loop, "timer", "i", uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); if (!timer) goto clear0; @@ -330,6 +434,10 @@ void gevent_loop() { uwsgi_log("!!! Running gevent without threads IS NOT recommended, enable them with --enable-threads !!!\n"); } + if (uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT] < 30) { + uwsgi_log("!!! Running gevent with a socket-timeout lower than 30 seconds is not recommended, tune it with --socket-timeout !!!\n"); + } + // get the GIL UWSGI_GET_GIL @@ -344,6 +452,7 @@ void gevent_loop() { } uwsgi.current_wsgi_req = uwsgi_gevent_current_wsgi_req; + up.hook_write_string = uwsgi_gevent_nb_write; PyObject *gevent_dict = get_uwsgi_pydict("gevent"); if (!gevent_dict) uwsgi_pyexit; diff --git a/plugins/python/python_plugin.c b/plugins/python/python_plugin.c index bc540144..3cbd8440 100644 --- a/plugins/python/python_plugin.c +++ b/plugins/python/python_plugin.c @@ -9,6 +9,12 @@ extern struct http_status_codes hsc[]; extern PyTypeObject uwsgi_InputType; +void python_simple_hook_write_string(struct wsgi_request *wsgi_req, PyObject *str) { + UWSGI_RELEASE_GIL + wsgi_req->response_size += wsgi_req->socket->proto_write(wsgi_req, PyString_AsString(str), PyString_Size(str)); + UWSGI_GET_GIL +} + void uwsgi_opt_pythonpath(char *opt, char *value, void *foobar) { int i; @@ -232,6 +238,8 @@ pep405: up.wsgi_spitout = PyCFunction_New(uwsgi_spit_method, NULL); up.wsgi_writeout = PyCFunction_New(uwsgi_write_method, NULL); + up.hook_write_string = python_simple_hook_write_string; + up.main_thread = PyThreadState_Get(); // by default set a fake GIL (little impact on performance) diff --git a/plugins/python/uwsgi_python.h b/plugins/python/uwsgi_python.h index be8890bf..931089c4 100644 --- a/plugins/python/uwsgi_python.h +++ b/plugins/python/uwsgi_python.h @@ -186,6 +186,8 @@ struct uwsgi_python { char *pyrun; int start_response_nodelay; + void (*hook_write_string)(struct wsgi_request *, PyObject *); + char *programname; }; diff --git a/plugins/python/wsgi_subhandler.c b/plugins/python/wsgi_subhandler.c index 6ffd5d9d..0b9b45bb 100644 --- a/plugins/python/wsgi_subhandler.c +++ b/plugins/python/wsgi_subhandler.c @@ -163,16 +163,13 @@ int uwsgi_response_subhandler_wsgi(struct wsgi_request *wsgi_req) { // return or yield ? if (PyString_Check((PyObject *)wsgi_req->async_result)) { - char *content = PyString_AsString(wsgi_req->async_result); - size_t content_len = PyString_Size(wsgi_req->async_result); + size_t content_len = PyString_Size((PyObject *)wsgi_req->async_result); if (content_len > 0 && !wsgi_req->headers_sent) { if (uwsgi_python_do_send_headers(wsgi_req)) { goto clear; } } - UWSGI_RELEASE_GIL - wsgi_req->response_size += wsgi_req->socket->proto_write(wsgi_req, content, content_len); - UWSGI_GET_GIL + up.hook_write_string(wsgi_req, (PyObject *) wsgi_req->async_result); uwsgi_py_check_write_errors { uwsgi_py_write_exception(wsgi_req); } @@ -241,16 +238,13 @@ exception: if (PyString_Check(pychunk)) { - char *content = PyString_AsString(pychunk); size_t content_len = PyString_Size(pychunk); if (content_len > 0 && !wsgi_req->headers_sent) { if (uwsgi_python_do_send_headers(wsgi_req)) { goto clear; } } - UWSGI_RELEASE_GIL - wsgi_req->response_size += wsgi_req->socket->proto_write(wsgi_req, content, content_len); - UWSGI_GET_GIL + up.hook_write_string(wsgi_req, pychunk); uwsgi_py_check_write_errors { uwsgi_py_write_exception(wsgi_req); Py_DECREF(pychunk);