diff --git a/plugins/gevent/gevent.c b/plugins/gevent/gevent.c index 6847e86e..8fde0ff2 100644 --- a/plugins/gevent/gevent.c +++ b/plugins/gevent/gevent.c @@ -10,6 +10,10 @@ extern struct uwsgi_python up; if (ret) { Py_DECREF(ret); }\ ret = PyObject_CallMethod(watcher, "stop", NULL);\ if (ret) { Py_DECREF(ret); } +#define stop_the_watchers_and_clear stop_the_watchers\ + Py_DECREF(current); Py_DECREF(current_greenlet);\ + Py_DECREF(watcher);\ + Py_DECREF(timer); struct uwsgi_gevent { @@ -202,6 +206,123 @@ clear: return Py_None; } +ssize_t uwsgi_gevent_hook_input_read(struct wsgi_request *wsgi_req, char *tmp_buf, size_t remains) { + + size_t tmp_pos = 0; + /// create a watcher for reads + PyObject *watcher = PyObject_CallMethod(ugevent.hub_loop, "io", "ii", wsgi_req->poll.fd, 1); + if (!watcher) return -1; + + PyObject *timer = PyObject_CallMethod(ugevent.hub_loop, "timer", "i", uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + if (!timer) { + Py_DECREF(watcher); + return -1; + } + + PyObject *current_greenlet = GET_CURRENT_GREENLET; + PyObject *current = PyObject_GetAttrString(current_greenlet, "switch"); + + while(remains) { + + PyObject *ret = PyObject_CallMethod(watcher, "start", "OO", current, watcher); + if (!ret) { + stop_the_watchers_and_clear + return -1; + } + Py_DECREF(ret); + + ret = PyObject_CallMethod(timer, "start", "OO", current, timer); + if (!ret) { + stop_the_watchers_and_clear + return -1; + } + Py_DECREF(ret); + + ret = PyObject_CallMethod(ugevent.hub, "switch", NULL); + if (!ret) { + stop_the_watchers_and_clear + return -1; + } + Py_DECREF(ret); + + if (ret == timer) { + stop_the_watchers_and_clear + return 0; + } + + UWSGI_RELEASE_GIL; + ssize_t rlen = read(wsgi_req->poll.fd, tmp_buf+tmp_pos, remains); + if (rlen <= 0) { + UWSGI_GET_GIL + stop_the_watchers_and_clear + return -1; + } + tmp_pos += rlen; + remains -= rlen; + UWSGI_GET_GIL + stop_the_watchers + } + + return tmp_pos; + +} + + +ssize_t uwsgi_gevent_hook_input_readline(struct wsgi_request *wsgi_req, char *readline, size_t max_size) { + ssize_t rlen = 0; + + /// create a watcher for reads + PyObject *watcher = PyObject_CallMethod(ugevent.hub_loop, "io", "ii", wsgi_req->poll.fd, 1); + if (!watcher) return -1; + + PyObject *timer = PyObject_CallMethod(ugevent.hub_loop, "timer", "i", uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + if (!timer) { + Py_DECREF(watcher); + return -1; + } + + PyObject *current_greenlet = GET_CURRENT_GREENLET; + PyObject *current = PyObject_GetAttrString(current_greenlet, "switch"); + + PyObject *ret = PyObject_CallMethod(watcher, "start", "OO", current, watcher); + if (!ret) { + stop_the_watchers_and_clear + return -1; + } + Py_DECREF(ret); + + ret = PyObject_CallMethod(timer, "start", "OO", current, timer); + if (!ret) { + stop_the_watchers_and_clear + return -1; + } + Py_DECREF(ret); + + ret = PyObject_CallMethod(ugevent.hub, "switch", NULL); + if (!ret) { + stop_the_watchers_and_clear + return -1; + } + Py_DECREF(ret); + + if (ret == timer) { + stop_the_watchers_and_clear + return 0; + } + + UWSGI_RELEASE_GIL; + if (max_size > 0 && max_size < UWSGI_PY_READLINE_BUFSIZE) { + rlen = read(wsgi_req->poll.fd, readline, max_size); + } + else { + rlen = read(wsgi_req->poll.fd, readline, UWSGI_PY_READLINE_BUFSIZE); + } + UWSGI_GET_GIL; + stop_the_watchers_and_clear + return rlen; +} + + void uwsgi_gevent_nb_write(struct wsgi_request *wsgi_req, PyObject *str) { PyObject *ret; char *content = PyString_AsString(str); @@ -226,30 +347,21 @@ void uwsgi_gevent_nb_write(struct wsgi_request *wsgi_req, PyObject *str) { 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); + stop_the_watchers_and_clear 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); + stop_the_watchers_and_clear 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); + stop_the_watchers_and_clear goto error; } Py_DECREF(ret); @@ -280,10 +392,7 @@ void uwsgi_gevent_nb_write(struct wsgi_request *wsgi_req, PyObject *str) { } fail: - stop_the_watchers - Py_DECREF(current); Py_DECREF(current_greenlet); - Py_DECREF(watcher); - Py_DECREF(timer); + stop_the_watchers_and_clear goto error; } @@ -453,6 +562,8 @@ void gevent_loop() { uwsgi.current_wsgi_req = uwsgi_gevent_current_wsgi_req; up.hook_write_string = uwsgi_gevent_nb_write; + up.hook_wsgi_input_read = uwsgi_gevent_hook_input_read; + up.hook_wsgi_input_readline = uwsgi_gevent_hook_input_readline; 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 3cbd8440..a26edf76 100644 --- a/plugins/python/python_plugin.c +++ b/plugins/python/python_plugin.c @@ -239,6 +239,8 @@ pep405: up.wsgi_writeout = PyCFunction_New(uwsgi_write_method, NULL); up.hook_write_string = python_simple_hook_write_string; + up.hook_wsgi_input_read = uwsgi_python_hook_simple_input_read; + up.hook_wsgi_input_readline = uwsgi_python_hook_simple_input_readline; up.main_thread = PyThreadState_Get(); diff --git a/plugins/python/uwsgi_python.h b/plugins/python/uwsgi_python.h index bd784899..efd21a2c 100644 --- a/plugins/python/uwsgi_python.h +++ b/plugins/python/uwsgi_python.h @@ -187,6 +187,8 @@ struct uwsgi_python { int start_response_nodelay; void (*hook_write_string)(struct wsgi_request *, PyObject *); + ssize_t (*hook_wsgi_input_read)(struct wsgi_request *, char *, size_t); + ssize_t (*hook_wsgi_input_readline)(struct wsgi_request *, char *, size_t); char *programname; }; @@ -279,6 +281,9 @@ int uwsgi_python_do_send_headers(struct wsgi_request *); void *uwsgi_python_tracebacker_thread(void *); PyObject *uwsgi_python_setup_thread(char *); +ssize_t uwsgi_python_hook_simple_input_read(struct wsgi_request *, char *, size_t); +ssize_t uwsgi_python_hook_simple_input_readline(struct wsgi_request *, char *, size_t); + #ifdef UWSGI_PYPY #undef UWSGI_MINTERPRETERS #endif diff --git a/plugins/python/wsgi_handlers.c b/plugins/python/wsgi_handlers.c index 49aa4bfc..362a6e25 100644 --- a/plugins/python/wsgi_handlers.c +++ b/plugins/python/wsgi_handlers.c @@ -9,6 +9,24 @@ PyObject *uwsgi_Input_iter(PyObject *self) { return self; } +ssize_t uwsgi_python_hook_simple_input_readline(struct wsgi_request *wsgi_req, char *readline, size_t max_size) { + ssize_t rlen = 0; + UWSGI_RELEASE_GIL; + if (uwsgi_waitfd(wsgi_req->poll.fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]) <= 0) { + UWSGI_GET_GIL + return 0; + } + + if (max_size > 0 && max_size < UWSGI_PY_READLINE_BUFSIZE) { + rlen = read(wsgi_req->poll.fd, readline, max_size); + } + else { + rlen = read(wsgi_req->poll.fd, readline, UWSGI_PY_READLINE_BUFSIZE); + } + UWSGI_GET_GIL; + return rlen; +} + PyObject *uwsgi_Input_getline(uwsgi_Input *self) { size_t i; ssize_t rlen; @@ -40,29 +58,18 @@ PyObject *uwsgi_Input_getline(uwsgi_Input *self) { } - UWSGI_RELEASE_GIL; - if (uwsgi_waitfd(wsgi_req->poll.fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]) <= 0) { - UWSGI_GET_GIL + rlen = up.hook_wsgi_input_readline(wsgi_req, self->readline, self->readline_max_size); + if (rlen < 0) { + return PyErr_Format(PyExc_IOError, "error reading for wsgi.input data (readline/getline)"); + } + else if (rlen == 0) { return PyErr_Format(PyExc_IOError, "error waiting for wsgi.input data (readline/getline)"); - } - - if (self->readline_max_size > 0 && self->readline_max_size < UWSGI_PY_READLINE_BUFSIZE) { - rlen = read(wsgi_req->poll.fd, self->readline, self->readline_max_size); } - else { - rlen = read(wsgi_req->poll.fd, self->readline, UWSGI_PY_READLINE_BUFSIZE); - } - if (rlen <= 0) { - UWSGI_GET_GIL - return PyErr_Format(PyExc_IOError, "error reading wsgi.input data (readline/getline)"); - } self->readline_size = rlen; self->readline_pos = 0; self->pos += rlen; - UWSGI_GET_GIL; - for(i=0;i<(size_t)rlen;i++) { if (self->readline[i] == '\n') { res = PyString_FromStringAndSize(self->readline, i+1); @@ -91,11 +98,37 @@ static void uwsgi_Input_free(uwsgi_Input *self) { PyObject_Del(self); } +ssize_t uwsgi_python_hook_simple_input_read(struct wsgi_request *wsgi_req, char *tmp_buf, size_t remains) { + + size_t tmp_pos = 0; + + UWSGI_RELEASE_GIL + + while(remains) { + if (uwsgi_waitfd(wsgi_req->poll.fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]) <= 0) { + UWSGI_GET_GIL + return 0; + } + + ssize_t rlen = read(wsgi_req->poll.fd, tmp_buf+tmp_pos, remains); + if (rlen <= 0) { + UWSGI_GET_GIL + return -1; + } + tmp_pos += rlen; + remains -= rlen; + } + + UWSGI_GET_GIL + return tmp_pos; + +} + static PyObject *uwsgi_Input_read(uwsgi_Input *self, PyObject *args) { long len = 0; - size_t remains, tmp_pos = 0; - ssize_t rlen; + size_t remains; + ssize_t tmp_pos = 0; char *tmp_buf; PyObject *res; @@ -144,28 +177,18 @@ static PyObject *uwsgi_Input_read(uwsgi_Input *self, PyObject *args) { return res; } - UWSGI_RELEASE_GIL - tmp_buf = uwsgi_malloc(remains); - while(remains) { - if (uwsgi_waitfd(self->wsgi_req->poll.fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]) <= 0) { - free(tmp_buf); - UWSGI_GET_GIL - return PyErr_Format(PyExc_IOError, "error waiting for wsgi.input data: Content-Length %llu requested %llu received %llu", (unsigned long long) self->wsgi_req->post_cl, (unsigned long long) remains + tmp_pos, (unsigned long long) tmp_pos); - } - - rlen = read(self->wsgi_req->poll.fd, tmp_buf+tmp_pos, remains); - if (rlen <= 0) { - free(tmp_buf); - UWSGI_GET_GIL - return PyErr_Format(PyExc_IOError, "error reading wsgi.input data: Content-Length %llu requested %llu received %llu", (unsigned long long) self->wsgi_req->post_cl, (unsigned long long) remains + tmp_pos, (unsigned long long) tmp_pos); - } - tmp_pos += rlen; - remains -= rlen; + tmp_pos = up.hook_wsgi_input_read(self->wsgi_req, tmp_buf, remains); + if (tmp_pos < 0) { + free(tmp_buf); + return PyErr_Format(PyExc_IOError, "error reading for wsgi.input data: Content-Length %llu requested %llu received %llu", (unsigned long long) self->wsgi_req->post_cl, (unsigned long long) (remains + (tmp_pos+1)), (unsigned long long) (tmp_pos+1)); + } + else if (tmp_pos == 0) { + free(tmp_buf); + return PyErr_Format(PyExc_IOError, "error waiting for wsgi.input data: Content-Length %llu requested %llu received %llu", (unsigned long long) self->wsgi_req->post_cl, (unsigned long long) (remains + tmp_pos), (unsigned long long) tmp_pos); } - UWSGI_GET_GIL self->pos += tmp_pos; res = PyString_FromStringAndSize(tmp_buf, tmp_pos); free(tmp_buf);