clearlinux/uwsgi
2
0
Fork 0
mirror of https://github.com/clearlinux/uwsgi.git synced 2026-09-06 21:51:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

added postgresql probe

Browse Source
This commit is contained in:
roberto@oneiric64
2011-11-26 12:44:38 +01:00
parent 571029cfd3
commit a1564a44fa
7 changed files with 185 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
master.c
+4
View File
@@ -830,6 +830,10 @@ healthy:
if (ushared->probes_cnt > 0) {
uwsgi_lock(uwsgi.probe_table_lock);
for(i=0;i<ushared->probes_cnt;i++) {
if (interesting_fd == -1) {
// increment cycles
ushared->probes[i].cycles++;
}
if (ushared->probes[i].func(interesting_fd, &ushared->probes[i])) {
uwsgi_route_signal(ushared->probes[i].sig);
}
plugins/probeconnect/connectprobe.c
+18 -16
View File
@@ -8,8 +8,7 @@ int connect_prober_callback(int interesting_fd, struct uwsgi_signal_probe *up) {
if (interesting_fd == -1) {
// am i wating for something ?
if (up->fd != -1) {
up->cycles++;
if (up->cycles > 3) {
if (up->cycles > (uint64_t) up->timeout) {
// reset the cycle
up->cycles = 0;
close(up->fd);
@@ -25,17 +24,20 @@ int connect_prober_callback(int interesting_fd, struct uwsgi_signal_probe *up) {
}
// 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;
if ((up->cycles % up->freq) == 0) {
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;
}
}
}
}
@@ -50,8 +52,8 @@ int connect_prober_callback(int interesting_fd, struct uwsgi_signal_probe *up) {
up->cycles = 0;
close(up->fd);
up->fd = -1;
// state = NOOP
up->state = 0;
// state = NOOP
up->state = 0;
if (!up->bad) {
up->bad = 1;
return 1;
@@ -83,6 +85,6 @@ int probeconnect_init() {
}
struct uwsgi_plugin probeconnect_plugin = {
.init = probeconnect_init,
};
plugins/probepg/pgprobe.c
+132
View File
@@ -0,0 +1,132 @@
#include "../../uwsgi.h"
#include <libpq-fe.h>
extern struct uwsgi_server uwsgi;
int pg_prober_callback(int, struct uwsgi_signal_probe *);
int probepg_init(void);
int pg_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) {
if (up->cycles > (uint64_t) up->timeout) {
// reset the cycle
up->cycles = 0;
PQfinish((PGconn *) up->data);
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 {
if ((up->cycles % up->freq) == 0) {
up->last_event = event_queue_write();
up->data = (void *) PQconnectStart(up->args);
if (up->data) {
// status = CONNECTING
up->state = PQstatus((PGconn *) up->data);
if (up->state == CONNECTION_BAD)
goto bad;
up->fd = PQsocket((PGconn *) up->data);
event_queue_add_fd_write(uwsgi.master_queue, up->fd);
return 0;
}
bad:
// 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) {
// check the state
up->state = PQstatus((PGconn *) up->data);
if (up->state == CONNECTION_BAD) {
// signal the bad connection (if needed)
up->cycles = 0;
PQfinish((PGconn *) up->data);
up->fd = -1;
up->state = 0;
if (!up->bad) {
up->bad = 1;
return 1;
}
return 0;
}
else if (up->state == CONNECTION_OK) {
up->cycles = 0;
PQfinish((PGconn *) up->data);
up->fd = -1;
// state = NOOP
up->state = 0;
if (up->bad) {
up->bad = 0;
return 1;
}
}
// still wait...
else {
PostgresPollingStatusType wait_type = PQconnectPoll((PGconn *) up->data);
// the connection is good
if (wait_type == PGRES_POLLING_ACTIVE || wait_type == PGRES_POLLING_FAILED || wait_type == PGRES_POLLING_OK) {
if (wait_type == PGRES_POLLING_ACTIVE)
wait_type = PQconnectPoll((PGconn *) up->data);
up->cycles = 0;
up->fd = -1;
// state = NOOP
up->state = 0;
PQfinish((PGconn *) up->data);
if (wait_type == PGRES_POLLING_FAILED) {
if (!up->bad) {
up->bad = 1;
return 1;
}
}
else {
if (up->bad) {
up->bad = 0;
return 1;
}
}
}
else if (wait_type == PGRES_POLLING_READING) {
event_queue_del_fd(uwsgi.master_queue, up->fd, up->last_event);
event_queue_add_fd_read(uwsgi.master_queue, up->fd);
up->last_event = event_queue_read();
}
else if (wait_type == PGRES_POLLING_WRITING) {
event_queue_del_fd(uwsgi.master_queue, up->fd, up->last_event);
event_queue_add_fd_write(uwsgi.master_queue, up->fd);
up->last_event = event_queue_write();
}
}
}
}
// default action
return 0;
}
int probepg_init() {
uwsgi_probe_register(&uwsgi.probes, "pg", pg_prober_callback);
return 0;
}
struct uwsgi_plugin probepg_plugin = {
.init = probepg_init,
};
plugins/probepg/uwsgiplugin.py
+9
View File
@@ -0,0 +1,9 @@
import os
NAME='probepg'
CFLAGS = os.popen('pg_config --cflags').read().rstrip().split()
CFLAGS.append('-I' + os.popen('pg_config --includedir').read().rstrip())
LDFLAGS = os.popen('pg_config --ldflags').read().rstrip().split()
LIBS = ['-L' + os.popen('pg_config --libdir').read().rstrip(), '-lpq']
GCC_LIST = ['pgprobe']
plugins/python/uwsgi_pymodule.c
+4 -2
View File
@@ -224,13 +224,15 @@ PyObject *py_uwsgi_add_cron(PyObject * self, PyObject * args) {
PyObject *py_uwsgi_add_probe(PyObject * self, PyObject * args) {
uint8_t uwsgi_signal;
int timeout = 0;
int freq = 0;
char *probe, *probe_args;
if (!PyArg_ParseTuple(args, "Bss:add_probe", &uwsgi_signal, &probe, &probe_args)) {
if (!PyArg_ParseTuple(args, "Bss|ii:add_probe", &uwsgi_signal, &probe, &probe_args, &timeout, &freq)) {
return NULL;
}
if (uwsgi_add_probe(uwsgi_signal, probe ,probe_args))
if (uwsgi_add_probe(uwsgi_signal, probe, probe_args, timeout, freq))
return PyErr_Format(PyExc_ValueError, "unable to add probe");
Py_INCREF(Py_None);
signal.c
+12 -1
View File
@@ -156,7 +156,7 @@ struct uwsgi_probe *uwsgi_probe_register(struct uwsgi_probe **up, char *name, in
}
int uwsgi_add_probe(uint8_t sig, char *kind, char *args) {
int uwsgi_add_probe(uint8_t sig, char *kind, char *args, int timeout, int freq) {
uwsgi_lock(uwsgi.probe_table_lock);
@@ -183,8 +183,19 @@ int uwsgi_add_probe(uint8_t sig, char *kind, char *args) {
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].last_event = 0;
ushared->probes[ushared->probes_cnt].data = NULL;
ushared->probes[ushared->probes_cnt].cycles = 0;
ushared->probes[ushared->probes_cnt].bad = 0;
if (!timeout) {
timeout = uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT];
}
ushared->probes[ushared->probes_cnt].timeout = timeout;
if (!freq) {
freq = 1;
}
ushared->probes[ushared->probes_cnt].freq = freq;
ushared->probes_cnt++;
}
else {
uwsgi.h
+6 -1
View File
@@ -976,8 +976,13 @@ struct uwsgi_signal_probe {
int fd;
int state;
int bad;
int last_event;
void *data;
uint64_t cycles;
int timeout;
int freq;
int registered;
uint8_t sig;
};
@@ -2505,7 +2510,7 @@ 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_add_probe(uint8_t sig, char *, char *, int, int);
int uwsgi_is_bad_connection(int);