From 93faeb417710d944cd4704ec265d323e85b1fdae Mon Sep 17 00:00:00 2001 From: "roberto@oneiric64" Date: Mon, 23 Jan 2012 16:04:10 +0100 Subject: [PATCH] first round of common code base for routers --- gateway.c | 28 +++++ lib/corerouter.h | 177 ++++++++++++++++++++++++++++++++ plugins/fastrouter/fastrouter.c | 170 +++--------------------------- plugins/http/http.c | 78 +++----------- uwsgi.c | 14 +-- uwsgi.h | 5 +- 6 files changed, 249 insertions(+), 223 deletions(-) create mode 100644 lib/corerouter.h diff --git a/gateway.c b/gateway.c index 8570dd4d..f79103eb 100644 --- a/gateway.c +++ b/gateway.c @@ -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; +} + + diff --git a/lib/corerouter.h b/lib/corerouter.h new file mode 100644 index 00000000..fc1f349c --- /dev/null +++ b/lib/corerouter.h @@ -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); + } + } + } + } + +} diff --git a/plugins/fastrouter/fastrouter.c b/plugins/fastrouter/fastrouter.c index 2ce4e050..68c958f2 100644 --- a/plugins/fastrouter/fastrouter.c +++ b/plugins/fastrouter/fastrouter.c @@ -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"); diff --git a/plugins/http/http.c b/plugins/http/http.c index f0096e23..db17f4b0 100644 --- a/plugins/http/http.c +++ b/plugins/http/http.c @@ -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 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; diff --git a/uwsgi.c b/uwsgi.c index 5acfbdd2..c5b95cbd 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -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) { diff --git a/uwsgi.h b/uwsgi.h index 8e92fd8b..e40eaac6 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -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 *);