From 571029cfd3654e8ce49df2e2d08cac8e8316668a Mon Sep 17 00:00:00 2001 From: "roberto@oneiric64" Date: Sat, 26 Nov 2011 11:05:49 +0100 Subject: [PATCH] added probe infrastructure --- master.c | 14 +++++ plugins/probeconnect/connectprobe.c | 88 +++++++++++++++++++++++++++++ plugins/probeconnect/uwsgiplugin.py | 7 +++ plugins/python/uwsgi_pymodule.c | 18 ++++++ signal.c | 70 +++++++++++++++++++++++ socket.c | 13 +++++ uwsgi.c | 4 ++ uwsgi.h | 40 ++++++++++++- 8 files changed, 252 insertions(+), 2 deletions(-) create mode 100644 plugins/probeconnect/connectprobe.c create mode 100644 plugins/probeconnect/uwsgiplugin.py diff --git a/master.c b/master.c index e595f926..75dda669 100644 --- a/master.c +++ b/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;iprobes_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;itimers_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; diff --git a/plugins/probeconnect/connectprobe.c b/plugins/probeconnect/connectprobe.c new file mode 100644 index 00000000..d80b0358 --- /dev/null +++ b/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, +}; diff --git a/plugins/probeconnect/uwsgiplugin.py b/plugins/probeconnect/uwsgiplugin.py new file mode 100644 index 00000000..d42968a0 --- /dev/null +++ b/plugins/probeconnect/uwsgiplugin.py @@ -0,0 +1,7 @@ + +NAME='probeconnect' +CFLAGS = [] +LDFLAGS = [] +LIBS = [] + +GCC_LIST = ['connectprobe'] diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index b6f8e924..8e5b8a9c 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/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, ""}, diff --git a/signal.c b/signal.c index 605037c7..04e0c57a 100644 --- a/signal.c +++ b/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); diff --git a/socket.c b/socket.c index e98ae898..05c79645 100644 --- a/socket.c +++ b/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; +} + diff --git a/uwsgi.c b/uwsgi.c index e2c7aecd..e9e04313 100644 --- a/uwsgi.c +++ b/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); diff --git a/uwsgi.h b/uwsgi.h index 92189219..d4cb084a 100644 --- a/uwsgi.h +++ b/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);