diff --git a/async.c b/async.c index 82c4ff2a..fa787da3 100644 --- a/async.c +++ b/async.c @@ -270,7 +270,7 @@ int async_del(int queuefd, int fd, int etype) { int async_get_timeout() { - struct wsgi_request* wsgi_req = uwsgi.wsgi_requests[0]; + struct wsgi_request* wsgi_req; int i; time_t curtime, tdelta = 0; int ret = 0; @@ -278,6 +278,7 @@ int async_get_timeout() { if (!uwsgi.async_running) return 0; for(i=0;iasync_status == UWSGI_AGAIN) { if (wsgi_req->async_timeout_expired) { return 0; @@ -288,7 +289,6 @@ int async_get_timeout() { } } } - wsgi_req = uwsgi.wsgi_requests[i+1]; } curtime = time(NULL); @@ -303,19 +303,24 @@ int async_get_timeout() { void async_expire_timeouts() { - struct wsgi_request* wsgi_req = uwsgi.wsgi_requests[0]; + struct wsgi_request* wsgi_req; int i; time_t deadline = time(NULL); for(i=0;iasync_status == UWSGI_AGAIN && wsgi_req->async_timeout > 0) { if (wsgi_req->async_timeout <= deadline) { wsgi_req->async_timeout = 0; wsgi_req->async_timeout_expired = 1; + if (wsgi_req->async_waiting_fd != -1) { + async_del(uwsgi.async_queue, wsgi_req->async_waiting_fd, wsgi_req->async_waiting_fd_type); + wsgi_req->async_waiting_fd = -1; + wsgi_req->async_waiting_fd_monitored = 0; + } } } - wsgi_req = uwsgi.wsgi_requests[i+1]; } } diff --git a/http.c b/http.c index 592ed040..bbdd41bb 100644 --- a/http.c +++ b/http.c @@ -312,7 +312,7 @@ static void *http_request(void *u_h_r) { } } - uwsgi_fd = uwsgi_connect(uwsgi.sockets[0].name, 10); + uwsgi_fd = uwsgi_connect(uwsgi.sockets[0].name, 10, 0); if (uwsgi_fd >= 0) { ulen = (up - uwsgipkt) - 4; uwsgipkt[1] = (unsigned char) (ulen & 0xff); diff --git a/loop.c b/loop.c index 91822028..66a8477f 100644 --- a/loop.c +++ b/loop.c @@ -113,6 +113,8 @@ void complex_loop() { continue; } + + for(i=0; iasync_status = UWSGI_AGAIN; uwsgi.wsgi_req->async_waiting_fd = -1; uwsgi.wsgi_req->async_waiting_fd_monitored = 0; + uwsgi.wsgi_req->async_timeout = 0; } async_del(uwsgi.async_queue, uwsgi.async_events[i].ASYNC_FD, uwsgi.async_events[i].ASYNC_EV); diff --git a/plugins/nagios/nagios.c b/plugins/nagios/nagios.c index dd66d4af..383bcb7a 100644 --- a/plugins/nagios/nagios.c +++ b/plugins/nagios/nagios.c @@ -34,7 +34,7 @@ int nagios() { tcp_port[0] = 0; - nagios_poll.fd = connect_to_tcp(uwsgi.sockets[0].name, atoi(tcp_port + 1), uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + nagios_poll.fd = connect_to_tcp(uwsgi.sockets[0].name, atoi(tcp_port + 1), uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT], 0); if (nagios_poll.fd < 0) { fprintf(stdout, "UWSGI CRITICAL: could not connect() to workers\n"); exit(2); diff --git a/plugins/ping/ping_plugin.c b/plugins/ping/ping_plugin.c index 7b1c8774..6cca28f9 100644 --- a/plugins/ping/ping_plugin.c +++ b/plugins/ping/ping_plugin.c @@ -23,7 +23,7 @@ static void ping() { uwsgi_log("PING uwsgi host %s (timeout: %d)\n", uping.ping, uping.ping_timeout); - uwsgi_poll.fd = uwsgi_connect(uping.ping, uping.ping_timeout); + uwsgi_poll.fd = uwsgi_connect(uping.ping, uping.ping_timeout, 0); if (uwsgi_poll.fd < 0) { exit(1); } diff --git a/plugins/python/pyutils.c b/plugins/python/pyutils.c index 8f9c3446..2e91497f 100644 --- a/plugins/python/pyutils.c +++ b/plugins/python/pyutils.c @@ -5,6 +5,7 @@ extern struct uwsgi_python up; int manage_python_response(struct wsgi_request *wsgi_req) { // use standard WSGI response parse + wsgi_req->switches++; return uwsgi_response_subhandler_wsgi(wsgi_req); } diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index 3a5e7481..30c6e248 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -20,15 +20,195 @@ extern struct uwsgi_python up; #define UWSGI_LOGBASE "[- uWSGI -" -PyObject *py_uwsgi_send(PyObject * self, PyObject * args) { +char *uwsgi_encode_pydict(PyObject *pydict, uint16_t *size) { - char *data; + int i; + PyObject *zero, *key, *val; + uint16_t keysize, valsize; - if (!PyArg_ParseTuple(args, "s:send", &data)) { + + char *buf, *bufptr; + + PyObject *vars = PyDict_Items(pydict); + + if (!vars) { + PyErr_Print(); + return NULL; + } + + *size = 0; + + // calc the packet size + // try to fallback whenever possible + for (i = 0; i < PyList_Size(vars); i++) { + zero = PyList_GetItem(vars, i); + if (!zero) { + PyErr_Print(); + continue; + } + + if (!PyTuple_Check(zero)) { + uwsgi_log("invalid python dictionary item\n"); + Py_DECREF(zero); + continue; + } + + if (PyTuple_Size(zero) < 2) { + uwsgi_log("invalid python dictionary item\n"); + Py_DECREF(zero); + continue; + } + key = PyTuple_GetItem(zero, 0); + val = PyTuple_GetItem(zero, 1); + + if (!PyString_Check(key) || !PyString_Check(val)) { + Py_DECREF(zero); + continue; + } + + + keysize = PyString_Size(key); + valsize = PyString_Size(val); + + *size += (keysize + 2 + valsize + 2); + + // do not DECREF here !!! + //Py_DECREF(zero); + } + + if (*size <= 4) { + uwsgi_log("empty python dictionary\n"); + return NULL; + } + + // remember to free this memory !!! + buf = malloc(*size); + if (!buf) { + uwsgi_error("malloc()"); return NULL; } - if (write(uwsgi.wsgi_req->poll.fd, data, strlen(data)) < 0) { + bufptr = buf; + + for (i = 0; i < PyList_Size(vars); i++) { + zero = PyList_GetItem(vars, i); + if (!zero) { + PyErr_Print(); + continue; + } + + if (!PyTuple_Check(zero)) { + uwsgi_log("invalid python dictionary item\n"); + Py_DECREF(zero); + continue; + } + + if (PyTuple_Size(zero) < 2) { + uwsgi_log("invalid python dictionary item\n"); + Py_DECREF(zero); + continue; + } + key = PyTuple_GetItem(zero, 0); + val = PyTuple_GetItem(zero, 1); + + if (!PyString_Check(key) || !PyString_Check(val)) { + Py_DECREF(zero); + continue; + } + + + keysize = PyString_Size(key); + valsize = PyString_Size(val); + if (bufptr + keysize + 2 + valsize + 2 <= buf + *size) { +#ifdef __BIG_ENDIAN__ + keysize = uwsgi_swap16(keysize); +#endif + memcpy(bufptr, &keysize, 2); + bufptr += 2; +#ifdef __BIG_ENDIAN__ + keysize = uwsgi_swap16(keysize); +#endif + memcpy(bufptr, PyString_AsString(key), keysize); + bufptr += keysize; +#ifdef __BIG_ENDIAN__ + valsize = uwsgi_swap16(valsize); +#endif + memcpy(bufptr, &valsize, 2); + bufptr += 2; +#ifdef __BIG_ENDIAN__ + valsize = uwsgi_swap16(valsize); +#endif + memcpy(bufptr, PyString_AsString(val), valsize); + bufptr += valsize; + } + + Py_DECREF(zero); + + } + + return buf; + +} + +PyObject *py_uwsgi_close(PyObject * self, PyObject * args) { + + int fd; + + if (!PyArg_ParseTuple(args, "i:close", &fd)) { + return NULL; + } + + close(fd); + + + Py_INCREF(Py_None); + return Py_None; + +} + +PyObject *py_uwsgi_recv(PyObject * self, PyObject * args) { + + int fd, max_size = 4096; + char buf[4096]; + ssize_t rlen ; + + + if (!PyArg_ParseTuple(args, "i|i:recv", &fd, &max_size)) { + return NULL; + } + + // security check + if (max_size > 4096) max_size = 4096; + + rlen = read(fd, buf, max_size) ; + if ( rlen > 0) { + return PyString_FromStringAndSize(buf, rlen); + } + + Py_INCREF(Py_None); + return Py_None; +} + +PyObject *py_uwsgi_send(PyObject * self, PyObject * args) { + + PyObject *data; + PyObject *arg1, *arg2; + + int uwsgi_fd = uwsgi.wsgi_req->poll.fd ; + + if (!PyArg_ParseTuple(args, "O|O:send", &arg1, &arg2)) { + return NULL; + } + + if (PyTuple_Size(args) > 1) { + uwsgi_fd = PyInt_AsLong(arg1); + data = arg2; + } + else { + data = arg1; + } + + if (write(uwsgi_fd, PyString_AsString(data), PyString_Size(data)) < 0) { uwsgi_error("write()"); Py_INCREF(Py_None); return Py_None; @@ -765,64 +945,205 @@ clear: return Py_None; } - PyObject *py_uwsgi_send_message(PyObject * self, PyObject * args) { +typedef struct { + PyObject_HEAD + int fd; + int timeout; +} uwsgi_Iter; - PyObject *arg_message = NULL; - const char *arg_host = NULL; - int arg_port = 0; - int arg_modifier1 = 0; - int arg_modifier2 = 0; - int arg_timeout = 0; +PyObject* uwsgi_Iter_iter(PyObject *self) { + Py_INCREF(self); + return self; +} - //PyObject *marshalled; - //PyObject *retobject; +PyObject* uwsgi_Iter_next(PyObject *self) { + int rlen; + uwsgi_Iter *ui = (uwsgi_Iter *)self; + char buf[4096]; - if (!PyArg_ParseTuple(args, "siiiO|i:send_uwsgi_message", &arg_host, &arg_port, &arg_modifier1, &arg_modifier2, &arg_message, &arg_timeout)) { - return NULL; + rlen = uwsgi_waitfd(ui->fd, ui->timeout); + if (rlen > 0) { + rlen = read(ui->fd, buf, 4096); + if (rlen < 0) { + uwsgi_error("read()"); + } + else if (rlen > 0) { + return PyString_FromStringAndSize(buf, rlen); } - - /* - switch (arg_modifier1) { - case UWSGI_MODIFIER_MESSAGE_MARSHAL: - marshalled = PyMarshal_WriteObjectToString(arg_message, 1); - if (!marshalled) { - PyErr_Print(); - Py_INCREF(Py_None); - return Py_None; - } - retobject = uwsgi_send_message(arg_host, arg_port, arg_modifier1, arg_modifier2, PyString_AsString(marshalled), PyString_Size(marshalled), arg_timeout); - Py_DECREF(marshalled); - if (!retobject) { - PyErr_Print(); - PyErr_Clear(); - } - else { - return retobject; - } - break; - case UWSGI_MODIFIER_ADMIN_REQUEST: - if (PyString_Check(arg_message)) { - retobject = uwsgi_send_message(arg_host, arg_port, arg_modifier1, arg_modifier2, PyString_AsString(arg_message), PyString_Size(arg_message), arg_timeout); - if (!retobject) { - PyErr_Print(); - PyErr_Clear(); - } - else { - return retobject; - } - } - break; - default: - break; - } - -*/ - - Py_INCREF(Py_None); - return Py_None; - } + else if (rlen == 0) { + uwsgi_log("uwsgi request timed out waiting for response\n"); + } + + PyErr_SetNone(PyExc_StopIteration); + + return NULL; +} + +static PyTypeObject uwsgi_IterType = { + PyObject_HEAD_INIT(NULL) + 0, /*ob_size*/ + "uwsgi._Iter", /*tp_name*/ + sizeof(uwsgi_Iter), /*tp_basicsize*/ + 0, /*tp_itemsize*/ + 0, /*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_getattro*/ + 0, /*tp_setattro*/ + 0, /*tp_as_buffer*/ + Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_ITER, + "uwsgi response iterator object.", /* tp_doc */ + 0, /* tp_traverse */ + 0, /* tp_clear */ + 0, /* tp_richcompare */ + 0, /* tp_weaklistoffset */ + uwsgi_Iter_iter, /* tp_iter: __iter__() method */ + uwsgi_Iter_next /* tp_iternext: next() method */ +}; + + +PyObject *py_uwsgi_async_connect(PyObject * self, PyObject * args) { + + char *socket_name = NULL; + if (!PyArg_ParseTuple(args, "s:async_connect", &socket_name)) { + return NULL; + } + + return PyInt_FromLong(uwsgi_connect(socket_name, 0, 1)); +} + +PyObject *py_uwsgi_async_send_message(PyObject * self, PyObject * args) { + + PyObject *pyobj = NULL, *marshalled = NULL; + + int uwsgi_fd; + int modifier1 = 0; + int modifier2 = 0; + + ssize_t ret ; + + char *encoded; + uint16_t esize = 0; + + if (!PyArg_ParseTuple(args, "iiiO:async_send_message", &uwsgi_fd, &modifier1, &modifier2, &pyobj)) { + return NULL; + } + + if (uwsgi_fd < 0) goto clear; + + // now check for the type of object to send (fallback to marshal) + if (PyDict_Check(pyobj)) { + encoded = uwsgi_encode_pydict(pyobj, &esize); + if (esize > 0) { + ret = uwsgi_send_message(uwsgi_fd, (uint8_t) modifier1, (uint8_t) modifier2, encoded, esize, -1, 0, 0); + free(encoded); + } + } + else if (PyString_Check(pyobj)) { + ret = uwsgi_send_message(uwsgi_fd, (uint8_t) modifier1, (uint8_t) modifier2, PyString_AsString(pyobj), PyString_Size(pyobj), -1, 0, 0); + } + else { + marshalled = PyMarshal_WriteObjectToString(pyobj, 1); + if (!marshalled) { + PyErr_Print(); + goto clear; + } + ret = uwsgi_send_message(uwsgi_fd, (uint8_t) modifier1, (uint8_t) modifier2, PyString_AsString(marshalled), PyString_Size(marshalled), -1, 0, 0); + } + +clear: + + Py_INCREF(Py_None); + return Py_None; + +} + +PyObject *py_uwsgi_send_message(PyObject * self, PyObject * args) { + + PyObject *destination = NULL, *pyobj = NULL, *marshalled = NULL; + + int modifier1 = 0; + int modifier2 = 0; + int timeout = uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]; + int fd = -1; + int cl = 0; + + ssize_t ret ; + + int uwsgi_fd = -1; + char *encoded; + uint16_t esize = 0; + int close_fd = 0; + + uwsgi_Iter *ui; + + if (!PyArg_ParseTuple(args, "OiiO|iii:send_message", &destination, &modifier1, &modifier2, &pyobj, &timeout, &fd, &cl)) { + return NULL; + } + + // first of all get the fd for the destination + if (PyInt_Check(destination)) { + uwsgi_fd = PyInt_AsLong(destination); + } + else if (PyString_Check(destination)) { + uwsgi_fd = uwsgi_connect(PyString_AsString(destination), timeout, 0); + close_fd = 1; + } + + if (uwsgi_fd < 0) goto clear; + + + // now check for the type of object to send (fallback to marshal) + if (PyDict_Check(pyobj)) { + encoded = uwsgi_encode_pydict(pyobj, &esize); + if (esize > 0) { + ret = uwsgi_send_message(uwsgi_fd, (uint8_t) modifier1, (uint8_t) modifier2, encoded, esize, fd, cl, timeout); + free(encoded); + } + } + else if (PyString_Check(pyobj)) { + ret = uwsgi_send_message(uwsgi_fd, (uint8_t) modifier1, (uint8_t) modifier2, PyString_AsString(pyobj), PyString_Size(pyobj), fd, cl, timeout); + } + else { + marshalled = PyMarshal_WriteObjectToString(pyobj, 1); + if (!marshalled) { + PyErr_Print(); + goto clear2; + } + ret = uwsgi_send_message(uwsgi_fd, (uint8_t) modifier1, (uint8_t) modifier2, PyString_AsString(marshalled), PyString_Size(marshalled), fd, cl, timeout); + } + + // request sent, return the iterator response + ui = PyObject_New(uwsgi_Iter, &uwsgi_IterType); + if (!ui) { + PyErr_Print(); + goto clear2; + } + + ui->fd = uwsgi_fd; + ui->timeout = timeout; + + return (PyObject *) ui; + +clear2: + if (close_fd) close(uwsgi_fd); +clear: + + Py_INCREF(Py_None); + return Py_None; + +} /* uWSGI masterpid */ PyObject *py_uwsgi_masterpid(PyObject * self, PyObject * args) { @@ -1163,9 +1484,9 @@ PyObject *py_uwsgi_suspend(PyObject * self, PyObject * args) { } - static PyMethodDef uwsgi_advanced_methods[] = { - {"send_uwsgi_message", py_uwsgi_send_message, METH_VARARGS, ""}, - {"send_multi_uwsgi_message", py_uwsgi_send_multi_message, METH_VARARGS, ""}, +static PyMethodDef uwsgi_advanced_methods[] = { + {"send_message", py_uwsgi_send_message, METH_VARARGS, ""}, + {"send_multi_message", py_uwsgi_send_multi_message, METH_VARARGS, ""}, {"reload", py_uwsgi_reload, METH_VARARGS, ""}, {"workers", py_uwsgi_workers, METH_VARARGS, ""}, {"masterpid", py_uwsgi_masterpid, METH_VARARGS, ""}, @@ -1196,27 +1517,35 @@ PyObject *py_uwsgi_suspend(PyObject * self, PyObject * args) { #endif #ifdef UWSGI_ASYNC {"async_sleep", py_uwsgi_async_sleep, METH_VARARGS, ""}, + {"async_connect", py_uwsgi_async_connect, METH_VARARGS, ""}, + {"async_send_message", py_uwsgi_async_send_message, METH_VARARGS, ""}, {"green_schedule", py_uwsgi_suspend, METH_VARARGS, ""}, {"suspend", py_uwsgi_suspend, METH_VARARGS, ""}, + {"wait_fd_read", py_eventfd_read, METH_VARARGS, ""}, + {"wait_fd_write", py_eventfd_write, METH_VARARGS, ""}, #endif + {"send", py_uwsgi_send, METH_VARARGS, ""}, + {"recv", py_uwsgi_recv, METH_VARARGS, ""}, + {"close", py_uwsgi_close, METH_VARARGS, ""}, + {"parsefile", py_uwsgi_parse_file, METH_VARARGS, ""}, //{"call_hook", py_uwsgi_call_hook, METH_VARARGS, ""}, {NULL, NULL}, - }; +}; - static PyMethodDef uwsgi_sa_methods[] = { - {"sharedarea_read", py_uwsgi_sharedarea_read, METH_VARARGS, ""}, - {"sharedarea_write", py_uwsgi_sharedarea_write, METH_VARARGS, ""}, - {"sharedarea_readbyte", py_uwsgi_sharedarea_readbyte, METH_VARARGS, ""}, - {"sharedarea_writebyte", py_uwsgi_sharedarea_writebyte, METH_VARARGS, ""}, - {"sharedarea_readlong", py_uwsgi_sharedarea_readlong, METH_VARARGS, ""}, - {"sharedarea_writelong", py_uwsgi_sharedarea_writelong, METH_VARARGS, ""}, - {"sharedarea_inclong", py_uwsgi_sharedarea_inclong, METH_VARARGS, ""}, - {NULL, NULL}, - }; +static PyMethodDef uwsgi_sa_methods[] = { + {"sharedarea_read", py_uwsgi_sharedarea_read, METH_VARARGS, ""}, + {"sharedarea_write", py_uwsgi_sharedarea_write, METH_VARARGS, ""}, + {"sharedarea_readbyte", py_uwsgi_sharedarea_readbyte, METH_VARARGS, ""}, + {"sharedarea_writebyte", py_uwsgi_sharedarea_writebyte, METH_VARARGS, ""}, + {"sharedarea_readlong", py_uwsgi_sharedarea_readlong, METH_VARARGS, ""}, + {"sharedarea_writelong", py_uwsgi_sharedarea_writelong, METH_VARARGS, ""}, + {"sharedarea_inclong", py_uwsgi_sharedarea_inclong, METH_VARARGS, ""}, + {NULL, NULL}, +}; @@ -1246,7 +1575,7 @@ PyObject *py_uwsgi_suspend(PyObject * self, PyObject * args) { } #endif - void init_uwsgi_module_advanced(PyObject * current_uwsgi_module) { +void init_uwsgi_module_advanced(PyObject * current_uwsgi_module) { PyMethodDef *uwsgi_function; PyObject *uwsgi_module_dict; @@ -1256,13 +1585,20 @@ PyObject *py_uwsgi_suspend(PyObject * self, PyObject * args) { exit(1); } - for (uwsgi_function = uwsgi_advanced_methods; uwsgi_function->ml_name != NULL; uwsgi_function++) { - PyObject *func = PyCFunction_New(uwsgi_function, NULL); - PyDict_SetItemString(uwsgi_module_dict, uwsgi_function->ml_name, func); - Py_DECREF(func); - } + uwsgi_IterType.tp_new = PyType_GenericNew; + if (PyType_Ready(&uwsgi_IterType) < 0) { + PyErr_Print(); + exit(1); } + for (uwsgi_function = uwsgi_advanced_methods; uwsgi_function->ml_name != NULL; uwsgi_function++) { + PyObject *func = PyCFunction_New(uwsgi_function, NULL); + PyDict_SetItemString(uwsgi_module_dict, uwsgi_function->ml_name, func); + Py_DECREF(func); + } + +} + void init_uwsgi_module_sharedarea(PyObject * current_uwsgi_module) { PyMethodDef *uwsgi_function; PyObject *uwsgi_module_dict; diff --git a/plugins/python/wsgi_handlers.c b/plugins/python/wsgi_handlers.c index 5f4dcc67..2a1d0ffb 100644 --- a/plugins/python/wsgi_handlers.c +++ b/plugins/python/wsgi_handlers.c @@ -26,7 +26,7 @@ PyObject *py_uwsgi_write(PyObject * self, PyObject * args) { #ifdef UWSGI_ASYNC PyObject *py_eventfd_read(PyObject * self, PyObject * args) { - int fd, timeout; + int fd, timeout = 0; struct wsgi_request *wsgi_req = current_wsgi_req(); @@ -38,6 +38,9 @@ PyObject *py_eventfd_read(PyObject * self, PyObject * args) { wsgi_req->async_waiting_fd = fd; wsgi_req->async_waiting_fd_type = ASYNC_IN; wsgi_req->async_waiting_fd_monitored = 0; + if (timeout > 0) { + wsgi_req->async_timeout = time(NULL) + timeout; + } } return PyString_FromString(""); @@ -45,7 +48,7 @@ PyObject *py_eventfd_read(PyObject * self, PyObject * args) { PyObject *py_eventfd_write(PyObject * self, PyObject * args) { - int fd, timeout; + int fd, timeout = 0; struct wsgi_request *wsgi_req = current_wsgi_req(); @@ -57,6 +60,9 @@ PyObject *py_eventfd_write(PyObject * self, PyObject * args) { wsgi_req->async_waiting_fd = fd; wsgi_req->async_waiting_fd_type = ASYNC_OUT; wsgi_req->async_waiting_fd_monitored = 0; + if (timeout > 0) { + wsgi_req->async_timeout = time(NULL) + timeout; + } } return PyString_FromString(""); @@ -290,14 +296,14 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { UWSGI_RELEASE_GIL - while ( wi->response_subhandler(wsgi_req) != UWSGI_OK) { - wsgi_req->switches++; + if (wi->response_subhandler(wsgi_req) != UWSGI_OK) { + wsgi_req->switches++; #ifdef UWSGI_ASYNC - if (uwsgi.async > 1) { - return UWSGI_AGAIN; - } -#endif + if (uwsgi.async > 1) { + return UWSGI_AGAIN; } +#endif + } } diff --git a/plugins/python/wsgi_subhandler.c b/plugins/python/wsgi_subhandler.c index d6469abc..faa1fe46 100644 --- a/plugins/python/wsgi_subhandler.c +++ b/plugins/python/wsgi_subhandler.c @@ -135,7 +135,7 @@ int uwsgi_response_subhandler_wsgi(struct wsgi_request *wsgi_req) { if (uwsgi.async > 1) { if (wsgi_req->response_size < wsgi_req->sendfile_fd_size) { UWSGI_RELEASE_GIL - return UWSGI_AGAIN; + return UWSGI_AGAIN; } } #endif @@ -152,7 +152,7 @@ int uwsgi_response_subhandler_wsgi(struct wsgi_request *wsgi_req) { #ifdef UWSGI_ASYNC if (uwsgi.async > 1) { UWSGI_RELEASE_GIL - return UWSGI_AGAIN; + return UWSGI_AGAIN; } #endif } diff --git a/protocol.c b/protocol.c index 79ad03c8..c1e19e01 100644 --- a/protocol.c +++ b/protocol.c @@ -100,7 +100,7 @@ int uwsgi_enqueue_message(char *host, int port, uint8_t modifier1, uint8_t modif uwsgi_poll.events = POLLIN; - if (timed_connect(&uwsgi_poll, (const struct sockaddr *) &uws_addr, sizeof(struct sockaddr_in), timeout)) { + if (timed_connect(&uwsgi_poll, (const struct sockaddr *) &uws_addr, sizeof(struct sockaddr_in), timeout, 0)) { uwsgi_error("connect()"); close(uwsgi_poll.fd); return -1; @@ -127,95 +127,77 @@ int uwsgi_enqueue_message(char *host, int port, uint8_t modifier1, uint8_t modif return uwsgi_poll.fd; } -/* -PyObject *uwsgi_send_message(const char *host, int port, uint8_t modifier1, uint8_t modifier2, char *message, int size, int timeout) { +ssize_t uwsgi_send_message(int fd, uint8_t modifier1, uint8_t modifier2, char *message, uint16_t size, int pfd, size_t plen, int timeout) { struct pollfd uwsgi_mpoll; - struct sockaddr_in uws_addr; - int cnt; + ssize_t cnt; struct uwsgi_header uh; - char buffer[0xFFFF]; + char buffer[4096]; + ssize_t ret = 0; + int pret; - - if (!timeout) - timeout = 1; - - if (size > 0xFFFF) { - uwsgi_log( "invalid object (marshalled) size\n"); - Py_INCREF(Py_None); - return Py_None; - } - - uwsgi_mpoll.events = POLLIN; - - uwsgi_mpoll.fd = socket(AF_INET, SOCK_STREAM, 0); - if (uwsgi_mpoll.fd < 0) { - uwsgi_error("socket()"); - Py_INCREF(Py_None); - return Py_None; - } - - memset(&uws_addr, 0, sizeof(struct sockaddr_in)); - uws_addr.sin_family = AF_INET; - uws_addr.sin_port = htons(port); - uws_addr.sin_addr.s_addr = inet_addr(host); - - UWSGI_SET_BLOCKING; - - if (timed_connect(&uwsgi_mpoll, (const struct sockaddr *) &uws_addr, sizeof(struct sockaddr_in), timeout)) { - uwsgi_error("connect()"); - close(uwsgi_mpoll.fd); - Py_INCREF(Py_None); - return Py_None; - } + if (!timeout) timeout = uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]; uh.modifier1 = modifier1; - uh.pktsize = (uint16_t) size; + uh.pktsize = size; uh.modifier2 = modifier2; - cnt = write(uwsgi_mpoll.fd, &uh, 4); + cnt = write(fd, &uh, 4); if (cnt != 4) { uwsgi_error("write()"); - close(uwsgi_mpoll.fd); - Py_INCREF(Py_None); - return Py_None; + return -1; } - cnt = write(uwsgi_mpoll.fd, message, size); + ret += cnt; + + cnt = write(fd, message, size); if (cnt != size) { uwsgi_error("write()"); - close(uwsgi_mpoll.fd); - Py_INCREF(Py_None); - return Py_None; + return -1; } + ret += cnt; - if (!uwsgi_parse_response(&uwsgi_mpoll, timeout, &uh, buffer)) { - UWSGI_UNSET_BLOCKING; - Py_INCREF(Py_None); - return Py_None; - } - - UWSGI_UNSET_BLOCKING; - - close(uwsgi_mpoll.fd); - - if (uh.modifier1 == UWSGI_MODIFIER_RESPONSE) { - if (!uh.modifier2) { - Py_INCREF(Py_None); - return Py_None; - } - else { - Py_INCREF(Py_True); - return Py_True; + // transfer data from one socket to another + if (pfd >= 0 && plen > 0) { + uwsgi_mpoll.fd = pfd; + uwsgi_mpoll.events = POLLIN; + + while(plen > 0) { + pret = poll(&uwsgi_mpoll, 1, timeout*1000); + if (pret < 0) { + uwsgi_error("poll()"); + return -1; + } + else if (pret == 0) { + uwsgi_log("timeout waiting for socket data\n"); + return -1; + } + else { + cnt = read(pfd, buffer, UMIN(4096, plen)); + if (cnt < 0) { + uwsgi_error("read()"); + return -1; + } + else if (cnt == 0) { + return ret; + } + // send to peer + if (write(fd, buffer, cnt) != cnt) { + uwsgi_error("write()"); + return -1; + } + ret += cnt; + plen -= cnt; + } } } - return PyMarshal_ReadObjectFromString(buffer, uh.pktsize); + + return ret; } -*/ int uwsgi_parse_response(struct pollfd *upoll, int timeout, struct uwsgi_header *uh, char *buffer) { int rlen, i; @@ -538,7 +520,7 @@ int uwsgi_ping_node(int node, struct wsgi_request *wsgi_req) { return -1; } - if (timed_connect(&uwsgi_poll, (const struct sockaddr *) &ucn->ucn_addr, sizeof(struct sockaddr_in), uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT])) { + if (timed_connect(&uwsgi_poll, (const struct sockaddr *) &ucn->ucn_addr, sizeof(struct sockaddr_in), uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT], 0)) { close(uwsgi_poll.fd); return -1; } diff --git a/socket.c b/socket.c index e1422ebf..178b186d 100644 --- a/socket.c +++ b/socket.c @@ -207,20 +207,24 @@ int bind_to_unix(char *socket_name, int listen_queue, int chmod_socket, int abst } #endif - int uwsgi_connect(char *socket_name, int timeout) { +int uwsgi_connect(char *socket_name, int timeout, int async) { - char *tcp_port = strchr(socket_name, ':'); + int ret; + char *tcp_port = strchr(socket_name, ':'); - if (tcp_port) { - tcp_port[0] = 0; - tcp_port++; - return connect_to_tcp(socket_name, atoi(tcp_port), timeout); - } - - return connect_to_unix(socket_name, timeout); + if (tcp_port) { + tcp_port[0] = 0; + tcp_port++; + ret = connect_to_tcp(socket_name, atoi(tcp_port), timeout, async); + // reset the socket name + tcp_port--; tcp_port[0] = ':'; + return ret; } - int connect_to_unix(char *socket_name, int timeout) { + return connect_to_unix(socket_name, timeout, async); +} + + int connect_to_unix(char *socket_name, int timeout, int async) { struct pollfd uwsgi_poll; struct sockaddr_un uws_addr; @@ -238,7 +242,7 @@ int bind_to_unix(char *socket_name, int listen_queue, int chmod_socket, int abst uwsgi_poll.events = POLLIN; - if (timed_connect(&uwsgi_poll, (const struct sockaddr *) &uws_addr, sizeof(struct sockaddr_un), timeout)) { + if (timed_connect(&uwsgi_poll, (const struct sockaddr *) &uws_addr, sizeof(struct sockaddr_un), timeout, async)) { uwsgi_error("connect()"); close(uwsgi_poll.fd); return -1; @@ -248,7 +252,7 @@ int bind_to_unix(char *socket_name, int listen_queue, int chmod_socket, int abst } - int connect_to_tcp(char *socket_name, int port, int timeout) { + int connect_to_tcp(char *socket_name, int port, int timeout, int async) { struct pollfd uwsgi_poll; struct sockaddr_in uws_addr; @@ -275,8 +279,8 @@ int bind_to_unix(char *socket_name, int listen_queue, int chmod_socket, int abst uwsgi_poll.events = POLLIN; - if (timed_connect(&uwsgi_poll, (const struct sockaddr *) &uws_addr, sizeof(struct sockaddr_in), timeout)) { - uwsgi_error("connect()"); + if (timed_connect(&uwsgi_poll, (const struct sockaddr *) &uws_addr, sizeof(struct sockaddr_in), timeout, async)) { + //uwsgi_error("connect()"); close(uwsgi_poll.fd); return -1; } @@ -407,7 +411,7 @@ int bind_to_tcp(char *socket_name, int listen_queue, char *tcp_port) { return serverfd; } - int timed_connect(struct pollfd *fdpoll, const struct sockaddr *addr, int addr_size, int timeout) { +int timed_connect(struct pollfd *fdpoll, const struct sockaddr *addr, int addr_size, int timeout, int async) { int arg, ret; int soopt; @@ -427,6 +431,17 @@ int bind_to_tcp(char *socket_name, int listen_queue, char *tcp_port) { } ret = connect(fdpoll->fd, addr, addr_size); + + + /* re-set blocking socket */ + arg &= (~O_NONBLOCK); + if (fcntl(fdpoll->fd, F_SETFL, arg) < 0) { + uwsgi_error("fcntl()"); + return -1; + } + + if (async) return 0; + if (ret < 0) { /* check what happened */ @@ -462,12 +477,7 @@ int bind_to_tcp(char *socket_name, int listen_queue, char *tcp_port) { } } - /* re-set blocking socket */ - arg &= (~O_NONBLOCK); - if (fcntl(fdpoll->fd, F_SETFL, arg) < 0) { - uwsgi_error("fcntl()"); - return -1; - } + return 0; diff --git a/utils.c b/utils.c index cdcda6b9..7cd2cc9f 100644 --- a/utils.c +++ b/utils.c @@ -1139,6 +1139,8 @@ int uwsgi_waitfd(int fd, int timeout) { upoll[0].fd = fd; upoll[0].events = POLLIN; + if (!timeout) timeout = uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]; + ret = poll(upoll, 1, timeout*1000); if (ret < 0) { diff --git a/uwsgi.c b/uwsgi.c index b6ce0537..52cd90aa 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -29,7 +29,7 @@ extern char **environ; static char *short_options = NULL; -static char *base_short_options = "s:p:t:x:d:l:v:b:mcaCTiMhrR:z:A:Q:L:y:"; +static char *base_short_options = "s:p:t:x:d:l:v:b:mcaCTiMhrR:z:A:Q:Ly:"; UWSGI_DECLARE_EMBEDDED_PLUGINS diff --git a/uwsgi.h b/uwsgi.h index 10633c9f..9184eb2a 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -925,10 +925,10 @@ void end_me(void); int bind_to_unix(char *, int, int, int); int bind_to_tcp(char *, int, char *); int bind_to_udp(char *, int); -int timed_connect(struct pollfd *, const struct sockaddr *, int, int); -int uwsgi_connect(char *, int); -int connect_to_tcp(char *, int, int); -int connect_to_unix(char *, int); +int timed_connect(struct pollfd *, const struct sockaddr *, int, int, int); +int uwsgi_connect(char *, int, int); +int connect_to_tcp(char *, int, int, int); +int connect_to_unix(char *, int, int); #ifdef UWSGI_SCTP int bind_to_sctp(char *, int, char *); #endif @@ -1179,3 +1179,7 @@ void uwsgi_stdin_sendto(char *, uint8_t, uint8_t); int uwsgi_cluster_add_me(void); char *generate_socket_name(char *); + +#define UMIN(a,b) ((a)>(b)?(b):(a)) + +ssize_t uwsgi_send_message(int, uint8_t, uint8_t, char *, uint16_t, int, size_t, int); diff --git a/uwsgirouter.py b/uwsgirouter.py index 7b75aad0..0e3647a6 100644 --- a/uwsgirouter.py +++ b/uwsgirouter.py @@ -2,25 +2,56 @@ import uwsgi def application(env, start_response): - print env - # open the socket - fd = uwsgi.async_connect("127.0.0.1:3032") + fd = uwsgi.async_connect("192.168.173.100:3032") + # wait for connection ready - yield uwsgi.wait_fd_write(fd, 30) + yield uwsgi.wait_fd_write(fd, 3) + + if env['x-wsgiorg.fdevent.timeout']: + print "connection timed out !!!" + raise StopIteration + + if fd < 0: + print "unable to connect" + raise StopIteration + # send request + # env can contains python objects, but send_message will discard them. + # In this way we will automagically have a congruent and valid uwsgi packet uwsgi.async_send_message(fd, 0, 0, env) # send the http body # ready body in async mode and resend to fd + # uwsgi.recv is a bit of magic as it will check for the wsgi_req timeout flag. If it is set None will be returned + # uwsgi.recv will use always an internal buffer of 4096, but can be limited in the number of bytes to read + cl = 0 + if env.has_key('CONTENT_LENGTH'): + cl = int(env['CONTENT_LENGTH']) - while 1: + if cl > 0: + input = env['wsgi.input'].fileno() + yield uwsgi.wait_fd_read(input, 30) + bufsize = min(cl, 4096) + body = uwsgi.recv(input, bufsize) + while body and cl > 0: + uwsgi.send(fd, body) + cl = cl - len(body) + yield uwsgi.wait_fd_read(input, 30) + bufsize = min(cl, 4096) + body = uwsgi.recv(input, bufsize) + + + # wait for response + yield uwsgi.wait_fd_read(fd, 30) + data = uwsgi.recv(fd) + # recv the data, if it returns None the callable will end + while data: + yield data # wait for response yield uwsgi.wait_fd_read(fd, 30) - - # recv the data, if it returns None the callable will end - yield uwsgi.recv(fd) + data = uwsgi.recv(fd) uwsgi.close(fd) diff --git a/uwsgirouter2.py b/uwsgirouter2.py new file mode 100644 index 00000000..c80acb71 --- /dev/null +++ b/uwsgirouter2.py @@ -0,0 +1,12 @@ + +import uwsgi + + +def application(e,s): + + cl = 0 + if e.has_key('CONTENT_LENGTH'): + cl = int(e['CONTENT_LENGTH']) + + for part in uwsgi.send_message("192.168.173.100:3032", 0, 0, e, 0, e['wsgi.input'].fileno(), cl): + yield part