added probe infrastructure

master.c

@@ -825,6 +825,18 @@ healthy:
 					uwsgi_manage_command_cron(time(NULL));
 				}
 
+					
+				// check for probes
+				if (ushared->probes_cnt > 0) {
+					uwsgi_lock(uwsgi.probe_table_lock);
+					for(i=0;i<ushared->probes_cnt;i++) {
+						if (ushared->probes[i].func(interesting_fd, &ushared->probes[i])) { 
+                                               		uwsgi_route_signal(ushared->probes[i].sig);
+                                        	}
+					}
+					uwsgi_unlock(uwsgi.probe_table_lock);
+				}
+
 				if (rlen > 0) {
 
 					if (uwsgi.log_master) {
@@ -1086,6 +1098,7 @@ healthy:
 
 					next_iteration = 0;
 
+					uwsgi_lock(uwsgi.timer_table_lock);
 					for(i=0;i<ushared->timers_cnt;i++) {
                                                 if (ushared->timers[i].registered) {
                                                         if (interesting_fd == ushared->timers[i].fd) {
@@ -1096,6 +1109,7 @@ healthy:
                                                         }
                                                 }
                                         }
+					uwsgi_unlock(uwsgi.timer_table_lock);
                                         if (next_iteration) continue;
 
 

plugins/probeconnect/connectprobe.c

@@ -0,0 +1,88 @@
+#include "../../uwsgi.h"
+
+extern struct uwsgi_server uwsgi;
+
+int connect_prober_callback(int interesting_fd, struct uwsgi_signal_probe *up) {
+
+	// is this a timeout event ?
+	if (interesting_fd == -1) {
+		// am i wating for something ?
+		if (up->fd != -1) {
+			up->cycles++;
+			if (up->cycles > 3) {
+				// reset the cycle
+				up->cycles = 0;
+				close(up->fd);
+				up->fd = -1;
+				// state = NOOP
+				up->state = 0;
+				// avoid duplicated events
+				if (!up->bad) {
+					up->bad = 1;
+					return 1;
+				}
+			}
+		}
+		// ok register a new event
+		else {
+			up->fd = uwsgi_connect(up->args, -1, 1);
+			if (up->fd != -1) {
+				// status = CONNECTING
+				up->state = 1;
+				event_queue_add_fd_write(uwsgi.master_queue, up->fd);
+				return 0;
+			}
+			// signal the bad event (if not already bad)
+			if (!up->bad) {
+				up->bad = 1;
+				return 1;
+			}
+		}
+	}
+	else if (up->fd != -1) {
+		// is this event for me ?
+		if (interesting_fd == up->fd) {
+			// uselsess here (we have only one state), only to show a good practice
+			// check the state
+			if (up->state == 1) {
+				if (uwsgi_is_bad_connection(up->fd)) {
+					// signal the bad connection (if needed)
+					up->cycles = 0;
+					close(up->fd);
+					up->fd = -1;
+                                	// state = NOOP
+                                	up->state = 0;
+					if (!up->bad) {
+						up->bad = 1;
+						return 1;
+					}
+					return 0;
+				}
+				// this is a good connection
+				up->cycles = 0;
+				close(up->fd);
+				up->fd = -1;
+				// state = NOOP
+				up->state = 0;
+				if (up->bad) {
+					up->bad = 0;
+					return 1;
+				}
+			}
+		}
+	}
+
+	// default action
+	return 0;
+}
+
+int probeconnect_init() {
+
+	uwsgi_probe_register(&uwsgi.probes, "connect", connect_prober_callback);
+	return 0;
+}
+
+struct uwsgi_plugin probeconnect_plugin = {
+	
+	.init = probeconnect_init,
+};

plugins/probeconnect/uwsgiplugin.py

@@ -0,0 +1,7 @@
+
+NAME='probeconnect'
+CFLAGS = []
+LDFLAGS = []
+LIBS = []
+
+GCC_LIST = ['connectprobe']

plugins/python/uwsgi_pymodule.c

@@ -220,6 +220,23 @@ PyObject *py_uwsgi_add_cron(PyObject * self, PyObject * args) {
 	return Py_True;
 }
 
+
+PyObject *py_uwsgi_add_probe(PyObject * self, PyObject * args) {
+
+        uint8_t uwsgi_signal;
+        char *probe, *probe_args;
+
+        if (!PyArg_ParseTuple(args, "Bss:add_probe", &uwsgi_signal, &probe, &probe_args)) {
+                return NULL;
+        }
+
+        if (uwsgi_add_probe(uwsgi_signal, probe ,probe_args))
+                return PyErr_Format(PyExc_ValueError, "unable to add probe");
+
+        Py_INCREF(Py_None);
+        return Py_None;
+}
+
 	
 
 PyObject *py_uwsgi_add_timer(PyObject * self, PyObject * args) {
@@ -3011,6 +3028,7 @@ static PyMethodDef uwsgi_advanced_methods[] = {
 	{"signal_received", py_uwsgi_signal_received, METH_VARARGS, ""},
 	{"add_file_monitor", py_uwsgi_add_file_monitor, METH_VARARGS, ""},
 	{"add_timer", py_uwsgi_add_timer, METH_VARARGS, ""},
+	{"add_probe", py_uwsgi_add_probe, METH_VARARGS, ""},
 	{"add_rb_timer", py_uwsgi_add_rb_timer, METH_VARARGS, ""},
 	{"add_cron", py_uwsgi_add_cron, METH_VARARGS, ""},
 

signal.c

@@ -128,6 +128,76 @@ int uwsgi_add_file_monitor(uint8_t sig, char *filename) {
 
 }
 
+struct uwsgi_probe *uwsgi_probe_register(struct uwsgi_probe **up, char *name, int (*func)(int, struct uwsgi_signal_probe *)) {
+
+        struct uwsgi_probe *uwsgi_up = *up, *old_up;
+
+        if (!uwsgi_up) {
+                *up = uwsgi_malloc(sizeof(struct uwsgi_probe));
+                uwsgi_up = *up;
+        }
+        else {
+                while(uwsgi_up) {
+                        old_up = uwsgi_up;
+                        uwsgi_up = uwsgi_up->next;
+                }
+
+                uwsgi_up = uwsgi_malloc(sizeof(struct uwsgi_probe));
+                old_up->next = uwsgi_up;
+        }
+
+        uwsgi_up->name = name;
+        uwsgi_up->func = func;
+        uwsgi_up->next = NULL;
+
+	uwsgi_log("registered new probe \"%s\" at %p\n", name, uwsgi_up);
+
+        return uwsgi_up;
+}
+
+
+int uwsgi_add_probe(uint8_t sig, char *kind, char *args) {
+
+	uwsgi_lock(uwsgi.probe_table_lock);
+
+        if (ushared->probes_cnt < MAX_PROBES) {
+
+		struct uwsgi_probe *up = uwsgi.probes;
+		while(up) {
+			if (!strcmp(up->name, kind)) {
+				break;
+			}
+			up = up->next;
+		}
+
+		if (!up) {
+			uwsgi_log("unable to find probe \"%s\" !!!\n", kind);
+                	uwsgi_unlock(uwsgi.probe_table_lock);
+                	return -1;
+		}
+
+                // fill the probe table
+                ushared->probes[ushared->probes_cnt].func = up->func;
+		strncpy(ushared->probes[ushared->probes_cnt].args, args, 1024-1);
+                ushared->probes[ushared->probes_cnt].registered = 0;
+                ushared->probes[ushared->probes_cnt].sig = sig;
+		ushared->probes[ushared->probes_cnt].fd = -1;
+		ushared->probes[ushared->probes_cnt].state = 0;
+		ushared->probes[ushared->probes_cnt].cycles = 0;
+		ushared->probes[ushared->probes_cnt].bad = 0;
+                ushared->probes_cnt++;
+        }
+        else {
+                uwsgi_log("you can register max %d probes !!!\n", MAX_PROBES);
+                uwsgi_unlock(uwsgi.probe_table_lock);
+                return -1;
+        }
+
+        uwsgi_unlock(uwsgi.probe_table_lock);
+
+	return 0;
+}
+
 int uwsgi_add_timer(uint8_t sig, int secs) {
 
 	uwsgi_lock(uwsgi.timer_table_lock);

socket.c

@@ -1014,3 +1014,16 @@ void uwsgi_del_sockets_from_queue(int queue) {
 
 }
 
+int uwsgi_is_bad_connection(int fd) {
+
+	int soopt = 0;
+	socklen_t solen = sizeof(int) ;
+
+	if (getsockopt(fd, SOL_SOCKET, SO_ERROR, (void *) (&soopt), &solen) < 0) {
+		return -1;
+	}
+
+	// will be 0 if all ok
+	return soopt;
+}
+

uwsgi.c

@@ -1851,6 +1851,10 @@ int uwsgi_start(void *v_argv) {
 		uwsgi.timer_table_lock = uwsgi_mmap_shared_lock();
 		uwsgi_lock_init(uwsgi.timer_table_lock);
 
+		// probe table lock
+		uwsgi.probe_table_lock = uwsgi_mmap_shared_lock();
+		uwsgi_lock_init(uwsgi.probe_table_lock);
+
 		// rb_timer table lock
 		uwsgi.rb_timer_table_lock = uwsgi_mmap_shared_lock();
 		uwsgi_lock_init(uwsgi.rb_timer_table_lock);

uwsgi.h

@@ -39,6 +39,8 @@ extern "C" {
 #define MAX_GENERIC_PLUGINS 64
 #define MAX_RPC 64
 #define MAX_GATEWAYS 64
+#define MAX_TIMERS 64
+#define MAX_PROBES 64
 #define MAX_CRONS 64
 
 #ifndef UWSGI_LOAD_EMBEDDED_PLUGINS
@@ -966,6 +968,29 @@ struct uwsgi_signal_rb_timer {
 	struct uwsgi_rb_timer *uwsgi_rb_timer;
 };
 
+struct uwsgi_signal_probe {
+	
+	int (*func)(int, struct uwsgi_signal_probe *);
+	char args[1024];
+
+	int fd;
+	int state;
+	int bad;
+	uint64_t cycles;
+
+	int registered;
+	uint8_t sig;
+};
+
+struct uwsgi_probe {
+
+	char *name;
+	int (*func)(int, struct uwsgi_signal_probe *);
+
+	struct uwsgi_probe *next;
+};
+
+
 struct uwsgi_server {
 
 
@@ -1159,6 +1184,8 @@ struct uwsgi_server {
 	int build_mime_dict;
 	char *mime_file;
 
+	struct uwsgi_probe *probes;
+
 	struct uwsgi_daemon *daemons;
 	int daemons_cnt;
 
@@ -1505,6 +1532,7 @@ struct uwsgi_server {
 	void *signal_table_lock;
 	void *fmon_table_lock;
 	void *timer_table_lock;
+	void *probe_table_lock;
 	void *rb_timer_table_lock;
 	void *cron_table_lock;
 	void *rpc_table_lock;
@@ -1651,10 +1679,13 @@ struct uwsgi_shared {
 	struct uwsgi_fmon files_monitored[64];
 	int files_monitored_cnt;
 
-	struct uwsgi_timer timers[64];
+	struct uwsgi_signal_probe probes[MAX_PROBES];
+	int probes_cnt;
+
+	struct uwsgi_timer timers[MAX_TIMERS];
 	int timers_cnt;
 
-	struct uwsgi_signal_rb_timer rb_timers[64];
+	struct uwsgi_signal_rb_timer rb_timers[MAX_TIMERS];
 	int rb_timers_cnt;
 
 	struct uwsgi_rpc rpc_table[MAX_RPC];
@@ -2473,6 +2504,11 @@ void uwsgi_subscribe(char *, uint8_t);
 
 struct uwsgi_daemon *uwsgi_daemon_new(struct uwsgi_daemon **, char *);
 
+struct uwsgi_probe *uwsgi_probe_register(struct uwsgi_probe **, char *, int (*)(int, struct uwsgi_signal_probe *));
+int uwsgi_add_probe(uint8_t sig, char *, char *);
+
+int uwsgi_is_bad_connection(int);
+
 #ifdef __linux__
 #ifdef MADV_MERGEABLE
 void uwsgi_linux_ksm_map(void);