diff --git a/master.c b/master.c
index 910e0472..edc06d7d 100644
--- a/master.c
+++ b/master.c
@@ -67,56 +67,7 @@ void expire_rb_timeouts(struct rb_root *root) {
         }
 }
 
-void uwsgi_send_subscription(char *udp_address, char *key, size_t keysize, char *modifier1, size_t modifier1_len) {
-	char *ssb;
-	char subscrbuf[4096];
-
-	uint16_t ustrlen;
-
-	ssb = subscrbuf;
-
-        ustrlen = 3;
-        *ssb++ = (uint8_t) (ustrlen  & 0xff);
-        *ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
-        memcpy(ssb, "key", ustrlen);
-        ssb+=ustrlen;
-
-        ustrlen = keysize;
-        *ssb++ = (uint8_t) (ustrlen  & 0xff);
-        *ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
-        memcpy(ssb, key, ustrlen);
-        ssb+=ustrlen;
-
-        ustrlen = 7;
-        *ssb++ = (uint8_t) (ustrlen  & 0xff);
-        *ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
-        memcpy(ssb, "address", ustrlen);
-        ssb+=ustrlen;
-
-        ustrlen = strlen(uwsgi.sockets->name);
-        *ssb++ = (uint8_t) (ustrlen  & 0xff);
-        *ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
-        memcpy(ssb, uwsgi.sockets->name, ustrlen);
-        ssb+=ustrlen;
-
-	if (modifier1) {
-		ustrlen = 9;
-        	*ssb++ = (uint8_t) (ustrlen  & 0xff);
-        	*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
-        	memcpy(ssb, "modifier1", ustrlen);
-        	ssb+=ustrlen;
-
-        	ustrlen = modifier1_len;
-        	*ssb++ = (uint8_t) (ustrlen  & 0xff);
-        	*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
-        	memcpy(ssb, modifier1, ustrlen);
-        	ssb+=ustrlen;
-	}
-
-        send_udp_message(224, udp_address, subscrbuf, ssb-subscrbuf);
-}
-
-void uwsgi_subscribe(char *subscription) {
+void uwsgi_subscribe(char *subscription, uint8_t cmd) {
 
 	int subfile_size;
 	int i;
@@ -149,7 +100,7 @@ void uwsgi_subscribe(char *subscription) {
                         					modifier1_len = strlen(modifier1);
 								keysize = strlen(key);
                 					}
-							uwsgi_send_subscription(udp_address, key, keysize, modifier1, modifier1_len);
+							uwsgi_send_subscription(udp_address, key, keysize, modifier1, modifier1_len, cmd);
 							modifier1 = NULL;
 							modifier1_len = 0;
 						}
@@ -167,7 +118,7 @@ void uwsgi_subscribe(char *subscription) {
                         					modifier1_len = strlen(modifier1);
 								keysize = strlen(key);
                 					}
-							uwsgi_send_subscription(udp_address, key, keysize, modifier1, modifier1_len);
+							uwsgi_send_subscription(udp_address, key, keysize, modifier1, modifier1_len, cmd);
 							modifier1 = NULL;
 							modifier1_len = 0;
 							lines[i] = '\n';
@@ -191,7 +142,7 @@ void uwsgi_subscribe(char *subscription) {
 			modifier1_len = strlen(modifier1);
 		}
 
-		uwsgi_send_subscription(udp_address, subscription_key+1, strlen(subscription_key+1), modifier1, modifier1_len);
+		uwsgi_send_subscription(udp_address, subscription_key+1, strlen(subscription_key+1), modifier1, modifier1_len, cmd);
 		if (modifier1)
 			modifier1[-1] = ',';
 	}
@@ -506,7 +457,7 @@ int master_loop(char **argv, char **environ) {
 	// first subscription
 	struct uwsgi_string_list *subscriptions = uwsgi.subscriptions;
 	while(subscriptions) {
-		uwsgi_subscribe(subscriptions->value);
+		uwsgi_subscribe(subscriptions->value, 0);
 		subscriptions = subscriptions->next;
 	}
 
@@ -1424,7 +1375,7 @@ healthy:
 			if (uwsgi.subscriptions && ((uwsgi.master_cycles % 10) == 0 || uwsgi.master_cycles == 1)) {
 				struct uwsgi_string_list *subscriptions = uwsgi.subscriptions;
 				while(subscriptions) {
-					uwsgi_subscribe(subscriptions->value);
+					uwsgi_subscribe(subscriptions->value, 0);
 					subscriptions = subscriptions->next;
 				}
 			}
diff --git a/plugins/fastrouter/fastrouter.c b/plugins/fastrouter/fastrouter.c
index e339e1c9..2acd4d50 100644
--- a/plugins/fastrouter/fastrouter.c
+++ b/plugins/fastrouter/fastrouter.c
@@ -73,6 +73,17 @@ struct uwsgi_fastrouter {
 	int i_am_cheap;
 } ufr;
 
+static void fastrouter_go_cheap(void) {
+
+	uwsgi_log("[uwsgi-fastrouter] no more nodes available. Going cheap...\n");
+        struct uwsgi_fastrouter_socket *ufr_sock = ufr.sockets;
+        while(ufr_sock) {
+        	event_queue_del_fd(ufr.queue, ufr_sock->fd, event_queue_read());
+                ufr_sock = ufr_sock->next;
+        }
+        ufr.i_am_cheap = 1;
+}
+
 
 static struct uwsgi_fastrouter_socket *uwsgi_fastrouter_new_socket(char *name) {
 
@@ -165,41 +176,28 @@ struct fastrouter_session {
 
 static void close_session(struct fastrouter_session **fr_table, struct fastrouter_session *fr_session) {
 
-	// check timeout expired
-	if (fr_session == NULL) {
-		if (ufr.subscription_server) {
-			//time_t current_time = time(NULL);
-			uwsgi_log("checking for node health\n");
-			struct uwsgi_subscribe_slot *slot = ufr.subscriptions;
-			while(slot) {
-				struct uwsgi_subscribe_node *node = slot->nodes;
-				while(node) {
-					uwsgi_log("%.*s (hits: %llu) %.*s\n", slot->keylen, slot->key, slot->hits, node->len, node->name);
-					node = node->next;
-				}
-				slot = slot->next;
-			}
-			del_check_timeout(ufr.subscriptions_check);
-			ufr.subscriptions_check = add_check_timeout(10);
-		}
-		return;
-	}
-
 	close(fr_session->fd);
 	fr_table[fr_session->fd] = NULL;
 	if (fr_session->instance_fd != -1) {
-		if (ufr.subscriptions && (fr_session->instance_failed || fr_session->status == FASTROUTER_STATUS_CONNECTING)) {
-			if (fr_session->un && fr_session->un->len > 0) {
-				uwsgi_log("[uwsgi-fastrouter] %.*s => marking %.*s as failed\n", (int) fr_session->hostname_len, fr_session->hostname, (int) fr_session->instance_address_len,fr_session->instance_address);
-                        	uwsgi_remove_subscribe_node(&ufr.subscriptions, fr_session->un);
+		if (ufr.subscriptions && fr_session->un && fr_session->un->len > 0) {
+			// decrease reference count
+#ifdef UWSGI_DEBUG
+			uwsgi_log("[1] node %.*s refcnt: %llu\n", fr_session->un->len, fr_session->un->name, fr_session->un->reference);
+#endif
+			fr_session->un->reference--;
+#ifdef UWSGI_DEBUG
+			uwsgi_log("[2] node %.*s refcnt: %llu\n", fr_session->un->len, fr_session->un->name, fr_session->un->reference);
+#endif
+			if (fr_session->instance_failed || fr_session->status == FASTROUTER_STATUS_CONNECTING) {
+				if (fr_session->un->death_mark == 0)
+					uwsgi_log("[uwsgi-fastrouter] %.*s => marking %.*s as failed\n", (int) fr_session->hostname_len, fr_session->hostname, (int) fr_session->instance_address_len,fr_session->instance_address);
+				fr_session->un->death_mark = 1;
+				// check if i can remove the node
+				if (fr_session->un->reference == 0) {
+                        		uwsgi_remove_subscribe_node(&ufr.subscriptions, fr_session->un);
+				}
                                 if (ufr.subscriptions == NULL && ufr.cheap && !ufr.i_am_cheap) {
-                                	uwsgi_log("[uwsgi-fastrouter] no more nodes available. Going cheap...\n");
-                                        struct uwsgi_fastrouter_socket *ufr_sock = ufr.sockets;
-                                        while(ufr_sock) {
-                                        	event_queue_del_fd(ufr.queue, ufr_sock->fd, event_queue_read());
-                                                ufr_sock = ufr_sock->next;
-                                        }
-					ufr.i_am_cheap = 1;
+					fastrouter_go_cheap();
                                 }
 
 			}
@@ -437,15 +435,34 @@ void fastrouter_loop() {
 				if (len > 0) {
 					memset(&usr, 0, sizeof(struct uwsgi_subscribe_req));
 					uwsgi_hooked_parse(bbuf+4, len-4, fastrouter_manage_subscription, &usr);
-					if (uwsgi_add_subscribe_node(&ufr.subscriptions, &usr, ufr.subscription_regexp) && ufr.i_am_cheap) {
-						struct uwsgi_fastrouter_socket *ufr_sock = ufr.sockets;
-                                                while(ufr_sock) {
-                                                	event_queue_add_fd_read(ufr.queue, ufr_sock->fd);
-                                                        ufr_sock = ufr_sock->next;
-                                                }
-						ufr.i_am_cheap = 0;
-						uwsgi_log("[uwsgi-fastrouter] leaving cheap mode...\n");	
+					// subscribe request ?
+					if (bbuf[3] == 0) {
+						if (uwsgi_add_subscribe_node(&ufr.subscriptions, &usr, ufr.subscription_regexp) && ufr.i_am_cheap) {
+							struct uwsgi_fastrouter_socket *ufr_sock = ufr.sockets;
+                                                	while(ufr_sock) {
+                                                		event_queue_add_fd_read(ufr.queue, ufr_sock->fd);
+                                                        	ufr_sock = ufr_sock->next;
+                                                	}
+							ufr.i_am_cheap = 0;
+							uwsgi_log("[uwsgi-fastrouter] leaving cheap mode...\n");	
+						}
 					}
+					//unsubscribe 
+					else {
+						struct uwsgi_subscribe_node *node = uwsgi_get_subscribe_node_by_name(&ufr.subscriptions, usr.key, usr.keylen, usr.address, usr.address_len, ufr.subscription_regexp);	
+						if (node && node->len) {
+							if (node->death_mark == 0)
+								uwsgi_log("[uwsgi-fastrouter] %.*s => marking %.*s as failed\n", (int) usr.keylen, usr.key, (int) usr.address_len, usr.address);
+							node->death_mark = 1;
+                                			// check if i can remove the node
+                                			if (node->reference == 0) {
+                                        			uwsgi_remove_subscribe_node(&ufr.subscriptions, node);
+                                			}
+                                			if (ufr.subscriptions == NULL && ufr.cheap && !ufr.i_am_cheap) {
+								fastrouter_go_cheap();
+                                			}
+						}
+					}	
 				}
 			}
 			else {
@@ -524,6 +541,9 @@ void fastrouter_loop() {
 									fr_session->instance_address_len = fr_session->un->len;
 									fr_session->modifier1 = fr_session->un->modifier1;
 								}
+								else if (ufr.subscriptions == NULL && ufr.cheap && !ufr.i_am_cheap) {
+									fastrouter_go_cheap();
+                                				}
 							}
 							else if (ufr.base) {
 								tmp_socket_name = uwsgi_concat2nn(ufr.base, ufr.base_len, fr_session->hostname, fr_session->hostname_len, &tmp_socket_name_len);
@@ -558,23 +578,6 @@ void fastrouter_loop() {
 							if (tmp_socket_name) free(tmp_socket_name);
 
 							if (fr_session->instance_fd < 0) {
-								/*
-								if (ufr.subscription_server) {
-									if (fr_session->un && fr_session->un->len > 0) { 
-	                                                        		uwsgi_log("[uwsgi-fastrouter] %.*s => marking %.*s as failed\n", (int) fr_session->hostname_len, fr_session->hostname, (int) fr_session->instance_address_len,fr_session->instance_address);
-                                						uwsgi_remove_subscribe_node(&ufr.subscriptions, fr_session->un);	 
-										if (ufr.subscriptions == NULL && ufr.cheap && !ufr.i_am_cheap) {
-											uwsgi_log("[uwsgi-fastrouter] no more nodes available. Going cheap...\n");
-											struct uwsgi_fastrouter_socket *ufr_sock = ufr.sockets;	
-											while(ufr_sock) {
-												event_queue_del_fd(ufr.queue, ufr_sock->fd, event_queue_read());	
-												ufr_sock = ufr_sock->next;
-											}
-											ufr.i_am_cheap = 1;
-										}
-                        						}
-	                                                        }
-								*/
 								fr_session->instance_failed = 1;
 								close_session(fr_table, fr_session);
                                                         	break;
diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c
index 201319fb..478309fb 100644
--- a/plugins/python/uwsgi_pymodule.c
+++ b/plugins/python/uwsgi_pymodule.c
@@ -1942,13 +1942,18 @@ PyObject *py_uwsgi_load_plugin(PyObject * self, PyObject * args) {
 PyObject *py_uwsgi_multicast(PyObject * self, PyObject * args) {
 
 	char *host, *message;
+	Py_ssize_t message_len;
 	ssize_t ret;
+	char *uwsgi_message;
 
-	if (!PyArg_ParseTuple(args, "ss:send_multicast_message", &host, &message)) {
+	if (!PyArg_ParseTuple(args, "ss#:send_multicast_message", &host, &message, &message_len)) {
 		return NULL;
 	}
 
-	ret = send_udp_message(UWSGI_MODIFIER_MULTICAST, host, message, strlen(message));
+	uwsgi_message = uwsgi_malloc(message_len+4);
+	memcpy(uwsgi_message+4, message, message_len);
+	ret = send_udp_message(UWSGI_MODIFIER_MULTICAST, 0, host, uwsgi_message, message_len);
+	free(uwsgi_message);
 
 	if (ret <= 0) {
 		Py_INCREF(Py_None);
diff --git a/protocol.c b/protocol.c
index f6aeae09..b4cd5d28 100644
--- a/protocol.c
+++ b/protocol.c
@@ -123,16 +123,21 @@ time_t parse_http_date(char *date, uint16_t len) {
 }
 
 #ifdef UWSGI_UDP
-ssize_t send_udp_message(uint8_t modifier1, char *host, char *message, uint16_t message_size) {
+ssize_t send_udp_message(uint8_t modifier1, uint8_t modifier2, char *host, char *message, uint16_t message_size) {
 
 	int fd;
 	struct sockaddr_in udp_addr;
 	char *udp_port;
 	ssize_t ret;
-	char udpbuff[1024];
 
-	if (message_size + 4 > 1024)
-		return -1;
+	struct uwsgi_header *uh;
+
+	if (message) {
+		uh = (struct uwsgi_header *) message;
+	}
+	else {
+		uh = (struct uwsgi_header *) uwsgi_malloc(4);
+	}
 
 	udp_port = strchr(host, ':');
 	if (udp_port == NULL) {
@@ -152,22 +157,15 @@ ssize_t send_udp_message(uint8_t modifier1, char *host, char *message, uint16_t
 	udp_addr.sin_port = htons(atoi(udp_port+1));
 	udp_addr.sin_addr.s_addr = inet_addr(host);
 
-	udpbuff[0] = modifier1;
+	uh->modifier1 = modifier1;
 #ifdef __BIG_ENDIAN__
-	message_size = uwsgi_swap16(message_size);
+	uh->pktsize = uwsgi_swap16(message_size);
+#else
+	uh->pktsize = message_size;
 #endif
+	uh->modifier2 = modifier2;
 
-	memcpy(udpbuff+1, &message_size, 2);
-
-	udpbuff[3] = 0;
-
-#ifdef __BIG_ENDIAN__
-	message_size = uwsgi_swap16(message_size);
-#endif
-
-	memcpy(udpbuff+4, message, message_size);
-
-	ret = sendto(fd, udpbuff, message_size+4, 0, (struct sockaddr *) &udp_addr, sizeof(udp_addr));
+	ret = sendto(fd, (char *) uh, message_size+4, 0, (struct sockaddr *) &udp_addr, sizeof(udp_addr));
 	if (ret < 0) {
 		uwsgi_error("sendto()");
 	}
@@ -175,6 +173,10 @@ ssize_t send_udp_message(uint8_t modifier1, char *host, char *message, uint16_t
 
 	udp_port[0] = ':';
 
+	if ((char *)uh != message) {
+		free(uh);
+	}
+
 	return ret;
 	
 }
diff --git a/subscription.c b/subscription.c
index d64d408d..4f9df0e7 100644
--- a/subscription.c
+++ b/subscription.c
@@ -14,8 +14,12 @@
 	This system is not mean to run on shared memory. If you have multiple processes for the same app, you have to create
 	a new subscriptions slot list.
 
+	To avoid removal of already using nodes, a reference count system has been implemented
+
 */
 
+extern struct uwsgi_server uwsgi;
+
 struct uwsgi_subscribe_slot *uwsgi_get_subscribe_slot(struct uwsgi_subscribe_slot **slot, char *key, uint16_t keylen, int regexp) {
 
 	struct uwsgi_subscribe_slot *current_slot = *slot;
@@ -72,23 +76,61 @@ struct uwsgi_subscribe_node *uwsgi_get_subscribe_node(struct uwsgi_subscribe_slo
 	if (current_slot) {
 		// node found, move up in the list increasing hits
 		current_slot->hits++;
+		time_t current = time(NULL);
 		struct uwsgi_subscribe_node *node = current_slot->nodes;
-		while(node) {
+		while(current_slot && node) {
+			// is the node alive ?
+			if (current - node->last_check > 10) {
+				node->death_mark = 1;
+			}
+			if (node->death_mark && node->reference == 0) {
+				// remove the node and move to next
+				struct uwsgi_subscribe_node *dead_node = node;
+				node = node->next;
+				// if the slot has been removed, return NULL;
+				if (uwsgi_remove_subscribe_node(slot, dead_node) == 1) {
+					return NULL;
+				}
+				continue;
+			}
 			if (rr_pos == current_slot->rr) {
 				current_slot->rr++;
+				node->reference++;
 				return node;
 			}
 			node = node->next;
 			rr_pos++;
 		}
 		current_slot->rr = 0;
+		if (current_slot->nodes) {
+			current_slot->nodes->reference++;
+		}
 		return current_slot->nodes;
 	}
 
 	return NULL;
 }
 
-void uwsgi_remove_subscribe_node(struct uwsgi_subscribe_slot **slot, struct uwsgi_subscribe_node *node) {
+struct uwsgi_subscribe_node *uwsgi_get_subscribe_node_by_name(struct uwsgi_subscribe_slot **slot, char *key, uint16_t keylen, char *val, uint16_t vallen, int regexp) {
+
+	if (keylen > 0xff) return NULL;
+	struct uwsgi_subscribe_slot *current_slot = uwsgi_get_subscribe_slot(slot, key, keylen, regexp);
+	if (current_slot) {
+		struct uwsgi_subscribe_node *node = current_slot->nodes;
+		while(node) {
+			if (!uwsgi_strncmp(val, vallen, node->name, node->len)) {
+				return node;
+			}			
+			node = node->next;
+		}
+	}
+
+	return NULL;
+}
+
+int uwsgi_remove_subscribe_node(struct uwsgi_subscribe_slot **slot, struct uwsgi_subscribe_node *node) {
+
+	int ret = 0;
 
 	struct uwsgi_subscribe_node *a_node;
 	struct uwsgi_subscribe_slot *node_slot = node->slot;
@@ -132,12 +174,15 @@ void uwsgi_remove_subscribe_node(struct uwsgi_subscribe_slot **slot, struct uwsg
 		}
 #endif
 
+		ret = 1;
 		free(node_slot);
 		// am i the only slot ?
 		if (!prev_slot && !next_slot) {
 			*slot = NULL;
 		}
 	}
+
+	return ret;
 }
 
 struct uwsgi_subscribe_node *uwsgi_add_subscribe_node(struct uwsgi_subscribe_slot **slot, struct uwsgi_subscribe_req *usr, int regexp) {
@@ -151,6 +196,8 @@ struct uwsgi_subscribe_node *uwsgi_add_subscribe_node(struct uwsgi_subscribe_slo
 		node = current_slot->nodes;
 		while(node) {
                         if (!uwsgi_strncmp(node->name, node->len, usr->address, usr->address_len)) {
+				// remove death mark
+				node->death_mark = 0;
                                 node->last_check = time(NULL);
                                 return node;
                         }
@@ -162,6 +209,8 @@ struct uwsgi_subscribe_node *uwsgi_add_subscribe_node(struct uwsgi_subscribe_slo
 		node->len = usr->address_len;
 		node->modifier1 = usr->modifier1;
 		node->modifier2 = usr->modifier2;
+		node->reference = 0;
+		node->death_mark = 0;
 		node->last_check = time(NULL);
 		node->slot = current_slot;
                 memcpy(node->name, usr->address, usr->address_len);
@@ -195,6 +244,8 @@ struct uwsgi_subscribe_node *uwsgi_add_subscribe_node(struct uwsgi_subscribe_slo
 		current_slot->nodes = uwsgi_malloc(sizeof(struct uwsgi_subscribe_node));
 		current_slot->nodes->slot = current_slot;
 		current_slot->nodes->len = usr->address_len;
+		current_slot->nodes->reference = 0;
+		current_slot->nodes->death_mark = 0;
 		current_slot->nodes->modifier1 = usr->modifier1;
 		current_slot->nodes->modifier2 = usr->modifier2;
 		memcpy(current_slot->nodes->name, usr->address, usr->address_len);
@@ -272,3 +323,59 @@ struct uwsgi_subscribe_node *uwsgi_add_subscribe_node(struct uwsgi_subscribe_slo
 }
 
 
+void uwsgi_send_subscription(char *udp_address, char *key, size_t keysize, char *modifier1, size_t modifier1_len, uint8_t cmd) {
+
+	size_t ssb_size = 4 + (2 + 3) + (2 + keysize) + (2 + 7) + (2 + strlen(uwsgi.sockets->name));
+
+	if (modifier1) {
+		ssb_size += (2 + 9) + (2 + modifier1_len);
+	}
+
+        char *subscrbuf = uwsgi_malloc(ssb_size);
+	// leave space for uwsgi header
+        char *ssb = subscrbuf+4;
+
+	// key = "domain"
+        uint16_t ustrlen = 3;
+        *ssb++ = (uint8_t) (ustrlen  & 0xff);
+        *ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
+        memcpy(ssb, "key", ustrlen);
+        ssb+=ustrlen;
+
+        ustrlen = keysize;
+        *ssb++ = (uint8_t) (ustrlen  & 0xff);
+        *ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
+        memcpy(ssb, key, ustrlen);
+        ssb+=ustrlen;
+
+	// address = "first uwsgi socket"
+        ustrlen = 7;
+        *ssb++ = (uint8_t) (ustrlen  & 0xff);
+        *ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
+        memcpy(ssb, "address", ustrlen);
+        ssb+=ustrlen;
+
+        ustrlen = strlen(uwsgi.sockets->name);
+        *ssb++ = (uint8_t) (ustrlen  & 0xff);
+        *ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
+        memcpy(ssb, uwsgi.sockets->name, ustrlen);
+        ssb+=ustrlen;
+
+	// modifier1 = "modifier1"
+        if (modifier1) {
+                ustrlen = 9;
+                *ssb++ = (uint8_t) (ustrlen  & 0xff);
+                *ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
+                memcpy(ssb, "modifier1", ustrlen);
+                ssb+=ustrlen;
+
+                ustrlen = modifier1_len;
+                *ssb++ = (uint8_t) (ustrlen  & 0xff);
+                *ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
+                memcpy(ssb, modifier1, ustrlen);
+                ssb+=ustrlen;
+        }
+
+        send_udp_message(224, cmd, udp_address, subscrbuf, ssb_size-4);
+}
+
diff --git a/uwsgi.c b/uwsgi.c
index 5f6b2274..acabf185 100644
--- a/uwsgi.c
+++ b/uwsgi.c
@@ -523,6 +523,15 @@ void kill_them_all(int signum) {
 	}
 
 	uwsgi_log("SIGINT/SIGQUIT received...killing workers...\n");
+
+	// unsubscribe if needed
+	struct uwsgi_string_list *subscriptions = uwsgi.subscriptions;
+        while(subscriptions) {
+                uwsgi_subscribe(subscriptions->value, 1);
+                subscriptions = subscriptions->next;
+        }
+
+
 	for (i = 1; i <= uwsgi.numproc; i++) {
 		if (uwsgi.workers[i].pid > 0)
 			kill(uwsgi.workers[i].pid, SIGINT);
@@ -608,6 +617,14 @@ void grace_them_all(int signum) {
 
 
 	uwsgi_log("...gracefully killing workers...\n");
+
+	// unsubscribe if needed
+	struct uwsgi_string_list *subscriptions = uwsgi.subscriptions;
+        while(subscriptions) {
+                uwsgi_subscribe(subscriptions->value, 1);
+                subscriptions = subscriptions->next;
+        }
+
 	for (i = 1; i <= uwsgi.numproc; i++) {
 		if (uwsgi.auto_snapshot) {
 			if (uwsgi.workers[i].snapshot > 0) {
@@ -689,6 +706,14 @@ void reap_them_all(int signum) {
 		return;
 
 	uwsgi_log("...brutally killing workers...\n");
+
+	// unsubscribe if needed
+	struct uwsgi_string_list *subscriptions = uwsgi.subscriptions;
+        while(subscriptions) {
+                uwsgi_subscribe(subscriptions->value, 1);
+                subscriptions = subscriptions->next;
+        }
+
 	for (i = 1; i <= uwsgi.numproc; i++) {
 		if (uwsgi.workers[i].pid > 0)
 			kill(uwsgi.workers[i].pid, SIGTERM);
@@ -2963,7 +2988,7 @@ static int manage_base_opt(int i, char *optarg) {
 		return 1;
 #ifdef UWSGI_UDP
 	case LONG_ARGS_CLUSTER_RELOAD:
-		send_udp_message(98, optarg, "", 0);
+		send_udp_message(98, 0, optarg, NULL, 0);
 		break;
 	case LONG_ARGS_CLUSTER_LOG:
 		uwsgi_stdin_sendto(optarg, 96, 0);
@@ -4013,8 +4038,9 @@ void uwsgi_stdin_sendto(char *socket_name, uint8_t modifier1, uint8_t modifier2)
 
 	char buf[4096];
 	ssize_t rlen;
-	size_t delta = 4096;
-	char *ptr = buf;
+	size_t delta = 4096-4;
+	// leave space for uwsgi header
+	char *ptr = buf+4;
 
 	rlen = read(0, ptr, delta);
 	while (rlen > 0) {
@@ -4026,9 +4052,9 @@ void uwsgi_stdin_sendto(char *socket_name, uint8_t modifier1, uint8_t modifier2)
 		rlen = read(0, ptr, delta);
 	}
 
-	if (ptr > buf) {
-		send_udp_message(modifier1, socket_name, buf, ptr - buf);
-		uwsgi_log("sent string \"%.*s\" to cluster node %s", ptr - buf, buf, socket_name);
+	if (ptr > buf+4) {
+		send_udp_message(modifier1, modifier2, socket_name, buf, (ptr - buf)-4);
+		uwsgi_log("sent string \"%.*s\" to cluster node %s", (ptr - buf)-4, buf+4, socket_name);
 	}
 
 }
diff --git a/uwsgi.h b/uwsgi.h
index 9a5e6c22..d66c47b2 100644
--- a/uwsgi.h
+++ b/uwsgi.h
@@ -1857,7 +1857,7 @@ uint64_t uwsgi_swap64(uint64_t);
 #endif
 
 #ifdef UWSGI_UDP
-ssize_t send_udp_message(uint8_t, char *, char *, uint16_t);
+ssize_t send_udp_message(uint8_t, uint8_t, char *, char *, uint16_t);
 #endif
 
 int uwsgi_parse_packet(struct wsgi_request *, int);
@@ -2428,6 +2428,9 @@ struct uwsgi_subscribe_node {
 	uint64_t requests;
 	uint64_t transferred;
 
+	int death_mark;
+	uint64_t reference;
+
 	struct uwsgi_subscribe_slot *slot;
 
         struct uwsgi_subscribe_node *next;
@@ -2458,8 +2461,9 @@ 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_by_name(struct uwsgi_subscribe_slot **, char *, uint16_t, 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 *);
+int 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);
@@ -2476,6 +2480,10 @@ void manage_cluster_announce(char *, uint16_t, char *, uint16_t, void *);
 int uwsgi_read_response(int, struct uwsgi_header *, int, char **);
 char *uwsgi_simple_file_read(char *);
 
+void uwsgi_send_subscription(char *, char *, size_t , char *, size_t, uint8_t);
+
+void uwsgi_subscribe(char *, uint8_t);
+
 #ifdef __linux__
 #ifdef MADV_MERGEABLE
 void uwsgi_linux_ksm_map(void);