From 245ab6d81519bf74003578e5ea8f4fd2e74770ca Mon Sep 17 00:00:00 2001 From: "roberto@sirius" Date: Sat, 11 Dec 2010 20:08:23 +0100 Subject: [PATCH] various test for trying to allow pre-connected requests --- loop.c | 2 +- plugins/lua/lua_plugin.c | 193 ++++++++++++++++++++++++++++++- plugins/lua/uwsgiplugin.py | 7 +- plugins/python/pyutils.c | 2 + plugins/python/uwsgi_pymodule.c | 23 +++- plugins/python/wsgi_handlers.c | 8 +- plugins/python/wsgi_subhandler.c | 11 +- protocol.c | 3 +- utils.c | 21 +++- uwsgi.h | 4 + uwsgi_API.txt | 52 +++++++++ uwsgirouter5.py | 10 ++ 12 files changed, 320 insertions(+), 16 deletions(-) create mode 100644 uwsgi_API.txt create mode 100644 uwsgirouter5.py diff --git a/loop.c b/loop.c index 555e81ac..1d7aa95b 100644 --- a/loop.c +++ b/loop.c @@ -85,7 +85,7 @@ void *simple_loop(void *arg1) { continue; } - + uwsgi_log("accepted\n"); if (wsgi_req_recv(wsgi_req)) { continue; } diff --git a/plugins/lua/lua_plugin.c b/plugins/lua/lua_plugin.c index 9482e3f1..ceaa7455 100644 --- a/plugins/lua/lua_plugin.c +++ b/plugins/lua/lua_plugin.c @@ -15,6 +15,8 @@ struct uwsgi_lua { #define LONG_ARGS_LUA_BASE 17000 + (6 * 100) #define LONG_ARGS_LUA LONG_ARGS_LUA_BASE + 1 +#define lca(L, n) ulua_check_args(L, __FUNCTION__, n) + struct option uwsgi_lua_options[] = { {"lua", required_argument, 0, LONG_ARGS_LUA}, @@ -23,6 +25,182 @@ struct option uwsgi_lua_options[] = { }; +static void ulua_check_args(lua_State *L, const char *func, int n) { + int args = lua_gettop(L); + char error[4096]; + if (args != n) { + if (n == 1) { + snprintf(error, 4096, "uwsgi.%s takes 1 parameter", func+10); + } + else { + snprintf(error, 4096, "uwsgi.%s takes %d parameters", func+10, n); + } + lua_pushstring(L, error); + lua_error(L); + } +} + +static int uwsgi_api_log(lua_State *L) { + + time_t tt; + const char *logline ; + + lca(L, 1); + + if (lua_isstring(L, 1)) { + logline = lua_tolstring(L, 1, NULL); + tt = time(NULL); + if (logline[strlen(logline)] != '\n') { + uwsgi_log( UWSGI_LOGBASE " %.*s] %s\n", 24, ctime(&tt), logline); + } + else { + uwsgi_log( UWSGI_LOGBASE " %.*s] %s", 24, ctime(&tt), logline); + } + } + + return 0; +} + + +static char *encode_lua_table(lua_State *L, int index, uint16_t *size) { + + char *buf, *ptrbuf; + char *key; + char *value; + size_t keylen; + size_t vallen; + + *size = 0; + + lua_pushnil(L); + while (lua_next(L, index) != 0) { + if (lua_isstring(L, -2) && lua_isstring(L, -1)) { + key = (char *) lua_tolstring(L, -2, &keylen); + value = (char *) lua_tolstring(L, -1, &vallen); + if (keylen > 0xffff || vallen > 0xffff) continue; + *size += (2+keylen+2+vallen); + } + lua_pop(L, 1); + } + + buf = malloc(*size); + if (!buf) { + uwsgi_error("malloc()"); + exit(1); + } + + ptrbuf = buf; + lua_pushnil(L); + while (lua_next(L, index) != 0) { + if (lua_isstring(L, -2) && lua_isstring(L, -1)) { + key = (char *) lua_tolstring(L, -2, &keylen); + value = (char *) lua_tolstring(L, -1, &vallen); + + if (keylen > 0xffff || vallen > 0xffff) continue; + + *ptrbuf++ = (uint8_t) (keylen & 0xff); + *ptrbuf++ = (uint8_t) ((keylen >>8) & 0xff); + memcpy(ptrbuf, key, keylen); ptrbuf += keylen; + *ptrbuf++ = (uint8_t) (vallen & 0xff); + *ptrbuf++ = (uint8_t) ((vallen >>8) & 0xff); + memcpy(ptrbuf, value, vallen); ptrbuf += vallen; + } + lua_pop(L, 1); + } + + return buf; +} + +static int uwsgi_api_send_message(lua_State *L) { + + int args = lua_gettop(L); + const char *host; + int uwsgi_fd; + uint8_t modifier1, modifier2; + char *pkt = NULL; + uint16_t pktsize = 0 ; + char buf[4096]; + int rlen; + int items = 0; + int input_fd = -1, timeout = -1, input_size = 0; + + // is this an fd ? + if (lua_isnumber(L, 1)) { + args = 1; + } + else if (lua_isstring(L, 1)) { + host = lua_tolstring(L, 1, NULL); + uwsgi_fd = uwsgi_connect((char *)host, timeout, 0); + modifier1 = lua_tonumber(L, 2); + modifier2 = lua_tonumber(L, 3); + if (args > 4) { + timeout = lua_tonumber(L, 5); + if (args == 7) { + input_fd = lua_tonumber(L, 6); + input_size = lua_tonumber(L, 7); + } + } + if (lua_istable(L,4)) { + // passed a table + pkt = encode_lua_table(L, 4, &pktsize); + } + uwsgi_send_message(uwsgi_fd, modifier1, modifier2, pkt, pktsize, input_fd, input_size, timeout); + free(pkt); + + for(;;) { + rlen = uwsgi_waitfd(uwsgi_fd, timeout); + if (rlen > 0) { + rlen = read(uwsgi_fd, buf, 4096); + if (rlen < 0) { + uwsgi_error("read()"); + break; + } + else if (rlen > 0) { + lua_pushlstring(L, buf, rlen); + items++; + } + else { + break; + } + } + else if (rlen == 0) { + uwsgi_log("uwsgi request timed out waiting for response\n"); + break; + } + } + + close(uwsgi_fd); + + } + + return items; +} + +static int uwsgi_api_cl(lua_State *L) { + + struct wsgi_request *wsgi_req = current_wsgi_req(); + + lua_pushnumber(L, wsgi_req->post_cl); + return 1; +} + +static int uwsgi_api_req_fd(lua_State *L) { + + struct wsgi_request *wsgi_req = current_wsgi_req(); + + lua_pushnumber(L, wsgi_req->poll.fd); + return 1; +} + +static const luaL_reg uwsgi_api[] = { + {"log", uwsgi_api_log}, + {"cl", uwsgi_api_cl}, + {"req_fd", uwsgi_api_req_fd}, + {"send_message", uwsgi_api_send_message}, + {NULL, NULL} +}; + + static void *uwsgi_lua_alloc(void *ud, void *ptr, size_t osize, size_t nsize) { if(nsize == 0) { @@ -95,6 +273,7 @@ void uwsgi_lua_app() { for(i=0;istatus = -1; + } // send headers @@ -235,9 +420,11 @@ int uwsgi_lua_request(struct wsgi_request *wsgi_req) { lua_pop(L, 1); } - if (write(wsgi_req->poll.fd, "\r\n", 2) != 2) { - perror("write()"); - goto clear; + if (!raw) { + if (write(wsgi_req->poll.fd, "\r\n", 2) != 2) { + perror("write()"); + goto clear; + } } // send body with coroutine diff --git a/plugins/lua/uwsgiplugin.py b/plugins/lua/uwsgiplugin.py index c95190b6..29c818c6 100644 --- a/plugins/lua/uwsgiplugin.py +++ b/plugins/lua/uwsgiplugin.py @@ -1,7 +1,12 @@ import os,sys +try: + LUALIB = os.environ['UWSGICONFIG_LUALIB'] +except: + LUALIB = 'lua5.1' + NAME='lua' CFLAGS = ['-I/usr/include/lua5.1/'] LDFLAGS = [] GCC_LIST = ['lua_plugin'] -LIBS = ['-llua5.1'] +LIBS = ['-l%s' % LUALIB] diff --git a/plugins/python/pyutils.c b/plugins/python/pyutils.c index 2e91497f..ce284fe0 100644 --- a/plugins/python/pyutils.c +++ b/plugins/python/pyutils.c @@ -27,6 +27,8 @@ PyObject *python_call(PyObject *callable, PyObject *args, int catch) { } #endif + uwsgi_log("python called\n"); + return pyret; } diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index 353d6978..aadce342 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -7,8 +7,6 @@ char *spool_buffer = NULL; extern struct uwsgi_server uwsgi; extern struct uwsgi_python up; -#define UWSGI_LOGBASE "[- uWSGI -" - char *uwsgi_encode_pydict(PyObject *pydict, uint16_t *size) { int i; @@ -38,20 +36,17 @@ char *uwsgi_encode_pydict(PyObject *pydict, uint16_t *size) { 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; } @@ -100,6 +95,11 @@ char *uwsgi_encode_pydict(PyObject *pydict, uint16_t *size) { key = PyTuple_GetItem(zero, 0); val = PyTuple_GetItem(zero, 1); + + if (!key || !val) { + PyErr_Print(); + } + if (!PyString_Check(key) || !PyString_Check(val)) { Py_DECREF(zero); continue; @@ -1155,6 +1155,7 @@ PyObject* uwsgi_Iter_next(PyObject *self) { char buf[4096]; UWSGI_RELEASE_GIL + uwsgi_log("waiting for data\n"); rlen = uwsgi_waitfd(ui->fd, ui->timeout); if (rlen > 0) { rlen = read(ui->fd, buf, 4096); @@ -1212,6 +1213,16 @@ static PyTypeObject uwsgi_IterType = { }; +PyObject *py_uwsgi_connect(PyObject * self, PyObject * args) { + + char *socket_name = NULL; + if (!PyArg_ParseTuple(args, "s:connect", &socket_name)) { + return NULL; + } + + return PyInt_FromLong(uwsgi_connect(socket_name, 0, 0)); +} + PyObject *py_uwsgi_async_connect(PyObject * self, PyObject * args) { char *socket_name = NULL; @@ -1795,6 +1806,8 @@ static PyMethodDef uwsgi_advanced_methods[] = { {"wait_fd_read", py_eventfd_read, METH_VARARGS, ""}, {"wait_fd_write", py_eventfd_write, METH_VARARGS, ""}, #endif + + {"connect", py_uwsgi_connect, METH_VARARGS, ""}, {"is_connected", py_uwsgi_is_connected, METH_VARARGS, ""}, {"send", py_uwsgi_send, METH_VARARGS, ""}, {"recv", py_uwsgi_recv, METH_VARARGS, ""}, diff --git a/plugins/python/wsgi_handlers.c b/plugins/python/wsgi_handlers.c index 96360f33..3755f893 100644 --- a/plugins/python/wsgi_handlers.c +++ b/plugins/python/wsgi_handlers.c @@ -274,8 +274,10 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { } } - - + if (wsgi_req->uh.modifier2 == 4) { + // for persistent connections + wsgi_req->leave_open = 1; + } if (uwsgi.post_buffering > 0 && wsgi_req->post_cl > (size_t) uwsgi.post_buffering) { @@ -285,7 +287,7 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { } UWSGI_GET_GIL } - else { + else if (!wsgi_req->leave_open) { wsgi_req->async_post = fdopen(wsgi_req->poll.fd, "r"); } diff --git a/plugins/python/wsgi_subhandler.c b/plugins/python/wsgi_subhandler.c index 8639bb39..1fd900d8 100644 --- a/plugins/python/wsgi_subhandler.c +++ b/plugins/python/wsgi_subhandler.c @@ -15,6 +15,7 @@ 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); @@ -125,6 +126,7 @@ int uwsgi_response_subhandler_wsgi(struct wsgi_request *wsgi_req) { goto clear; } + #ifdef UWSGI_SENDFILE if (wsgi_req->sendfile_obj == wsgi_req->async_result && wsgi_req->sendfile_fd != -1) { sf_len = uwsgi_sendfile(wsgi_req); @@ -142,6 +144,7 @@ int uwsgi_response_subhandler_wsgi(struct wsgi_request *wsgi_req) { } #endif + // ok its a yield if (!wsgi_req->async_placeholder) { wsgi_req->async_placeholder = PyObject_GetIter(wsgi_req->async_result); @@ -158,6 +161,7 @@ int uwsgi_response_subhandler_wsgi(struct wsgi_request *wsgi_req) { + pychunk = PyIter_Next(wsgi_req->async_placeholder); if (!pychunk) { @@ -167,6 +171,7 @@ int uwsgi_response_subhandler_wsgi(struct wsgi_request *wsgi_req) { + if (PyString_Check(pychunk)) { if ((wsize = write(wsgi_req->poll.fd, PyString_AsString(pychunk), PyString_Size(pychunk))) < 0) { uwsgi_error("write()"); @@ -184,11 +189,13 @@ int uwsgi_response_subhandler_wsgi(struct wsgi_request *wsgi_req) { } #endif + Py_DECREF(pychunk); UWSGI_RELEASE_GIL return UWSGI_AGAIN; clear: + if (wsgi_req->sendfile_fd != -1) { Py_DECREF((PyObject *)wsgi_req->async_sendfile); } @@ -196,7 +203,9 @@ clear: PyDict_Clear(wsgi_req->async_environ); } if (wsgi_req->async_post && !wsgi_req->fd_closed) { - fclose(wsgi_req->async_post); + if (!wsgi_req->leave_open) { + fclose(wsgi_req->async_post); + } if (!uwsgi.post_buffering || wsgi_req->post_cl <= (size_t) uwsgi.post_buffering) { wsgi_req->fd_closed = 1; } diff --git a/protocol.c b/protocol.c index c1e19e01..d5a7f47f 100644 --- a/protocol.c +++ b/protocol.c @@ -205,6 +205,7 @@ int uwsgi_parse_response(struct pollfd *upoll, int timeout, struct uwsgi_header if (!timeout) timeout = 1; /* first 4 byte header */ + uwsgi_log("poll()\n"); rlen = poll(upoll, 1, timeout * 1000); if (rlen < 0) { uwsgi_error("poll()"); @@ -334,7 +335,7 @@ int uwsgi_parse_vars(struct wsgi_request *wsgi_req) { #endif ptrbuf += 2; if (ptrbuf + strsize <= bufferend) { - //uwsgi_log("uwsgi %.*s = %.*s\n", wsgi_req->hvec[wsgi_req->var_cnt].iov_len, wsgi_req->hvec[wsgi_req->var_cnt].iov_base, strsize, ptrbuf); + uwsgi_log("uwsgi %.*s = %.*s\n", wsgi_req->hvec[wsgi_req->var_cnt].iov_len, wsgi_req->hvec[wsgi_req->var_cnt].iov_base, strsize, ptrbuf); if (!uwsgi_strncmp("SCRIPT_NAME", 11, wsgi_req->hvec[wsgi_req->var_cnt].iov_base, wsgi_req->hvec[wsgi_req->var_cnt].iov_len)) { wsgi_req->script_name = ptrbuf; wsgi_req->script_name_len = strsize; diff --git a/utils.c b/utils.c index bcd43f5f..9f8dc390 100644 --- a/utils.c +++ b/utils.c @@ -334,7 +334,11 @@ void uwsgi_as_root() { void uwsgi_close_request(struct wsgi_request *wsgi_req) { int waitpid_status; + int leave_open = 0 ; + int tmp_fd = -1; + void *async_post = NULL; + uwsgi_log("ending request\n"); gettimeofday(&wsgi_req->end_of_request, NULL); uwsgi.workers[uwsgi.mywid].running_time += (double) (((double) (wsgi_req->end_of_request.tv_sec * 1000000 + wsgi_req->end_of_request.tv_usec) - (double) (wsgi_req->start_of_request.tv_sec * 1000000 + wsgi_req->start_of_request.tv_usec)) / (double) 1000.0); @@ -345,7 +349,7 @@ void uwsgi_close_request(struct wsgi_request *wsgi_req) { // close the connection with the webserver - if (!wsgi_req->fd_closed) { + if (!wsgi_req->fd_closed && !wsgi_req->leave_open) { // NOTE, if we close the socket before receiving eventually sent data, socket layer will send a RST close(wsgi_req->poll.fd); } @@ -368,9 +372,20 @@ void uwsgi_close_request(struct wsgi_request *wsgi_req) { if (uwsgi.shared->options[UWSGI_OPTION_REAPER] == 1 || uwsgi.grunt) { while( waitpid(WAIT_ANY, &waitpid_status, WNOHANG) > 0); } + + if (wsgi_req->leave_open) { + tmp_fd = wsgi_req->poll.fd; + leave_open = 1; + async_post = wsgi_req->async_post; + } // reset request memset(wsgi_req, 0, sizeof(struct wsgi_request)); + if (leave_open) { + wsgi_req->leave_open = leave_open; + wsgi_req->poll.fd = tmp_fd; + wsgi_req->async_post = async_post; + } if (uwsgi.shared->options[UWSGI_OPTION_MAX_REQUESTS] > 0 && uwsgi.workers[uwsgi.mywid].requests >= uwsgi.shared->options[UWSGI_OPTION_MAX_REQUESTS]) { goodbye_cruel_world(); } @@ -410,6 +425,7 @@ int wsgi_req_recv(struct wsgi_request *wsgi_req) { gettimeofday(&wsgi_req->start_of_request, NULL); + if (!uwsgi_parse_response(&wsgi_req->poll, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT], (struct uwsgi_header *) wsgi_req, wsgi_req->buffer)) { return -1; } @@ -427,6 +443,7 @@ int wsgi_req_recv(struct wsgi_request *wsgi_req) { int wsgi_req_simple_accept(struct wsgi_request *wsgi_req, int fd) { + if (wsgi_req->leave_open) return 0; wsgi_req->poll.fd = accept(fd, (struct sockaddr *) &wsgi_req->c_addr, (socklen_t *) &wsgi_req->c_len); if (wsgi_req->poll.fd < 0) { @@ -446,6 +463,8 @@ int wsgi_req_accept(struct wsgi_request *wsgi_req) { int i; int ret; + if (wsgi_req->leave_open) return 0; + ret = poll(uwsgi.sockets_poll, uwsgi.sockets_cnt, -1); if (ret < 0) { diff --git a/uwsgi.h b/uwsgi.h index 60349f09..03dd4c9b 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -4,6 +4,8 @@ #define UWSGI_VERSION "0.9.7-dev" +#define UWSGI_LOGBASE "[- uWSGI -" + #define uwsgi_error(x) uwsgi_log("%s: %s [%s line %d]\n", x, strerror(errno), __FILE__, __LINE__); #define uwsgi_debug(x, ...) uwsgi_log("[uWSGI DEBUG] " x, __VA_ARGS__); @@ -571,6 +573,8 @@ struct wsgi_request { off_t frame_pos; int frame_len; + + int leave_open; }; #define LOADER_DYN 0 diff --git a/uwsgi_API.txt b/uwsgi_API.txt new file mode 100644 index 00000000..a6129e73 --- /dev/null +++ b/uwsgi_API.txt @@ -0,0 +1,52 @@ +send_message() +send_multi_message() +reload() +workers() +masterpid() +total_requests() +getoption(id) +get_option(id) +setoption(id, value) +set_option(id, value) +sorry_i_need_to_block() +request_id() +worker_id() +log(message) +disconnect() +grunt() +load_plugin() +lock() +unlock() +send() +cl() +req_fd() +signal(num,[payload]) +sendfile() +set_warning_message() +mem() +has_hook() +logsize() +send_multicast_message +cluster_nodes +cluster_best_node +async_sleep +async_connect +async_send_message +green_schedule +suspend +wait_fd_read(fd, [timeout]) +wait_fd_write(fd, timeout) +is_connected() +send +recv() +recv_block +recv_frame +close +parsefile +sharedarea_read +sharedarea_write +sharedarea_readbyte +sharedarea_writebyte +sharedarea_readlong +sharedarea_writelong +sharedarea_inclong diff --git a/uwsgirouter5.py b/uwsgirouter5.py new file mode 100644 index 00000000..f48626f9 --- /dev/null +++ b/uwsgirouter5.py @@ -0,0 +1,10 @@ + +import uwsgi + +fd = uwsgi.connect("127.0.0.1:3033") +uwsgi.send_message(fd, 0, 4, {"leave_open":"1"}) + +def application(e,s): + + for part in uwsgi.send_message(fd, 0, 4, e, 0, e['wsgi.input'].fileno(), uwsgi.cl()): + yield part