clearlinux/uwsgi
2
0
Fork 0
mirror of https://github.com/clearlinux/uwsgi.git synced 2026-08-27 08:55:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

first round of common code base for routers

Browse Source
This commit is contained in:
roberto@oneiric64
2012-01-23 16:04:10 +01:00
parent 095b9154e0
commit 93faeb4177
6 changed files with 249 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway.c
+28
View File
@@ -102,9 +102,37 @@ struct uwsgi_gateway_socket *uwsgi_new_gateway_socket(char *name, char *owner) {
}
memset(uwsgi_sock, 0, sizeof(struct uwsgi_gateway_socket));
uwsgi_sock->fd = -1;
uwsgi_sock->name = name;
uwsgi_sock->owner = owner;
return uwsgi_sock;
}
struct uwsgi_gateway_socket *uwsgi_new_gateway_socket_from_fd(int fd, char *owner) {
struct uwsgi_gateway_socket *uwsgi_sock = uwsgi.gateway_sockets, *old_uwsgi_sock;
if (!uwsgi_sock) {
uwsgi.gateway_sockets = uwsgi_malloc(sizeof(struct uwsgi_gateway_socket));
uwsgi_sock = uwsgi.gateway_sockets;
}
else {
while(uwsgi_sock) {
old_uwsgi_sock = uwsgi_sock;
uwsgi_sock = uwsgi_sock->next;
}
uwsgi_sock = uwsgi_malloc(sizeof(struct uwsgi_gateway_socket));
old_uwsgi_sock->next = uwsgi_sock;
}
memset(uwsgi_sock, 0, sizeof(struct uwsgi_gateway_socket));
uwsgi_sock->fd = fd;
uwsgi_sock->name = uwsgi_getsockname(fd);
uwsgi_sock->owner = owner;
return uwsgi_sock;
}
lib/corerouter.h
+177
View File
@@ -0,0 +1,177 @@
/*
common functions for various routers (fastrouter, http...)
*/
static void uwsgi_corerouter_setup_sockets(char *gw_id) {
struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets;
while (ugs) {
if (!strcmp(gw_id, ugs->owner)) {
if (!ugs->subscription) {
ugs->port = strchr(ugs->name, ':');
if (ugs->fd == -1) {
if (ugs->port) {
ugs->fd = bind_to_tcp(ugs->name, uwsgi.listen_queue, ugs->port);
}
else {
ugs->fd = bind_to_unix(ugs->name, uwsgi.listen_queue, uwsgi.chmod_socket, uwsgi.abstract_socket);
}
}
// put socket in non-blocking mode
uwsgi_socket_nb(ugs->fd);
ugs->port++;
ugs->port_len = strlen(ugs->port);
uwsgi_log("%s bound on %s fd %d\n", gw_id, ugs->name, ugs->fd);
}
else {
if (ugs->fd == -1) {
ugs->fd = bind_to_udp(ugs->name, 0, 0);
}
uwsgi_log("%s subscription server bound on %s fd %d\n", gw_id, ugs->name, ugs->fd);
}
}
ugs = ugs->next;
}
}
static void *uwsgi_corerouter_setup_event_queue(char *gw_id, int id, int nevents, int *efd, int cheap) {
*efd = event_queue_init();
struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets;
while (ugs) {
if (!strcmp(gw_id, ugs->owner)) {
if (!cheap || ugs->subscription) {
event_queue_add_fd_read(*efd, ugs->fd);
}
ugs->gateway = &uwsgi.gateways[id];
}
ugs = ugs->next;
}
return event_queue_alloc(nevents);
}
static void __attribute__ ((unused)) uwsgi_corerouter_go_cheap(char *gw_id, int queue, int *i_am_cheap) {
uwsgi_log("[%s pid %d] no more nodes available. Going cheap...\n", gw_id, (int) uwsgi.mypid);
struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets;
while (ugs) {
if (!strcmp(ugs->owner, gw_id) && !ugs->subscription) {
event_queue_del_fd(queue, ugs->fd, event_queue_read());
}
ugs = ugs->next;
}
*i_am_cheap = 1;
}
static void __attribute__ ((unused)) uwsgi_corerouter_manage_subscription(char *gw_id, int id, struct uwsgi_gateway_socket *ugs, int queue, struct uwsgi_subscribe_slot *subscriptions, int regexp, void (*parse_hook) (char *, uint16_t, char *, uint16_t, void *), int cheap, int *i_am_cheap) {
int i;
struct uwsgi_subscribe_req usr;
char bbuf[4096];
ssize_t len = recv(ugs->fd, bbuf, 4096, 0);
#ifdef UWSGI_EVENT_USE_PORT
event_queue_add_fd_read(queue, ugs->fd);
#endif
if (len > 0) {
memset(&usr, 0, sizeof(struct uwsgi_subscribe_req));
uwsgi_hooked_parse(bbuf + 4, len - 4, parse_hook, &usr);
// subscribe request ?
if (bbuf[3] == 0) {
if (uwsgi_add_subscribe_node(&subscriptions, &usr, regexp) && *i_am_cheap) {
struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets;
while (ugs) {
if (!strcmp(ugs->owner, gw_id) && !ugs->subscription) {
event_queue_add_fd_read(queue, ugs->fd);
}
ugs = ugs->next;
}
*i_am_cheap = 0;
uwsgi_log("[%s pid %d] leaving cheap mode...\n", gw_id, (int) uwsgi.mypid);
}
}
//unsubscribe
else {
struct uwsgi_subscribe_node *node = uwsgi_get_subscribe_node_by_name(&subscriptions, usr.key, usr.keylen, usr.address, usr.address_len, regexp);
if (node && node->len) {
if (node->death_mark == 0)
uwsgi_log("[%s pid %d] %.*s => marking %.*s as failed\n", gw_id, (int) uwsgi.mypid, (int) usr.keylen, usr.key, (int) usr.address_len, usr.address);
node->failcnt++;
node->death_mark = 1;
// check if i can remove the node
if (node->reference == 0) {
uwsgi_remove_subscribe_node(&subscriptions, node);
}
if (subscriptions == NULL && cheap && !*i_am_cheap) {
uwsgi_corerouter_go_cheap(gw_id, queue, i_am_cheap);
}
}
}
// propagate the subscription to other nodes
for (i = 0; i < uwsgi.gateways_cnt; i++) {
if (i == id)
continue;
if (!strcmp(uwsgi.gateways[i].name, gw_id)) {
if (send(uwsgi.gateways[i].internal_subscription_pipe[0], bbuf, len, 0) != len) {
uwsgi_error("send()");
}
}
}
}
}
static void __attribute__ ((unused)) uwsgi_corerouter_manage_internal_subscription(char *gw_id, int queue, int fd, struct uwsgi_subscribe_slot *subscriptions, int regexp, void (*parse_hook) (char *, uint16_t, char *, uint16_t, void *), int cheap, int *i_am_cheap) {
struct uwsgi_subscribe_req usr;
char bbuf[4096];
ssize_t len = recv(fd, bbuf, 4096, 0);
#ifdef UWSGI_EVENT_USE_PORT
event_queue_add_fd_read(queue, fd);
#endif
if (len > 0) {
memset(&usr, 0, sizeof(struct uwsgi_subscribe_req));
uwsgi_hooked_parse(bbuf + 4, len - 4, parse_hook, &usr);
// subscribe request ?
if (bbuf[3] == 0) {
if (uwsgi_add_subscribe_node(&subscriptions, &usr, regexp) && *i_am_cheap) {
struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets;
while (ugs) {
if (!strcmp(ugs->owner, gw_id) && !ugs->subscription) {
event_queue_add_fd_read(queue, ugs->fd);
}
ugs = ugs->next;
}
*i_am_cheap = 0;
uwsgi_log("[%s pid %d] leaving cheap mode...\n", gw_id, (int) uwsgi.mypid);
}
}
//unsubscribe
else {
struct uwsgi_subscribe_node *node = uwsgi_get_subscribe_node_by_name(&subscriptions, usr.key, usr.keylen, usr.address, usr.address_len, regexp);
if (node && node->len) {
if (node->death_mark == 0)
uwsgi_log("[%s] %.*s => marking %.*s as failed\n", gw_id, (int) usr.keylen, usr.key, (int) usr.address_len, usr.address);
node->failcnt++;
node->death_mark = 1;
// check if i can remove the node
if (node->reference == 0) {
uwsgi_remove_subscribe_node(&subscriptions, node);
}
if (subscriptions == NULL && cheap && !*i_am_cheap) {
uwsgi_corerouter_go_cheap(gw_id, queue, i_am_cheap);
}
}
}
}
}
plugins/fastrouter/fastrouter.c
+17 -153
View File
@@ -14,6 +14,9 @@
extern struct uwsgi_server uwsgi;
#include "../../lib/corerouter.h"
#define LONG_ARGS_FASTROUTER 150001
#define LONG_ARGS_FASTROUTER_EVENTS 150002
#define LONG_ARGS_FASTROUTER_USE_PATTERN 150003
@@ -94,7 +97,6 @@ struct uwsgi_fastrouter {
int cheap;
int i_am_cheap;
int tolerance;
int harakiri;
@@ -102,19 +104,6 @@ struct uwsgi_fastrouter {
} ufr;
static void fastrouter_go_cheap(void) {
uwsgi_log("[uwsgi-fastrouter] no more nodes available. Going cheap...\n");
struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets;
while (ugs) {
if (!strcmp(ugs->owner, "uWSGI fastrouter") && !ugs->subscription) {
event_queue_del_fd(ufr.queue, ugs->fd, event_queue_read());
}
ugs = ugs->next;
}
ufr.i_am_cheap = 1;
}
struct option fastrouter_options[] = {
{"fastrouter", required_argument, 0, LONG_ARGS_FASTROUTER},
{"fastrouter-processes", required_argument, 0, LONG_ARGS_FASTROUTER_PROCESSES},
@@ -241,7 +230,7 @@ static void close_session(struct fastrouter_session *fr_session) {
uwsgi_remove_subscribe_node(&ufr.subscriptions, fr_session->un);
}
if (ufr.subscriptions == NULL && ufr.cheap && !ufr.i_am_cheap) {
fastrouter_go_cheap();
uwsgi_corerouter_go_cheap("uWSGI fastrouter", ufr.queue, &ufr.i_am_cheap);
}
}
@@ -330,18 +319,9 @@ void fastrouter_loop(int id) {
ufr.fr_table[i] = NULL;
}
ufr.queue = event_queue_init();
ufr.i_am_cheap = ufr.cheap;
void *events = event_queue_alloc(ufr.nevents);
struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets;
while (ugs) {
if (!strcmp("uWSGI fastrouter", ugs->owner)) {
event_queue_add_fd_read(ufr.queue, ugs->fd);
ugs->gateway = &uwsgi.gateways[id];
}
ugs = ugs->next;
}
void *events = uwsgi_corerouter_setup_event_queue("uWSGI fastrouter", id, ufr.nevents, &ufr.queue, ufr.i_am_cheap);
if (ufr.has_subscription_sockets)
event_queue_add_fd_read(ufr.queue, uwsgi.gateways[id].internal_subscription_pipe[1]);
@@ -384,8 +364,6 @@ void fastrouter_loop(int id) {
ufr.pb_base_dir = "/tmp";
}
char bbuf[UMAX16];
int nevents;
time_t delta;
@@ -393,7 +371,6 @@ void fastrouter_loop(int id) {
char *post_tmp_buf[0xffff];
int tmp_socket_name_len;
struct uwsgi_subscribe_req usr;
struct uwsgi_rb_timer *min_timeout;
@@ -487,57 +464,8 @@ void fastrouter_loop(int id) {
event_queue_add_fd_read(ufr.queue, new_connection);
}
else {
len = recv(ugs->fd, bbuf, 4096, 0);
#ifdef UWSGI_EVENT_USE_PORT
event_queue_add_fd_read(ufr.queue, ugs->fd);
#endif
if (len > 0) {
memset(&usr, 0, sizeof(struct uwsgi_subscribe_req));
uwsgi_hooked_parse(bbuf + 4, len - 4, fastrouter_manage_subscription, &usr);
// subscribe request ?
if (bbuf[3] == 0) {
if (uwsgi_add_subscribe_node(&ufr.subscriptions, &usr, ufr.subscription_regexp) && ufr.i_am_cheap) {
struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets;
while (ugs) {
if (!strcmp(ugs->owner, "uWSGI fastrouter") && !ugs->subscription) {
event_queue_add_fd_read(ufr.queue, ugs->fd);
}
ugs = ugs->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->failcnt++;
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();
}
}
}
// propagate the subscription to other nodes
for (i = 0; i < uwsgi.gateways_cnt; i++) {
if (i == id)
continue;
if (!strcmp(uwsgi.gateways[i].name, "uWSGI fastrouter")) {
if (send(uwsgi.gateways[i].internal_subscription_pipe[0], bbuf, len, 0) != len) {
uwsgi_error("send()");
}
}
}
}
uwsgi_corerouter_manage_subscription("uWSGI fastrouter", id, ugs, ufr.queue, ufr.subscriptions,
ufr.subscription_regexp, fastrouter_manage_subscription, ufr.cheap, &ufr.i_am_cheap);
}
taken = 1;
@@ -553,47 +481,8 @@ void fastrouter_loop(int id) {
}
if (interesting_fd == uwsgi.gateways[id].internal_subscription_pipe[1]) {
len = recv(interesting_fd, bbuf, 4096, 0);
#ifdef UWSGI_EVENT_USE_PORT
event_queue_add_fd_read(ufr.queue, interesting_fd);
#endif
if (len > 0) {
memset(&usr, 0, sizeof(struct uwsgi_subscribe_req));
uwsgi_hooked_parse(bbuf + 4, len - 4, fastrouter_manage_subscription, &usr);
// subscribe request ?
if (bbuf[3] == 0) {
if (uwsgi_add_subscribe_node(&ufr.subscriptions, &usr, ufr.subscription_regexp) && ufr.i_am_cheap) {
struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets;
while (ugs) {
if (!strcmp(ugs->owner, "uWSGI fastrouter") && !ugs->subscription) {
event_queue_add_fd_read(ufr.queue, ugs->fd);
}
ugs = ugs->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->failcnt++;
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();
}
}
}
}
uwsgi_corerouter_manage_internal_subscription("uWSGI fastrouter", ufr.queue, interesting_fd, ufr.subscriptions,
ufr.subscription_regexp, fastrouter_manage_subscription, ufr.cheap, &ufr.i_am_cheap);
}
else if (interesting_fd == ufr.fr_stats_server) {
fastrouter_send_stats(ufr.fr_stats_server);
@@ -677,7 +566,7 @@ void fastrouter_loop(int id) {
fr_session->modifier1 = fr_session->un->modifier1;
}
else if (ufr.subscriptions == NULL && ufr.cheap && !ufr.i_am_cheap) {
fastrouter_go_cheap();
uwsgi_corerouter_go_cheap("uWSGI fastrouter", ufr.queue, &ufr.i_am_cheap);
}
}
else if (ufr.base) {
@@ -956,34 +845,13 @@ int fastrouter_init() {
if (!ufr.nevents)
ufr.nevents = 64;
struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets;
while (ugs) {
if (!strcmp("uWSGI fastrouter", ugs->owner)) {
if (!ugs->subscription) {
ugs->port = strchr(ugs->name, ':');
if (ugs->port) {
ugs->fd = bind_to_tcp(ugs->name, uwsgi.listen_queue, ugs->port);
}
else {
ugs->fd = bind_to_unix(ugs->name, uwsgi.listen_queue, uwsgi.chmod_socket, uwsgi.abstract_socket);
}
// put socket in non-blocking mode
uwsgi_socket_nb(ugs->fd);
ugs->port++;
ugs->port_len = strlen(ugs->port);
uwsgi_log("uwsgi fastrouter bound on %s fd %d\n", ugs->name, ugs->fd);
}
else {
ugs->fd = bind_to_udp(ugs->name, 0, 0);
uwsgi_log("uwsgi fastrouter subscription server bound on %s fd %d\n", ugs->name, ugs->fd);
}
}
ugs = ugs->next;
}
uwsgi_corerouter_setup_sockets("uWSGI fastrouter");
if (ufr.processes < 1)
ufr.processes = 1;
if (ufr.cheap) {
uwsgi_log("starting fastrouter in cheap mode\n");
}
for (i = 0; i < ufr.processes; i++) {
if (register_gateway("uWSGI fastrouter", fastrouter_loop) == NULL) {
uwsgi_log("unable to register the fastrouter gateway\n");
@@ -1000,11 +868,9 @@ int fastrouter_opt(int i, char *optarg) {
char *cs;
char *cs_code;
char *cs_func;
/*
int j;
int zerg_fd;
int *zerg;
*/
struct uwsgi_gateway_socket *ugs;
switch (i) {
@@ -1013,7 +879,6 @@ int fastrouter_opt(int i, char *optarg) {
ufr.has_sockets++;
return 1;
case LONG_ARGS_FASTROUTER_ZERG:
/*
zerg_fd = uwsgi_connect(optarg, 30, 0);
if (zerg_fd < 0) {
uwsgi_log("--- unable to connect to zerg server ---\n");
@@ -1027,10 +892,9 @@ int fastrouter_opt(int i, char *optarg) {
close(zerg_fd);
for(j=0;j<8;j++) {
if (zerg[j] == -1) break;
fr_sock = uwsgi_fastrouter_new_socket(NULL, zerg[j]);
fr_sock->zerg = optarg;
ugs = uwsgi_new_gateway_socket_from_fd(zerg[j], "uWSGI fastrouter");
ugs->zerg = optarg;
}
*/
return 1;
case LONG_ARGS_FASTROUTER_SUBSCRIPTION_SERVER:
ugs = uwsgi_new_gateway_socket(optarg, "uWSGI fastrouter");
plugins/http/http.c
+15 -63
View File
@@ -12,6 +12,10 @@
#include "../../uwsgi.h"
extern struct uwsgi_server uwsgi;
#include "../../lib/corerouter.h"
#define MAX_HTTP_VEC 128
#define MAX_HTTP_EXTRA_VARS 64
@@ -41,6 +45,9 @@ struct uwsgi_http {
int use_cluster;
int nevents;
int cheap;
int i_am_cheap;
int has_subscription_sockets;
int subscription_regexp;
@@ -475,7 +482,8 @@ void http_loop(int id) {
struct http_session *uhttp_session;
struct http_session **uhttp_table;
struct uwsgi_subscribe_req usr;
int uhttp_queue;
int soopt;
socklen_t solen = sizeof(int);
@@ -485,18 +493,7 @@ void http_loop(int id) {
uhttp_table[i] = NULL;
}
int uhttp_queue = event_queue_init();
void *events = event_queue_alloc(uhttp.nevents);
struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets;
while (ugs) {
if (!strcmp("uWSGI http", ugs->owner)) {
event_queue_add_fd_read(uhttp_queue, ugs->fd);
ugs->gateway = &uwsgi.gateways[id];
}
ugs = ugs->next;
}
void *events = uwsgi_corerouter_setup_event_queue("uWSGI http", id, uhttp.nevents, &uhttp_queue, 0);
if (uhttp.has_subscription_sockets)
event_queue_add_fd_read(uhttp_queue, uwsgi.gateways[id].internal_subscription_pipe[1]);
@@ -575,24 +572,8 @@ void http_loop(int id) {
event_queue_add_fd_read(uhttp_queue, new_connection);
}
else {
len = recv(ugs->fd, bbuf, 4096, 0);
#ifdef UWSGI_EVENT_USE_PORT
event_queue_add_fd_read(uhttp_queue, ugs->fd);
#endif
if (len > 0) {
memset(&usr, 0, sizeof(struct uwsgi_subscribe_req));
uwsgi_hooked_parse(bbuf + 4, len - 4, http_manage_subscription, &usr);
uwsgi_add_subscribe_node(&uhttp.subscriptions, &usr, uhttp.subscription_regexp);
// propagate the subscription to other nodes
for(i=0;i<uwsgi.gateways_cnt;i++) {
if (i == id) continue;
if (!strcmp(uwsgi.gateways[i].name, "uWSGI http")) {
if (send(uwsgi.gateways[i].internal_subscription_pipe[0], bbuf, len, 0) != len) {
uwsgi_error("send()");
}
}
}
}
uwsgi_corerouter_manage_subscription("uWSGI http", id, ugs, uhttp_queue, uhttp.subscriptions,
uhttp.subscription_regexp, http_manage_subscription, 0, &uhttp.i_am_cheap);
}
taken = 1;
@@ -607,15 +588,8 @@ void http_loop(int id) {
continue;
if (interesting_fd == uwsgi.gateways[id].internal_subscription_pipe[1]) {
len = recv(interesting_fd, bbuf, 4096, 0);
#ifdef UWSGI_EVENT_USE_PORT
event_queue_add_fd_read(uhttp_queue, interesting_fd);
#endif
if (len > 0) {
memset(&usr, 0, sizeof(struct uwsgi_subscribe_req));
uwsgi_hooked_parse(bbuf + 4, len - 4, http_manage_subscription, &usr);
uwsgi_add_subscribe_node(&uhttp.subscriptions, &usr, uhttp.subscription_regexp);
}
uwsgi_corerouter_manage_internal_subscription("uWSGI http", uhttp_queue, interesting_fd, uhttp.subscriptions,
uhttp.subscription_regexp, http_manage_subscription, 0, &uhttp.i_am_cheap);
}
// process already active sessions;
@@ -952,29 +926,7 @@ int http_init() {
uwsgi_new_socket(uwsgi_concat2("127.0.0.1:0", ""));
}
struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets;
while (ugs) {
if (!strcmp("uWSGI http", ugs->owner)) {
if (!ugs->subscription) {
ugs->port = strchr(ugs->name, ':');
if (!ugs->port) {
uwsgi_log("invalid HTTP ip:port syntax\n");
exit(1);
}
ugs->fd = bind_to_tcp(ugs->name, uwsgi.listen_queue, ugs->port);
// put socket in non-blocking mode
uwsgi_socket_nb(ugs->fd);
ugs->port++;
ugs->port_len = strlen(ugs->port);
uwsgi_log("HTTP router/proxy bound on %s fd %d\n", ugs->name, ugs->fd);
}
else {
ugs->fd = bind_to_udp(ugs->name, 0, 0);
uwsgi_log("HTTP subscription server bound on %s fd %d\n", ugs->name, ugs->fd);
}
}
ugs = ugs->next;
}
uwsgi_corerouter_setup_sockets("uWSGI http");
if (uhttp.processes < 1)
uhttp.processes = 1;
uwsgi.c
+8 -6
View File
@@ -221,6 +221,7 @@ static struct option long_base_options[] = {
{"subscribe", required_argument, 0, LONG_ARGS_SUBSCRIBE_TO},
{"subscribe-freq", required_argument, 0, LONG_ARGS_SUBSCRIBE_FREQ},
{"subscription-tolerance", required_argument, 0, LONG_ARGS_SUBSCR_TOLERANCE},
{"unsubscribe-on-graceful-reload", no_argument, &uwsgi.unsubscribe_on_graceful_reload, 1},
#ifdef UWSGI_SNMP
{"snmp", optional_argument, 0, LONG_ARGS_SNMP},
{"snmp-community", required_argument, 0, LONG_ARGS_SNMP_COMMUNITY},
@@ -741,12 +742,13 @@ 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;
}
if (uwsgi.unsubscribe_on_graceful_reload) {
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) {
uwsgi.h
+4 -1
View File
@@ -1629,6 +1629,7 @@ struct uwsgi_server {
// subscription client
int subscribe_freq;
int subscription_tolerance;
int unsubscribe_on_graceful_reload;
struct uwsgi_string_list *subscriptions;
#ifdef __linux__
@@ -2618,6 +2619,9 @@ struct uwsgi_logger *uwsgi_get_logger(char *);
char *uwsgi_getsockname(int);
char *uwsgi_get_var(struct wsgi_request *, char *, uint16_t, uint16_t *);
struct uwsgi_gateway_socket *uwsgi_new_gateway_socket(char *, char *);
struct uwsgi_gateway_socket *uwsgi_new_gateway_socket_from_fd(int, char *);
void escape_shell_arg(char *, size_t, char *);
void *uwsgi_malloc_shared(size_t);
@@ -2635,4 +2639,3 @@ int uwsgi_init(int, char **, char **);
#endif
struct uwsgi_gateway_socket *uwsgi_new_gateway_socket(char *, char *);