first attempt of a better wsgi.input

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

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

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

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

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

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")) {
+	}
 }

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

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;

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