From e195a5d72c186175535c7c34a0efc8aad89fc609 Mon Sep 17 00:00:00 2001
From: "roberto@debian32" <roberto@debian32>
Date: Sun, 13 Nov 2011 15:08:29 +0100
Subject: [PATCH] improved signal handling and rack mule_msg

---
 master.c                        |   9 +--
 mule.c                          | 123 ++++++++++++++++++++++++++++
 mulefunc.py                     |  27 +++++++
 plugins/python/uwsgi_pymodule.c | 131 +++++-------------------------
 plugins/rack/rack_api.c         | 139 +++++++++++++++++++++++++-------
 signal.c                        |  76 ++++++++++++-----
 uwsgi.c                         |  53 ++++--------
 uwsgi.h                         |   7 ++
 8 files changed, 359 insertions(+), 206 deletions(-)
 create mode 100644 mulefunc.py

diff --git a/master.c b/master.c
index d69c14a8..681b5856 100644
--- a/master.c
+++ b/master.c
@@ -1650,14 +1650,7 @@ next:
 		}
 		gettimeofday(&last_respawn, NULL);
 		uwsgi.respawn_delta = last_respawn.tv_sec;
-		// close the communication pipe
-		/*
-		close(uwsgi.workers[uwsgi.mywid].pipe[0]);
-		if (socketpair(AF_UNIX, SOCK_STREAM, 0, uwsgi.workers[uwsgi.mywid].pipe)) {
-			uwsgi_error("socketpair()\n");
-			continue;
-		}
-		*/
+
 		if (uwsgi_respawn_worker(uwsgi.mywid)) return 0;
 
 	}
diff --git a/mule.c b/mule.c
index 088eb041..342e86e7 100644
--- a/mule.c
+++ b/mule.c
@@ -12,6 +12,24 @@ extern struct uwsgi_server uwsgi;
 
 void uwsgi_mule_handler(void);
 
+void mule_send_msg(int fd, char *message, size_t len) {
+
+	socklen_t so_bufsize_len = sizeof(int);
+	int so_bufsize = 0;
+
+	if (write(fd, message, len) != (ssize_t) len) {
+		if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                        if (getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &so_bufsize, &so_bufsize_len)) {
+                                uwsgi_error("getsockopt()");
+                        }
+                        uwsgi_log("*** MULE MSG QUEUE IS FULL: buffer size %d bytes (you can tune it with --signal-bufsize) ***\n", so_bufsize);
+                }
+                else {
+                        uwsgi_error("mule_send_msg()");
+                }
+	}
+}
+
 void uwsgi_mule(int id) {
 
 	int i;
@@ -233,3 +251,108 @@ struct uwsgi_mule_farm *uwsgi_mule_farm_new(struct uwsgi_mule_farm **umf, struct
         return uwsgi_mf;
 }
 
+ssize_t uwsgi_mule_get_msg(int manage_signals, int manage_farms, char *message, size_t buffer_size, int timeout) {
+
+        ssize_t len = 0;
+        struct pollfd *mulepoll;
+        int count = 4;
+        int farms_count = 0;
+        uint8_t uwsgi_signal;
+        int i;
+
+	if (uwsgi.muleid == 0) return -1;
+
+        if (manage_signals) count = 2;
+
+        if (!manage_farms) goto next;
+
+        for(i=0;i<uwsgi.farms_cnt;i++) {
+                if (uwsgi_farm_has_mule(&uwsgi.farms[i], uwsgi.muleid)) farms_count++;
+        }
+next:
+
+        if (timeout > -1) timeout = timeout*1000;
+
+        mulepoll = uwsgi_malloc(sizeof(struct pollfd) * (count+farms_count));
+
+        mulepoll[0].fd = uwsgi.mules[uwsgi.muleid-1].queue_pipe[1];
+        mulepoll[0].events = POLLIN;
+        mulepoll[1].fd = uwsgi.shared->mule_queue_pipe[1];
+        mulepoll[1].events = POLLIN;
+        if (count > 2) {
+                mulepoll[2].fd = uwsgi.signal_socket;
+                mulepoll[2].events = POLLIN;
+                mulepoll[3].fd = uwsgi.my_signal_socket;
+                mulepoll[3].events = POLLIN;
+        }
+
+	if (farms_count > 0) {
+		int tmp_cnt = 0;
+        	for(i=0;i<uwsgi.farms_cnt;i++) {
+			if (uwsgi_farm_has_mule(&uwsgi.farms[i], uwsgi.muleid)) {
+                		mulepoll[count+tmp_cnt].fd = uwsgi.farms[i].queue_pipe[1];
+                		mulepoll[count+tmp_cnt].events = POLLIN;
+				tmp_cnt++;
+			}
+        	}
+	}
+
+        int ret = poll(mulepoll, count+farms_count, timeout);
+        if (ret <= 0) {
+                uwsgi_error("poll");
+        }
+        else {
+                if (mulepoll[0].revents & POLLIN) {
+                        len = read(uwsgi.mules[uwsgi.muleid-1].queue_pipe[1], message, buffer_size);
+                }
+                else if (mulepoll[1].revents & POLLIN) {
+                        len = read(uwsgi.shared->mule_queue_pipe[1], message, buffer_size);
+                }
+                else {
+                        if (count > 2) {
+                                int interesting_fd = -1;
+                                if (mulepoll[2].revents & POLLIN) {
+                                        interesting_fd = mulepoll[2].fd;
+                                }
+                                else if (mulepoll[3].revents & POLLIN) {
+                                        interesting_fd = mulepoll[3].fd;
+                                }
+
+                                if (interesting_fd > -1) {
+                                        len = read(interesting_fd, &uwsgi_signal, 1);
+                                        if (len <= 0) {
+                                                uwsgi_log_verbose("uWSGI mule %d braying: my master died, i will follow him...\n", uwsgi.muleid);
+                                                end_me(0);
+                                        }
+#ifdef UWSGI_DEBUG
+                                        uwsgi_log_verbose("master sent signal %d to mule %d\n", uwsgi_signal, uwsgi.muleid);
+#endif
+                                        if (uwsgi_signal_handler(uwsgi_signal)) {
+                                                uwsgi_log_verbose("error managing signal %d on mule %d\n", uwsgi_signal, uwsgi.mywid);
+                                        }
+					// set the error condition
+					len = -1;
+                                        goto clear;
+                                }
+                        }
+
+			// read messages in the farm
+                        for(i=0;i<farms_count;i++) {
+                                if (mulepoll[count+i].revents & POLLIN) {
+                                        len = read(mulepoll[count+i].fd, message, buffer_size);
+                                        break;
+                                }
+                        }
+                }
+        }
+
+        if (len < 0) {
+                uwsgi_error("read()");
+                goto clear;
+        }
+
+clear:
+        free(mulepoll);
+        return len;
+}
+
diff --git a/mulefunc.py b/mulefunc.py
new file mode 100644
index 00000000..6bacfafd
--- /dev/null
+++ b/mulefunc.py
@@ -0,0 +1,27 @@
+from uwsgidecorators import *
+
+@timer(3, target='mule1')
+def hello_timer(signum):
+    print "3 seconds elapsed"
+
+
+@mulefunc
+def conto_fino_a_dieci(uno, due, tre):
+    print "MULE ID %d: conto_fino_a_dieci" % uwsgi.mule_id(),uno,due,tre
+
+@mulefunc(2)
+def conto_fino_a_venti(uno, due, tre):
+    print "MULE ID %d: conto_fino_a_venti" % uwsgi.mule_id(),uno,due,tre
+
+@mulefunc('topogigio')
+def conto_fino_a_trenta(uno, due, tre):
+    print "MULE ID %d: conto_fino_a_trenta" % uwsgi.mule_id(),uno,due,tre
+
+
+
+def application(e, sr):
+   conto_fino_a_dieci(1,2,3)
+   conto_fino_a_venti(4,5,6)
+   conto_fino_a_trenta(7,8,9)
+   sr('200 OK', [('Content-Type','text/html')])
+   return "MULE !!!"
diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c
index 6e92a148..46dbff45 100644
--- a/plugins/python/uwsgi_pymodule.c
+++ b/plugins/python/uwsgi_pymodule.c
@@ -518,7 +518,6 @@ PyObject *py_uwsgi_register_signal(PyObject * self, PyObject * args) {
 PyObject *py_uwsgi_signal(PyObject * self, PyObject * args) {
 
 	uint8_t uwsgi_signal;
-	ssize_t rlen;
 
 	if (!PyArg_ParseTuple(args, "B:signal", &uwsgi_signal)) {
 		return NULL;
@@ -528,11 +527,7 @@ PyObject *py_uwsgi_signal(PyObject * self, PyObject * args) {
 	uwsgi_log("sending %d to master\n", uwsgi_signal);
 #endif
 
-	rlen = write(uwsgi.signal_socket, &uwsgi_signal, 1);
-	if (rlen != 1) {
-		uwsgi_error("write()");
-	}
-
+	uwsgi_signal_send(uwsgi.signal_socket, uwsgi_signal);
 
 	Py_INCREF(Py_None);
 	return Py_None;
@@ -1110,7 +1105,6 @@ PyObject *py_uwsgi_mule_msg(PyObject * self, PyObject * args) {
 	char *message = NULL;
 	Py_ssize_t message_len = 0;
 	PyObject *mule_obj = NULL;
-	ssize_t len;
 	int fd = -1;
 	int mule_id = -1;
 
@@ -1122,10 +1116,7 @@ PyObject *py_uwsgi_mule_msg(PyObject * self, PyObject * args) {
 		return PyErr_Format(PyExc_ValueError, "no mule configured");
 
 	if (mule_obj == NULL) {
-		len = write(uwsgi.shared->mule_queue_pipe[0], message, message_len);
-		if (len <= 0) {
-			uwsgi_error("write()");
-		}
+		mule_send_msg(uwsgi.shared->mule_queue_pipe[0], message, message_len);
 	}
 	else {
 		if (PyString_Check(mule_obj)) {
@@ -1152,10 +1143,7 @@ PyObject *py_uwsgi_mule_msg(PyObject * self, PyObject * args) {
 		}
 
 		if (fd > -1) {
-			len = write(fd, message, message_len);
-			if (len < 0) {
-				uwsgi_error("write()");
-			}
+			mule_send_msg(fd, message, message_len);
 		}
 	}
 
@@ -1167,17 +1155,13 @@ PyObject *py_uwsgi_mule_msg(PyObject * self, PyObject * args) {
 PyObject *py_uwsgi_mule_get_msg(PyObject * self, PyObject * args, PyObject *kwargs) {
 
 	ssize_t len = 0;
-	// this buffer will be configurable
+	// this buffer is configurable (default 64k)
 	char *message;
-	struct pollfd *mulepoll;
-	int count = 4;
-	int farms_count = 0;
-	uint8_t uwsgi_signal;
-	PyObject *manage_signals = NULL;
-	PyObject *manage_farms = NULL;
-	int buffer_size = 65536;
+	PyObject *py_manage_signals = NULL;
+	PyObject *py_manage_farms = NULL;
+	size_t buffer_size = 65536;
 	int timeout = -1;
-	int i;
+	int manage_signals = 1, manage_farms = 1;
 
 	static char *kwlist[] = {"signals", "buffer_size", "timeout", "farms", NULL};
 
@@ -1185,109 +1169,34 @@ PyObject *py_uwsgi_mule_get_msg(PyObject * self, PyObject * args, PyObject *kwar
 		return PyErr_Format(PyExc_ValueError, "you can receive mule messages only in a mule !!!");
 	}
 
-	if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|OOii:mule_get_msg", kwlist, &manage_signals, &manage_farms, &buffer_size, &timeout)) {
+	if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|OOii:mule_get_msg", kwlist, &py_manage_signals, &py_manage_farms, &buffer_size, &timeout)) {
 		return NULL;
 	}
 
-	if (manage_signals == Py_None || manage_signals == Py_False) {
-		count = 2;
+	// signals and farms are managed by default
+	if (py_manage_signals == Py_None || py_manage_signals == Py_False) {
+		manage_signals = 0;
 	}
 
-	if (manage_farms == Py_None || manage_farms == Py_False) {
-		goto next;
+	if (py_manage_farms == Py_None || py_manage_farms == Py_False) {
+		manage_farms = 0;
 	}
-	
-	for(i=0;i<uwsgi.farms_cnt;i++) {
-        	if (uwsgi_farm_has_mule(&uwsgi.farms[i], uwsgi.muleid)) farms_count++;
-        }
-
-
-next:
-
-	UWSGI_RELEASE_GIL;
-	if (timeout > -1) timeout = timeout*1000;
 
 	message = uwsgi_malloc(buffer_size);
-	mulepoll = uwsgi_malloc(sizeof(struct pollfd) * (count+farms_count));
-
-	mulepoll[0].fd = uwsgi.mules[uwsgi.muleid-1].queue_pipe[1];
-	mulepoll[0].events = POLLIN;
-	mulepoll[1].fd = uwsgi.shared->mule_queue_pipe[1];
-	mulepoll[1].events = POLLIN;
-	if (count > 2) {
-		mulepoll[2].fd = uwsgi.signal_socket;
-		mulepoll[2].events = POLLIN;
-		mulepoll[3].fd = uwsgi.my_signal_socket;
-		mulepoll[3].events = POLLIN;
-	}
-
-	for(i=0;i<farms_count;i++) {
-		mulepoll[count+i].fd = uwsgi.farms[i].queue_pipe[1];
-		mulepoll[count+i].events = POLLIN;
-	}
-
-	int ret = poll(mulepoll, count+farms_count, timeout);
-	if (ret <= 0) {
-		uwsgi_error("poll");
-	}
-	else {
-		if (mulepoll[0].revents & POLLIN) {
-			len = read(uwsgi.mules[uwsgi.muleid-1].queue_pipe[1], message, buffer_size);
-		}
-		else if (mulepoll[1].revents & POLLIN) {
-			len = read(uwsgi.shared->mule_queue_pipe[1], message, buffer_size);
-		}
-		else {
-			if (count > 2) {
-				int interesting_fd = -1;
-				if (mulepoll[2].revents & POLLIN) {
-					interesting_fd = mulepoll[2].fd;
-				}
-				else if (mulepoll[3].revents & POLLIN) {
-                                	interesting_fd = mulepoll[3].fd;
-                        	}
 	
-				if (interesting_fd > -1) {
-					len = read(interesting_fd, &uwsgi_signal, 1);
-                        		if (len <= 0) {
-                                		uwsgi_log_verbose("uWSGI mule %d braying: my master died, i will follow him...\n", uwsgi.muleid);
-                                		end_me(0);
-                        		}
-#ifdef UWSGI_DEBUG
-                        		uwsgi_log_verbose("master sent signal %d to mule %d\n", uwsgi_signal, uwsgi.muleid);
-#endif
-                        		if (uwsgi_signal_handler(uwsgi_signal)) {
-                                		uwsgi_log_verbose("error managing signal %d on mule %d\n", uwsgi_signal, uwsgi.mywid);
-                        		}
-					goto clear;
-				}
-			}
-
-			for(i=0;i<farms_count;i++) {
-				if (mulepoll[count+i].revents & POLLIN) {
-					len = read(mulepoll[count+i].fd, message, buffer_size);
-					break;
-				}
-			}
-		}
-	}
+	UWSGI_RELEASE_GIL;
+	len = uwsgi_mule_get_msg(manage_signals, manage_farms, message, buffer_size, timeout) ;
 	UWSGI_GET_GIL;
+
 	if (len < 0) {
-		uwsgi_error("read()");
-		goto clear2;
+		free(message);
+		Py_INCREF(Py_None);
+        	return Py_None;
 	}
 
 	PyObject *msg = PyString_FromStringAndSize(message, len);
 	free(message);
-	free(mulepoll);
 	return msg;
-clear:
-	UWSGI_GET_GIL;
-clear2:
-	free(message);
-	free(mulepoll);
-	Py_INCREF(Py_None);
-        return Py_None;
 }
 
 PyObject *py_uwsgi_farm_get_msg(PyObject * self, PyObject * args) {
diff --git a/plugins/rack/rack_api.c b/plugins/rack/rack_api.c
index 69ab2203..33204d45 100644
--- a/plugins/rack/rack_api.c
+++ b/plugins/rack/rack_api.c
@@ -4,6 +4,8 @@ extern struct uwsgi_server uwsgi;
 
 extern struct uwsgi_rack ur;
 
+#define uwsgi_rack_api(x, y, z) rb_define_module_function(rb_uwsgi_embedded, x, y, z)
+
 VALUE rack_uwsgi_setprocname(VALUE *class, VALUE rbname) {
 
 	Check_Type(rbname, T_STRING);
@@ -27,6 +29,83 @@ VALUE rack_uwsgi_mem(VALUE *class) {
 
 }
 
+int uwsgi_ruby_hash_mule_callback(VALUE key, VALUE val, VALUE arg_array) {
+	Check_Type(key, T_SYMBOL);
+	ID key_id = SYM2ID(key);
+	const char *key_name = rb_id2name(key_id);
+
+	if (!strcmp(key_name, "signals")) {
+		rb_ary_store(arg_array, 0, val);
+	}
+	else if (!strcmp(key_name, "farms")) {
+		rb_ary_store(arg_array, 1, val);
+	}
+	else if (!strcmp(key_name, "timeout")) {
+		rb_ary_store(arg_array, 2, val);
+	}
+	else if (!strcmp(key_name, "buffer_size")) {
+		rb_ary_store(arg_array, 3, val);
+	}
+
+	return 0;
+}
+
+VALUE rack_uwsgi_mule_get_msg(int argc, VALUE *argv, VALUE *class) {
+
+	int manage_signals = 1;
+	int manage_farms = 1;
+	int timeout = -1;
+	size_t buffer_size = 65536;
+	ssize_t len = 0;
+	char *message;
+
+        if (uwsgi.muleid == 0) {
+                rb_raise(rb_eRuntimeError, "you can receive mule messages only in a mule !!!");
+		return Qnil;
+        }
+
+	if (argc > 0) {
+		// 0 = manage_signals
+		// 1 = manage_farms
+		// 2 = timeout
+		// 3 = buffer_size
+		VALUE arg_array = rb_ary_new2(4);
+		Check_Type(argv[0], T_HASH);
+		rb_hash_foreach(argv[0], uwsgi_ruby_hash_mule_callback, arg_array);
+		
+		if (rb_ary_entry(arg_array, 0) == Qfalse) {
+			manage_signals = 0;
+		}
+
+		if (rb_ary_entry(arg_array, 1) == Qfalse) {
+			manage_farms = 0;
+		}
+
+		if (TYPE(rb_ary_entry(arg_array,2)) == T_FIXNUM) {
+			timeout = NUM2INT(rb_ary_entry(arg_array,2));
+		}
+
+		if (TYPE(rb_ary_entry(arg_array,3)) == T_FIXNUM || TYPE(rb_ary_entry(arg_array,3)) == T_BIGNUM) {
+			buffer_size = NUM2ULONG(rb_ary_entry(arg_array,3));
+		}
+		
+	}
+
+        message = uwsgi_malloc(buffer_size);
+
+        len = uwsgi_mule_get_msg(manage_signals, manage_farms, message, buffer_size, timeout) ;
+
+        if (len < 0) {
+                free(message);
+                return Qnil;
+        }
+
+        VALUE msg = rb_str_new(message, len);
+        free(message);
+        return msg;
+}
+
+
 VALUE rack_uwsgi_lock(int argc, VALUE *argv, VALUE *class) {
 
         int lock_num = 0;
@@ -488,40 +567,42 @@ VALUE uwsgi_ruby_signal(VALUE *class, VALUE signum) {
 void uwsgi_rack_init_api() {
 
 	VALUE rb_uwsgi_embedded = rb_define_module("UWSGI");
-        rb_define_module_function(rb_uwsgi_embedded, "suspend", uwsgi_ruby_suspend, 0);
-        rb_define_module_function(rb_uwsgi_embedded, "masterpid", uwsgi_ruby_masterpid, 0);
-        rb_define_module_function(rb_uwsgi_embedded, "async_sleep", uwsgi_ruby_async_sleep, 1);
-        rb_define_module_function(rb_uwsgi_embedded, "wait_fd_read", uwsgi_ruby_wait_fd_read, 2);
-        rb_define_module_function(rb_uwsgi_embedded, "wait_fd_write", uwsgi_ruby_wait_fd_write, 2);
-        rb_define_module_function(rb_uwsgi_embedded, "async_connect", uwsgi_ruby_async_connect, 1);
-        rb_define_module_function(rb_uwsgi_embedded, "signal", uwsgi_ruby_signal, 1);
-        rb_define_module_function(rb_uwsgi_embedded, "register_signal", uwsgi_ruby_register_signal, 3);
-        rb_define_module_function(rb_uwsgi_embedded, "register_rpc", uwsgi_ruby_register_rpc, -1);
-        rb_define_module_function(rb_uwsgi_embedded, "signal_registered", uwsgi_ruby_signal_registered, 1);
-        rb_define_module_function(rb_uwsgi_embedded, "signal_wait", uwsgi_ruby_signal_wait, -1);
-        rb_define_module_function(rb_uwsgi_embedded, "signal_received", uwsgi_ruby_signal_received, 0);
-        rb_define_module_function(rb_uwsgi_embedded, "add_cron", rack_uwsgi_add_cron, 6);
-        rb_define_module_function(rb_uwsgi_embedded, "add_timer", rack_uwsgi_add_timer, 2);
-        rb_define_module_function(rb_uwsgi_embedded, "add_rb_timer", rack_uwsgi_add_rb_timer, 2);
-        rb_define_module_function(rb_uwsgi_embedded, "add_file_monitor", rack_uwsgi_add_file_monitor, 2);
+        uwsgi_rack_api("suspend", uwsgi_ruby_suspend, 0);
+        uwsgi_rack_api("masterpid", uwsgi_ruby_masterpid, 0);
+        uwsgi_rack_api("async_sleep", uwsgi_ruby_async_sleep, 1);
+        uwsgi_rack_api("wait_fd_read", uwsgi_ruby_wait_fd_read, 2);
+        uwsgi_rack_api("wait_fd_write", uwsgi_ruby_wait_fd_write, 2);
+        uwsgi_rack_api("async_connect", uwsgi_ruby_async_connect, 1);
+        uwsgi_rack_api("signal", uwsgi_ruby_signal, 1);
+        uwsgi_rack_api("register_signal", uwsgi_ruby_register_signal, 3);
+        uwsgi_rack_api("register_rpc", uwsgi_ruby_register_rpc, -1);
+        uwsgi_rack_api("signal_registered", uwsgi_ruby_signal_registered, 1);
+        uwsgi_rack_api("signal_wait", uwsgi_ruby_signal_wait, -1);
+        uwsgi_rack_api("signal_received", uwsgi_ruby_signal_received, 0);
+        uwsgi_rack_api("add_cron", rack_uwsgi_add_cron, 6);
+        uwsgi_rack_api("add_timer", rack_uwsgi_add_timer, 2);
+        uwsgi_rack_api("add_rb_timer", rack_uwsgi_add_rb_timer, 2);
+        uwsgi_rack_api("add_file_monitor", rack_uwsgi_add_file_monitor, 2);
 
-        rb_define_module_function(rb_uwsgi_embedded, "setprocname", rack_uwsgi_setprocname, 1);
-        rb_define_module_function(rb_uwsgi_embedded, "mem", rack_uwsgi_mem, 0);
+        uwsgi_rack_api("setprocname", rack_uwsgi_setprocname, 1);
+        uwsgi_rack_api("mem", rack_uwsgi_mem, 0);
 
-        rb_define_module_function(rb_uwsgi_embedded, "lock", rack_uwsgi_lock, -1);
-        rb_define_module_function(rb_uwsgi_embedded, "unlock", rack_uwsgi_unlock, -1);
+        uwsgi_rack_api("lock", rack_uwsgi_lock, -1);
+        uwsgi_rack_api("unlock", rack_uwsgi_unlock, -1);
+
+        uwsgi_rack_api("mule_get_msg", rack_uwsgi_mule_get_msg, -1);
 
 
 	if (uwsgi.cache_max_items > 0) {
-        	rb_define_module_function(rb_uwsgi_embedded, "cache_get", rack_uwsgi_cache_get, 1);
-        	rb_define_module_function(rb_uwsgi_embedded, "cache_get!", rack_uwsgi_cache_get_exc, 1);
-        	rb_define_module_function(rb_uwsgi_embedded, "cache_exists", rack_uwsgi_cache_exists, 1);
-        	rb_define_module_function(rb_uwsgi_embedded, "cache_exists?", rack_uwsgi_cache_exists, 1);
-        	rb_define_module_function(rb_uwsgi_embedded, "cache_del", rack_uwsgi_cache_del, 1);
-        	rb_define_module_function(rb_uwsgi_embedded, "cache_set", rack_uwsgi_cache_set, 2);
-        	rb_define_module_function(rb_uwsgi_embedded, "cache_set!", rack_uwsgi_cache_set_exc, 2);
-        	rb_define_module_function(rb_uwsgi_embedded, "cache_update", rack_uwsgi_cache_update, 2);
-        	rb_define_module_function(rb_uwsgi_embedded, "cache_update!", rack_uwsgi_cache_update_exc, 2);
+        	uwsgi_rack_api("cache_get", rack_uwsgi_cache_get, 1);
+        	uwsgi_rack_api("cache_get!", rack_uwsgi_cache_get_exc, 1);
+        	uwsgi_rack_api("cache_exists", rack_uwsgi_cache_exists, 1);
+        	uwsgi_rack_api("cache_exists?", rack_uwsgi_cache_exists, 1);
+        	uwsgi_rack_api("cache_del", rack_uwsgi_cache_del, 1);
+        	uwsgi_rack_api("cache_set", rack_uwsgi_cache_set, 2);
+        	uwsgi_rack_api("cache_set!", rack_uwsgi_cache_set_exc, 2);
+        	uwsgi_rack_api("cache_update", rack_uwsgi_cache_update, 2);
+        	uwsgi_rack_api("cache_update!", rack_uwsgi_cache_update_exc, 2);
 	}
 
         VALUE uwsgi_rb_opt_hash = rb_hash_new();
diff --git a/signal.c b/signal.c
index b7b11ba5..385180ee 100644
--- a/signal.c
+++ b/signal.c
@@ -198,24 +198,68 @@ int uwsgi_signal_add_rb_timer(uint8_t sig, int secs, int iterations) {
 
 }
 
+void create_signal_pipe(int *sigpipe) {
 
+	if (socketpair(AF_UNIX, SOCK_STREAM, 0, sigpipe)) {
+        	uwsgi_error("socketpair()\n");
+        	exit(1);
+        }
+        uwsgi_socket_nb(sigpipe[0]);
+        uwsgi_socket_nb(sigpipe[1]);
+
+	if (uwsgi.signal_bufsize) {
+		if (setsockopt(sigpipe[0],  SOL_SOCKET, SO_SNDBUF, &uwsgi.signal_bufsize, sizeof(int))) {
+                        	uwsgi_error("setsockopt()");
+		}
+		if (setsockopt(sigpipe[0],  SOL_SOCKET, SO_RCVBUF, &uwsgi.signal_bufsize, sizeof(int))) {
+                        	uwsgi_error("setsockopt()");
+		}
+
+		if (setsockopt(sigpipe[1],  SOL_SOCKET, SO_SNDBUF, &uwsgi.signal_bufsize, sizeof(int))) {
+                        	uwsgi_error("setsockopt()");
+		}
+		if (setsockopt(sigpipe[1],  SOL_SOCKET, SO_RCVBUF, &uwsgi.signal_bufsize, sizeof(int))) {
+                        	uwsgi_error("setsockopt()");
+		}
+	}
+}
+
+int uwsgi_signal_send(int fd, uint8_t sig) {
+
+	socklen_t so_bufsize_len = sizeof(int);
+	int so_bufsize = 0;
+
+	if (write(fd, &sig, 1) != 1) {
+		if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                	if (getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &so_bufsize, &so_bufsize_len)) {
+                        	uwsgi_error("getsockopt()");
+                	}
+                	uwsgi_log("*** SIGNAL QUEUE IS FULL: buffer size %d bytes (you can tune it with --signal-bufsize) ***\n", so_bufsize);
+		}
+		else {
+			uwsgi_error("uwsgi_signal_send()");
+		}
+		return -1;
+        }
+	return 0;
+
+}
 
 void uwsgi_route_signal(uint8_t sig) {
 
-	int i;
 	struct uwsgi_signal_entry *use = &ushared->signal_table[sig];
+	int i;
+
 	// send to first available worker
 	if (use->receiver[0] == 0 || !strcmp(use->receiver, "worker") || !strcmp(use->receiver, "worker0")) {
-		if (write(ushared->worker_signal_pipe[0], &sig, 1) != 1) {
-			uwsgi_error("write()");
+		if (uwsgi_signal_send(ushared->worker_signal_pipe[0], sig)) {
 			uwsgi_log("could not deliver signal %d to workers pool\n", sig);
 		}
 	}
 	// send to all workers
 	else if (!strcmp(use->receiver, "workers")) {
 		for(i=1;i<=uwsgi.numproc;i++) {
-			if (write(uwsgi.workers[i].signal_pipe[0], &sig, 1) != 1) {
-				uwsgi_error("write()");
+			if (uwsgi_signal_send(uwsgi.workers[i].signal_pipe[0], sig)) {
 				uwsgi_log("could not deliver signal %d to worker %d\n", sig, i);
 			}
 		}
@@ -226,8 +270,7 @@ void uwsgi_route_signal(uint8_t sig) {
 		if (i > uwsgi.numproc) {
 			uwsgi_log("invalid signal target: %s\n", use->receiver);
 		}
-		if (write(uwsgi.workers[i].signal_pipe[0], &sig, 1) != 1) {
-                        uwsgi_error("write()");
+		if (uwsgi_signal_send(uwsgi.workers[i].signal_pipe[0], sig)) {
                 	uwsgi_log("could not deliver signal %d to worker %d\n", sig, i);
                 }
 	}
@@ -238,8 +281,7 @@ void uwsgi_route_signal(uint8_t sig) {
 #ifdef UWSGI_SPOOLER
 	else if (!strcmp(use->receiver, "spooler")) {
 		if (ushared->worker_signal_pipe[0] != -1) {
-			if (write(ushared->spooler_signal_pipe[0], &sig, 1) != 1) {
-                        	uwsgi_error("write()");
+			if (uwsgi_signal_send(ushared->spooler_signal_pipe[0], sig)) {
                         	uwsgi_log("could not deliver signal %d to the spooler\n", sig);
                 	}
 		}
@@ -247,8 +289,7 @@ void uwsgi_route_signal(uint8_t sig) {
 #endif
 	else if (!strcmp(use->receiver, "mules")) {
 		for(i=0;i<uwsgi.mules_cnt;i++) {
-			if (write(uwsgi.mules[i].signal_pipe[0], &sig, 1) != 1) {
-                                uwsgi_error("write()");
+			if (uwsgi_signal_send(uwsgi.mules[i].signal_pipe[0], sig)) {
                                 uwsgi_log("could not deliver signal %d to mule %d\n", sig, i+1);
                         }
 		}
@@ -259,14 +300,12 @@ void uwsgi_route_signal(uint8_t sig) {
 			uwsgi_log("invalid signal target: %s\n", use->receiver);
 		}
 		else if (i == 0) {
-			if (write(ushared->mule_signal_pipe[0], &sig, 1) != 1) {
-                                uwsgi_error("write()");
+			if (uwsgi_signal_send(ushared->mule_signal_pipe[0], sig)) {
                                 uwsgi_log("could not deliver signal %d to a mule\n", sig);
                         }
 		}
 		else {
-			if (write(uwsgi.mules[i-1].signal_pipe[0], &sig, 1) != 1) {
-                                uwsgi_error("write()");
+			if (uwsgi_signal_send(uwsgi.mules[i-1].signal_pipe[0], sig)) {
                                 uwsgi_log("could not deliver signal %d to mule %d\n", sig, i);
                         }
 		}
@@ -278,8 +317,7 @@ void uwsgi_route_signal(uint8_t sig) {
 			uwsgi_log("unknown farm: %s\n", name);
 			return;
 		}
-		if (write(uf->signal_pipe[0], &sig, 1) != 1) {
-                	uwsgi_error("write()");
+		if (uwsgi_signal_send(uf->signal_pipe[0], sig)) {
                         uwsgi_log("could not deliver signal %d to farm %d (%s)\n", sig, uf->id, uf->name);
                 }
 	}
@@ -289,8 +327,7 @@ void uwsgi_route_signal(uint8_t sig) {
 			uwsgi_log("invalid signal target: %s\n", use->receiver);
 		}
 		else {
-			if (write(uwsgi.farms[i-1].signal_pipe[0], &sig, 1) != 1) {
-                                uwsgi_error("write()");
+			if (uwsgi_signal_send(uwsgi.farms[i-1].signal_pipe[0], sig)) {
                                 uwsgi_log("could not deliver signal %d to farm %d (%s)\n", sig, i, uwsgi.farms[i-1].name);
                         }
 		}
@@ -300,6 +337,7 @@ void uwsgi_route_signal(uint8_t sig) {
 		// unregistered signal, sending it to all the workers
 		uwsgi_log("^^^ UNSUPPORTED SIGNAL TARGET: %s ^^^\n", use->receiver);
 	}
+
 }
 
 uint8_t uwsgi_signal_wait(int signum) {
diff --git a/uwsgi.c b/uwsgi.c
index c6320bdd..42c80883 100644
--- a/uwsgi.c
+++ b/uwsgi.c
@@ -132,6 +132,8 @@ static struct option long_base_options[] = {
 #endif
 	{"mule", optional_argument, 0, LONG_ARGS_MULE},
 	{"mules", required_argument, 0, LONG_ARGS_MULES},
+	{"signal-bufsize", required_argument, 0, LONG_ARGS_SIGNAL_BUFSIZE},
+	{"signals-bufsize", required_argument, 0, LONG_ARGS_SIGNAL_BUFSIZE},
 	{"farm", required_argument, 0, LONG_ARGS_FARM},
 	{"disable-logging", no_argument, 0, 'L'},
 
@@ -2258,10 +2260,7 @@ skipzero:
 
 	if (uwsgi.master_process) {
 		for(i=1;i<=uwsgi.numproc;i++) {
-			if (socketpair(AF_UNIX, SOCK_STREAM, 0, uwsgi.workers[i].signal_pipe)) {
-                        	uwsgi_error("socketpair()\n");
-				exit(1);
-                	}
+			create_signal_pipe(uwsgi.workers[i].signal_pipe);
 		}
 	}
 
@@ -2278,26 +2277,13 @@ skipzero:
 		}
 		memset(uwsgi.mules, 0, sizeof(struct uwsgi_mule) * uwsgi.mules_cnt);
 
-		if (socketpair(AF_UNIX, SOCK_STREAM, 0, uwsgi.shared->mule_signal_pipe)) {
-			uwsgi_error("socketpair()");
-			exit(1);
-		}
-
-		if (socketpair(AF_UNIX, SOCK_DGRAM, 0, uwsgi.shared->mule_queue_pipe)) {
-			uwsgi_error("socketpair()");
-			exit(1);
-		}
+		create_signal_pipe(uwsgi.shared->mule_signal_pipe);
+		create_signal_pipe(uwsgi.shared->mule_queue_pipe);
 
 		for(i=0;i<uwsgi.mules_cnt;i++) {
 			// create the socket pipe
-        		if (socketpair(AF_UNIX, SOCK_STREAM, 0, uwsgi.mules[i].signal_pipe)) {
-                		uwsgi_error("socketpair()\n");
-        		}
-
-			if (socketpair(AF_UNIX, SOCK_DGRAM, 0, uwsgi.mules[i].queue_pipe)) {
-				uwsgi_error("socketpair()");
-				exit(1);
-			}
+			create_signal_pipe(uwsgi.mules[i].signal_pipe);
+			create_signal_pipe(uwsgi.mules[i].queue_pipe);
 
 			uwsgi.mules[i].id = i+1;
 
@@ -2330,15 +2316,8 @@ skipzero:
 			strncpy(uwsgi.farms[i].name, farm_value, 0xff);
 
 			// create the socket pipe
-        		if (socketpair(AF_UNIX, SOCK_STREAM, 0, uwsgi.farms[i].signal_pipe)) {
-                		uwsgi_error("socketpair()\n");
-        		}
-
-			if (socketpair(AF_UNIX, SOCK_DGRAM, 0, uwsgi.farms[i].queue_pipe)) {
-				uwsgi_error("socketpair()");
-				exit(1);
-			}
-
+			create_signal_pipe(uwsgi.farms[i].signal_pipe);
+			create_signal_pipe(uwsgi.farms[i].queue_pipe);
 
 			char *p = strtok(mules_list, ",");
 			while(p != NULL) {
@@ -2467,10 +2446,7 @@ skipzero:
 
 #ifdef UWSGI_SPOOLER
 	if (uwsgi.spool_dir != NULL && uwsgi.sockets) {
-		if (socketpair(AF_UNIX, SOCK_STREAM, 0, uwsgi.shared->spooler_signal_pipe)) {
-                        uwsgi_error("socketpair()\n");
-                        exit(1);
-                }
+		create_signal_pipe(uwsgi.shared->spooler_signal_pipe);
 		uwsgi.shared->spooler_pid = spooler_start();
 	}
 #endif
@@ -2499,11 +2475,7 @@ skipzero:
 	}
 	else {
 		// setup internal signalling system
-		if (socketpair(AF_UNIX, SOCK_STREAM, 0, uwsgi.shared->worker_signal_pipe)) {
-			uwsgi_error("socketpair()\n");
-			exit(1);
-		}
-
+		create_signal_pipe(uwsgi.shared->worker_signal_pipe);
 		uwsgi.signal_socket = uwsgi.shared->worker_signal_pipe[1];
 	}
 
@@ -3468,6 +3440,9 @@ static int manage_base_opt(int i, char *optarg) {
 		uwsgi.mules_cnt++;
 		uwsgi_string_new_list(&uwsgi.mules_patches, optarg);
 		return 1;
+	case LONG_ARGS_SIGNAL_BUFSIZE:
+		uwsgi.signal_bufsize = atoi(optarg);
+		return 1;
 	case LONG_ARGS_MULES:
 		uwsgi.master_process = 1;
 		for(i=0;i<atoi(optarg);i++) {
diff --git a/uwsgi.h b/uwsgi.h
index 3f2677cd..e4dae12f 100644
--- a/uwsgi.h
+++ b/uwsgi.h
@@ -538,6 +538,7 @@ struct uwsgi_opt {
 #define LONG_ARGS_PROCNAME_PREFIX_SP	17159
 #define LONG_ARGS_UMASK			17160
 #define LONG_ARGS_PAUSE			17161
+#define LONG_ARGS_SIGNAL_BUFSIZE	17162
 
 
 #define UWSGI_OK	0
@@ -1252,6 +1253,7 @@ struct uwsgi_server {
 	int lazy_respawned;
 
 	int buffer_size;
+	int signal_bufsize;
 
 	// post buffering
 	int post_buffering;
@@ -2441,14 +2443,19 @@ struct uwsgi_subscribe_slot {
         struct uwsgi_subscribe_slot *next;
 };
 
+void mule_send_msg(int, char *, size_t);
 
+void create_signal_pipe(int *);
 struct uwsgi_subscribe_slot *uwsgi_get_subscribe_slot(struct uwsgi_subscribe_slot **, char *, uint16_t, int);
 struct uwsgi_subscribe_node *uwsgi_get_subscribe_node(struct uwsgi_subscribe_slot **, char *, uint16_t, int);
 void uwsgi_remove_subscribe_node(struct uwsgi_subscribe_slot **, struct uwsgi_subscribe_node *);
 struct uwsgi_subscribe_node *uwsgi_add_subscribe_node(struct uwsgi_subscribe_slot **, struct uwsgi_subscribe_req *, int);
 
+ssize_t uwsgi_mule_get_msg(int, int, char *, size_t, int);
+
 uint8_t uwsgi_signal_wait(int);
 void uwsgi_add_app(int, uint8_t, char *, int);
+int uwsgi_signal_send(int, uint8_t);
 
 #ifdef UWSGI_CAP
 void uwsgi_build_cap(char *);