various test for trying to allow pre-connected requests

loop.c

@@ -85,7 +85,7 @@ void *simple_loop(void *arg1) {
 			continue;
 		}
 
-
+		uwsgi_log("accepted\n");
 		if (wsgi_req_recv(wsgi_req)) {
 			continue;
 		}

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;i<uwsgi.cores;i++) {
 			ulua.L[i] = lua_newstate(uwsgi_lua_alloc, NULL);
 			luaL_openlibs(ulua.L[i]);
+			luaL_register(ulua.L[i], "uwsgi", uwsgi_api);
 			if (luaL_loadfile(ulua.L[i], ulua.filename)) {
 				uwsgi_log("unable to load file %s\n", ulua.filename);
 				exit(1);
@@ -113,6 +292,7 @@ void uwsgi_lua_app() {
 int uwsgi_lua_request(struct wsgi_request *wsgi_req) {
 
 	int i;
+	int raw;
 	const char *http;
 	size_t slen;
 	ssize_t rlen;
@@ -185,6 +365,7 @@ int uwsgi_lua_request(struct wsgi_request *wsgi_req) {
 
 	//uwsgi_log("%d %s %s %s\n",i,lua_typename(L, lua_type(L, -3)), lua_typename(L, lua_type(L, -2)) ,  lua_typename(L, lua_type(L, -1)));
 
+	raw = 0;
 	// send status
 	if (lua_type(L, -3) == LUA_TSTRING || lua_type(L, -3) == LUA_TNUMBER) {
 		http = lua_tolstring(L, -3, &slen);
@@ -209,6 +390,10 @@ int uwsgi_lua_request(struct wsgi_request *wsgi_req) {
 			goto clear;
 		}
 	}
+	else {
+		raw = 1;
+		wsgi_req->status = -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

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]

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;
 }
 

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, ""},

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");
 	}
 

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;
 		}

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;

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) {

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

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

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