From c4eccd0d932edce7e9c435a2df1f6a7ea01f59ef Mon Sep 17 00:00:00 2001 From: "roberto@maverick64" Date: Mon, 14 Feb 2011 20:48:01 +0100 Subject: [PATCH] first attempt of a better wsgi.input --- plugins/python/python_plugin.c | 5 + plugins/python/uwsgi_pymodule.c | 4 +- plugins/python/uwsgi_python.h | 2 + plugins/python/wsgi_handlers.c | 192 ++++++++++++++++++++++++++++++- plugins/python/wsgi_subhandler.c | 6 +- signal.c | 39 ++++--- tests/signals.py | 12 +- utils.c | 35 ++++++ uwsgi.h | 5 +- 9 files changed, 268 insertions(+), 32 deletions(-) diff --git a/plugins/python/python_plugin.c b/plugins/python/python_plugin.c index e8539e97..61e147c2 100644 --- a/plugins/python/python_plugin.c +++ b/plugins/python/python_plugin.c @@ -3,6 +3,8 @@ extern struct uwsgi_server uwsgi; struct uwsgi_python up; +extern PyTypeObject uwsgi_InputType; + struct option uwsgi_python_options[] = { {"wsgi-file", required_argument, 0, LONG_ARGS_WSGI_FILE}, {"file", required_argument, 0, LONG_ARGS_FILE_CONFIG}, @@ -23,6 +25,7 @@ struct option uwsgi_python_options[] = { #endif {"catch-exceptions", no_argument, &up.catch_exceptions, 1}, {"ignore-script-name", no_argument, &up.ignore_script_name, 1}, + {"pep3333-input", no_argument, &up.pep3333_input, 1}, {"no-site", no_argument, &Py_NoSiteFlag, 1}, {0, 0, 0, 0}, @@ -324,6 +327,8 @@ void init_uwsgi_embedded_module() { PyObject *new_uwsgi_module, *zero; int i; + PyType_Ready(&uwsgi_InputType); + /* initialize for stats */ up.workers_tuple = PyTuple_New(uwsgi.numproc); for (i = 0; i < uwsgi.numproc; i++) { diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index c9c7efb9..d1c0fd59 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -402,10 +402,10 @@ PyObject *py_uwsgi_register_signal(PyObject * self, PyObject * args) { } if (payload == NULL) { - uwsgi_register_signal(uwsgi_signal, signal_kind, handler, 0, NULL, 0); + //uwsgi_register_signal(uwsgi_signal, signal_kind, handler, 0, NULL, 0); } else { - uwsgi_register_signal(uwsgi_signal, signal_kind, handler, 0, payload, strlen(payload)); + //uwsgi_register_signal(uwsgi_signal, signal_kind, handler, 0, payload, strlen(payload)); } Py_INCREF(Py_None); diff --git a/plugins/python/uwsgi_python.h b/plugins/python/uwsgi_python.h index 99faef57..c6393013 100644 --- a/plugins/python/uwsgi_python.h +++ b/plugins/python/uwsgi_python.h @@ -127,6 +127,8 @@ struct uwsgi_python { char *pymodule_alias[MAX_PYMODULE_ALIAS]; int pymodule_alias_cnt; + int pep3333_input; + void (*extension)(void); }; diff --git a/plugins/python/wsgi_handlers.c b/plugins/python/wsgi_handlers.c index 24d9d7c7..5a29260a 100644 --- a/plugins/python/wsgi_handlers.c +++ b/plugins/python/wsgi_handlers.c @@ -3,6 +3,166 @@ extern struct uwsgi_server uwsgi; extern struct uwsgi_python up; + +typedef struct uwsgi_Input { + PyObject_HEAD + off_t pos; + struct wsgi_request *wsgi_req; +} uwsgi_Input; + +PyObject *uwsgi_Input_iter(PyObject * self) { + Py_INCREF(self); + return self; +} + +PyObject *uwsgi_Input_next(PyObject * self) { + + PyErr_SetNone(PyExc_StopIteration); + + return NULL; +} + +static void uwsgi_Input_free(uwsgi_Input *self) { + PyObject_Del(self); +} + +static PyObject *uwsgi_Input_read(uwsgi_Input *self, PyObject *args) { + + long len = 0; + size_t remains, chunk_size; + ssize_t rlen; + char *tmp_buf; + int fd; + PyObject *res; + + if (!PyArg_ParseTuple(args, "|l:read", &len)) { + return NULL; + } + + // return empty string if no post_cl or pos >= post_cl + if (!self->wsgi_req->post_cl || (size_t) self->pos >= self->wsgi_req->post_cl) { + return PyString_FromString(""); + } + + if (uwsgi.post_buffering > 0) { + fd = -1; + if (self->wsgi_req->post_cl <= (size_t) uwsgi.post_buffering) { + fd = fileno(self->wsgi_req->async_post); + } + } + else { + fd = self->wsgi_req->poll.fd; + } + // return the whole input + if (len <= 0) { + remains = self->wsgi_req->post_cl; + } + else { + remains = len ; + } + + if (remains + self->pos > self->wsgi_req->post_cl) { + remains = self->wsgi_req->post_cl - self->pos; + } + + if (remains <= 0) { + return PyString_FromString(""); + } + + if (fd == -1) { + res = PyString_FromStringAndSize( self->wsgi_req->post_buffering_buf, remains); + self->pos += remains; + return res; + } + + chunk_size = remains; + tmp_buf = uwsgi_malloc(remains); + while(remains) { + if (uwsgi_waitfd(fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]) <= 0) { + free(tmp_buf); + return PyErr_Format(PyExc_ValueError, "error waiting for wsgi.input data"); + } + + rlen = read(fd, tmp_buf + self->pos, remains); + if (rlen < 0) { + free(tmp_buf); + return PyErr_Format(PyExc_ValueError, "error reading wsgi.input data"); + } + + if (!rlen) break; + + self->pos += rlen; + remains -= rlen; + } + + res = PyString_FromStringAndSize(tmp_buf, chunk_size); + free(tmp_buf); + return res; + +} + +static PyObject *uwsgi_Input_readline(uwsgi_Input *self, PyObject *args) { + + return NULL; +} + +static PyObject *uwsgi_Input_readlines(uwsgi_Input *self, PyObject *args) { + + return NULL; +} + +static PyObject *uwsgi_Input_close(uwsgi_Input *self, PyObject *args) { + + return NULL; +} + +static PyMethodDef uwsgi_Input_methods[] = { + { "read", (PyCFunction)uwsgi_Input_read, METH_VARARGS, 0 }, + { "readline", (PyCFunction)uwsgi_Input_readline, METH_VARARGS, 0 }, + { "readlines", (PyCFunction)uwsgi_Input_readlines, METH_VARARGS, 0 }, +// add close to allow mod_wsgi compatibility + { "close", (PyCFunction)uwsgi_Input_close, METH_VARARGS, 0 }, + { NULL, NULL} +}; + + +PyTypeObject uwsgi_InputType = { + PyVarObject_HEAD_INIT(NULL, 0) + "uwsgi._Input", /*tp_name */ + sizeof(uwsgi_Input), /*tp_basicsize */ + 0, /*tp_itemsize */ + (destructor) uwsgi_Input_free, /*tp_dealloc */ + 0, /*tp_print */ + 0, /*tp_getattr */ + 0, /*tp_setattr */ + 0, /*tp_compare */ + 0, /*tp_repr */ + 0, /*tp_as_number */ + 0, /*tp_as_sequence */ + 0, /*tp_as_mapping */ + 0, /*tp_hash */ + 0, /*tp_call */ + 0, /*tp_str */ + 0, /*tp_getattr */ + 0, /*tp_setattr */ + 0, /*tp_as_buffer */ +#if defined(Py_TPFLAGS_HAVE_ITER) + Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_ITER, +#else + Py_TPFLAGS_DEFAULT, +#endif + "uwsgi input object.", /* tp_doc */ + 0, /* tp_traverse */ + 0, /* tp_clear */ + 0, /* tp_richcompare */ + 0, /* tp_weaklistoffset */ + uwsgi_Input_iter, /* tp_iter: __iter__() method */ + uwsgi_Input_next, /* tp_iternext: next() method */ + uwsgi_Input_methods, + 0,0,0,0,0,0,0,0,0,0,0,0 +}; + + PyObject *py_uwsgi_write(PyObject * self, PyObject * args) { PyObject *data; char *content; @@ -82,6 +242,8 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { char *what; int what_len; + PyObject *wsgi_socket; + #ifdef UWSGI_ASYNC if (wsgi_req->async_status == UWSGI_AGAIN) { @@ -263,15 +425,38 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { - if (uwsgi.post_buffering > 0 && wsgi_req->post_cl > (size_t) uwsgi.post_buffering) { + if (uwsgi.post_buffering > 0) { UWSGI_RELEASE_GIL + // read to disk + if (wsgi_req->post_cl <= (size_t) uwsgi.post_buffering) { if (!uwsgi_read_whole_body(wsgi_req, wsgi_req->post_buffering_buf, uwsgi.post_buffering_bufsize)) { goto clear; } + } + else { + if (!uwsgi_read_whole_body_in_mem(wsgi_req, wsgi_req->post_buffering_buf)) { + goto clear; + } + } UWSGI_GET_GIL } + else { + wsgi_req->async_post = fdopen(wsgi_req->poll.fd, "r"); + } - wsgi_req->async_post = fdopen(wsgi_req->poll.fd, "r"); + if (!up.pep3333_input) { + wsgi_socket = PyFile_FromFile(wsgi_req->async_post, "wsgi_input", "r", NULL); + PyDict_SetItemString(wsgi_req->async_environ, "wsgi.input", wsgi_socket); + Py_DECREF(wsgi_socket); + } + else { + wsgi_socket = (PyObject *) PyObject_New(uwsgi_Input, &uwsgi_InputType); + ((uwsgi_Input*)wsgi_socket)->wsgi_req = wsgi_req; + ((uwsgi_Input*)wsgi_socket)->pos = 0; + PyDict_SetItemString(wsgi_req->async_environ, "wsgi.input", wsgi_socket); + } + + wsgi_req->async_result = wi->request_subhandler(wsgi_req, wi); @@ -327,6 +512,9 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { close(tmp_stderr); } + if (up.pep3333_input) { + Py_DECREF(wsgi_socket); + } clear: up.reset_ts(wsgi_req, wi); diff --git a/plugins/python/wsgi_subhandler.c b/plugins/python/wsgi_subhandler.c index 7312ea93..f006b335 100644 --- a/plugins/python/wsgi_subhandler.c +++ b/plugins/python/wsgi_subhandler.c @@ -5,7 +5,7 @@ extern struct uwsgi_python up; void *uwsgi_request_subhandler_wsgi(struct wsgi_request *wsgi_req, struct uwsgi_app *wi) { - PyObject *wsgi_socket, *zero; + PyObject *zero; //static PyObject *uwsgi_version = NULL; @@ -16,10 +16,6 @@ void *uwsgi_request_subhandler_wsgi(struct wsgi_request *wsgi_req, struct uwsgi_ */ - wsgi_socket = PyFile_FromFile(wsgi_req->async_post, "wsgi_input", "r", NULL); - PyDict_SetItemString(wsgi_req->async_environ, "wsgi.input", wsgi_socket); - Py_DECREF(wsgi_socket); - #ifdef UWSGI_SENDFILE PyDict_SetItemString(wsgi_req->async_environ, "wsgi.file_wrapper", wi->sendfile); #endif diff --git a/signal.c b/signal.c index a4f94dcc..d79a242d 100644 --- a/signal.c +++ b/signal.c @@ -8,9 +8,11 @@ int uwsgi_signal_handler(uint8_t sig) { use = &uwsgi.shared->signal_table[sig]; + /* if (!use->kind) { return -1; } + */ if (!uwsgi.p[use->modifier1]->signal_handler) { return -1; @@ -19,7 +21,7 @@ int uwsgi_signal_handler(uint8_t sig) { return uwsgi.p[use->modifier1]->signal_handler(sig, use->handler, use->payload, use->payload_size); } -void uwsgi_register_signal(uint8_t sig, uint8_t kind, void *handler, uint8_t modifier1, char *payload, uint8_t payload_size) { +void uwsgi_register_signal(uint8_t sig, char *receiver, void *handler, uint8_t modifier1, char *payload, uint8_t payload_size) { struct uwsgi_signal_entry *use = NULL; @@ -27,7 +29,7 @@ void uwsgi_register_signal(uint8_t sig, uint8_t kind, void *handler, uint8_t mod use = &uwsgi.shared->signal_table[sig]; - use->kind = kind; + strcpy(use->receiver, receiver); use->handler = handler; use->modifier1 = modifier1; @@ -57,7 +59,9 @@ void uwsgi_register_file_monitor(uint8_t sig, char *filename, uint8_t kind, void memcpy(ushared->files_monitored[ushared->files_monitored_cnt].filename, filename, strlen(filename)); ushared->files_monitored[ushared->files_monitored_cnt].registered = 0; ushared->files_monitored[ushared->files_monitored_cnt].sig = sig; - uwsgi_register_signal(sig, kind, handler, modifier1, filename, strlen(filename)); + + //uwsgi_register_signal(sig, kind, handler, modifier1, filename, strlen(filename)); + ushared->files_monitored_cnt++; } else { @@ -79,7 +83,7 @@ void uwsgi_register_timer(uint8_t sig, int secs, uint8_t kind, void *handler, ui snprintf(ushared->timers[ushared->timers_cnt].svalue, 0xff, "%d", secs); ushared->timers[ushared->timers_cnt].registered = 0; ushared->timers[ushared->timers_cnt].sig = sig; - uwsgi_register_signal(sig, kind, handler, modifier1, ushared->timers[ushared->timers_cnt].svalue, strlen(ushared->timers[ushared->timers_cnt].svalue)); + //uwsgi_register_signal(sig, kind, handler, modifier1, ushared->timers[ushared->timers_cnt].svalue, strlen(ushared->timers[ushared->timers_cnt].svalue)); ushared->timers_cnt++; } else { @@ -94,15 +98,20 @@ void uwsgi_register_timer(uint8_t sig, int secs, uint8_t kind, void *handler, ui void uwsgi_route_signal(uint8_t sig) { struct uwsgi_signal_entry *use = &ushared->signal_table[sig]; - switch(use->kind) { - case KIND_WORKER: - 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); - } - break; - case KIND_NULL: - // unmanaged signal, pass to all the children - break; - }; + // send to all workers + 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); + } + } + // send to first available worker + else if (!strcmp(use->receiver, "worker")) { + } + // loop back to master + else if (!strcmp(use->receiver, "master")) { + } + // route to subscribed + else if (!strcmp(use->receiver, "master")) { + } } diff --git a/tests/signals.py b/tests/signals.py index fc4ebe3c..4377768c 100644 --- a/tests/signals.py +++ b/tests/signals.py @@ -14,13 +14,13 @@ def hello_timer(num, secs): print "%s seconds elapsed" % secs #uwsgi.register_signal(30, uwsgi.SIGNAL_KIND_WORKER, hello_signal) -uwsgi.register_signal(30, uwsgi.KIND_WORKER, hello_signal) -uwsgi.register_signal(22, uwsgi.KIND_WORKER, hello_signal2, "*** PAYLOAD FOO ***") +uwsgi.register_signal(30, "workers", hello_signal) +uwsgi.register_signal(22, "worker", hello_signal2, "*** PAYLOAD FOO ***") -uwsgi.register_file_monitor(3, "/tmp", uwsgi.KIND_WORKER, hello_file) -uwsgi.register_timer(26, 2, uwsgi.KIND_WORKER, hello_timer) -uwsgi.register_timer(17, 4, uwsgi.KIND_WORKER, hello_timer) -uwsgi.register_timer(5, 8, uwsgi.KIND_WORKER, hello_timer) +uwsgi.register_file_monitor(3, "/tmp", "workers", hello_file) +uwsgi.register_timer(26, 2, "worker", hello_timer) +uwsgi.register_timer(17, 4, "worker2", hello_timer) +uwsgi.register_timer(5, 8, "worker3", hello_timer) def application(env, start_response): diff --git a/utils.c b/utils.c index 0184bdfd..615e740c 100644 --- a/utils.c +++ b/utils.c @@ -989,6 +989,41 @@ int count_options(struct option *lopt) { return count; } +int uwsgi_read_whole_body_in_mem(struct wsgi_request *wsgi_req, char *buf) { + + size_t post_remains = wsgi_req->post_cl; + int ret; + ssize_t len; + char *ptr = buf; + + while(post_remains) { + if (uwsgi.shared->options[UWSGI_OPTION_HARAKIRI] > 0) { + inc_harakiri(uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + } + + ret = uwsgi_waitfd(wsgi_req->poll.fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + if (ret < 0) { + return 0; + } + + if (!ret) { + uwsgi_log("buffering POST data timedout !!!\n"); + return 0; + } + + len = read(wsgi_req->poll.fd, ptr, post_remains); + if (len <= 0) { + uwsgi_error("read()"); + return 0; + } + ptr += len; + post_remains -= len; + } + + return 1; + +} + int uwsgi_read_whole_body(struct wsgi_request *wsgi_req, char *buf, size_t len) { size_t post_remains = wsgi_req->post_cl; diff --git a/uwsgi.h b/uwsgi.h index 82106578..11cace37 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -959,9 +959,9 @@ struct uwsgi_lb_group { #define KIND_MASTER 6 struct uwsgi_signal_entry { - uint8_t kind; uint8_t modifier1; uint8_t payload_size; + char receiver[0xff]; void *handler; char payload[0xff]; }; @@ -1303,6 +1303,7 @@ struct wsgi_request *threaded_current_wsgi_req(void); void build_options(void); int uwsgi_read_whole_body(struct wsgi_request *, char *, size_t); +int uwsgi_read_whole_body_in_mem(struct wsgi_request *, char *); ssize_t uwsgi_sendfile(struct wsgi_request *); @@ -1374,7 +1375,7 @@ struct uwsgi_fmon *event_queue_ack_file_monitor(int, int); void *uwsgi_mmap_shared_lock(void); void *uwsgi_mmap_shared_rwlock(void); -void uwsgi_register_signal(uint8_t, uint8_t, void *, uint8_t, char *, uint8_t); +void uwsgi_register_signal(uint8_t, char *, void *, uint8_t, char *, uint8_t); void uwsgi_register_file_monitor(uint8_t, char *, uint8_t, void *, uint8_t); void uwsgi_register_timer(uint8_t, int, uint8_t, void *, uint8_t); int uwsgi_signal_handler(uint8_t);