diff --git a/emperor.c b/emperor.c index ce4bb56f..b378d3c0 100644 --- a/emperor.c +++ b/emperor.c @@ -21,6 +21,9 @@ struct uwsgi_instance { int pipe[2]; int pipe_config[2]; + + char *config; + uint32_t config_len; }; struct uwsgi_instance *ui; @@ -77,8 +80,15 @@ void emperor_stop(struct uwsgi_instance *c_ui) { void emperor_respawn(struct uwsgi_instance *c_ui, time_t mod) { // reload the uWSGI instance - if (write(c_ui->pipe[0], "\1", 1) != 1) { - uwsgi_error("write()"); + if (c_ui->use_config) { + if (write(c_ui->pipe[0], "\0", 1) != 1) { + uwsgi_error("write()"); + } + } + else { + if (write(c_ui->pipe[0], "\1", 1) != 1) { + uwsgi_error("write()"); + } } c_ui->respawns++; @@ -87,7 +97,7 @@ void emperor_respawn(struct uwsgi_instance *c_ui, time_t mod) { uwsgi_log("reload the uwsgi instance %s\n", c_ui->name); } -void emperor_add(char *name, time_t born) { +void emperor_add(char *name, time_t born, char *config, uint32_t config_size) { struct uwsgi_instance *c_ui = ui; struct uwsgi_instance *n_ui = NULL; @@ -103,6 +113,12 @@ void emperor_add(char *name, time_t born) { n_ui = uwsgi_malloc(sizeof(struct uwsgi_instance)); memset(n_ui, 0, sizeof(struct uwsgi_instance)); + if (config) { + n_ui->use_config = 1; + n_ui->config = config; + n_ui->config_len = config_size; + } + c_ui->ui_next = n_ui; uwsgi_log("c_ui->ui_next = %p\n", c_ui->ui_next); n_ui->ui_prev = c_ui; @@ -115,7 +131,7 @@ void emperor_add(char *name, time_t born) { goto clear; } - if (c_ui->use_config) { + if (n_ui->use_config) { if (socketpair(AF_UNIX, SOCK_STREAM, 0, n_ui->pipe_config)) { uwsgi_error("socketpair()"); goto clear; @@ -131,9 +147,16 @@ void emperor_add(char *name, time_t born) { n_ui->pid = pid; // close the right side of the pipe close(n_ui->pipe[1]); - if (c_ui->use_config) { + if (n_ui->use_config) { close(n_ui->pipe_config[1]); } + + if (n_ui->use_config) { + if (write(n_ui->pipe_config[0], n_ui->config, n_ui->config_len) <= 0) { + uwsgi_error("write()"); + } + close(n_ui->pipe_config[0]); + } return; } else { @@ -147,7 +170,7 @@ void emperor_add(char *name, time_t born) { } free(uef); - if (c_ui->use_config) { + if (n_ui->use_config) { uef = uwsgi_num2str(n_ui->pipe_config[1]); if (setenv("UWSGI_EMPEROR_FD_CONFIG", uef, 1)) { uwsgi_error("setenv()"); @@ -181,7 +204,7 @@ void emperor_add(char *name, time_t born) { // close the left side of the pipe close(n_ui->pipe[0]); - if (c_ui->use_config) { + if (n_ui->use_config) { close(n_ui->pipe_config[0]); } @@ -231,6 +254,7 @@ void emperor_loop() { struct dirent *de; char *amqp_port; int amqp_fd = -1; + char *amqp_routing_key; signal(SIGPIPE, SIG_IGN); @@ -285,8 +309,39 @@ reconnect: if (amqp_fd > -1) { uint64_t msgsize; if (uwsgi_waitfd(amqp_fd, 3)) { - char *config = uwsgi_amqp_consume(amqp_fd, &msgsize); - if (config && msgsize) { + char *config = uwsgi_amqp_consume(amqp_fd, &msgsize, &amqp_routing_key); + + if (!config) continue; + + if (amqp_routing_key) { + uwsgi_log("AMQP routing_key = %s\n", amqp_routing_key); + char *config_file = uwsgi_concat2("emperor://", amqp_routing_key); + free(amqp_routing_key); + + ui_current = emperor_get(config_file); + + if (ui_current) { + free(ui_current->config); + ui_current->config = config; + ui_current->config_len = msgsize; + if (!msgsize) { + emperor_del(ui_current); + } + else { + emperor_respawn(ui_current, time(NULL)); + } + } + else { + if (msgsize > 0) { + emperor_add(config_file, time(NULL), config, msgsize); + } + } + + + free(config_file); + } + else { + if (msgsize) { if (msgsize >= 0xff) { free(config); continue; } uwsgi_log("%.*s\n", (int)msgsize, config); @@ -309,11 +364,12 @@ reconnect: emperor_respawn(ui_current, time(NULL)); } else { - emperor_add(config_file, time(NULL)); + emperor_add(config_file, time(NULL), NULL, 0); } free(config_file); } + } } } else if (simple_mode) { @@ -341,7 +397,7 @@ reconnect: } } else { - emperor_add(de->d_name, st.st_mtime); + emperor_add(de->d_name, st.st_mtime, NULL, 0); } } } @@ -379,7 +435,7 @@ reconnect: } } else { - emperor_add(g.gl_pathv[i], st.st_mtime); + emperor_add(g.gl_pathv[i], st.st_mtime, NULL, 0); } } @@ -399,6 +455,8 @@ reconnect: diedpid = waitpid(WAIT_ANY, &waitpid_status, WNOHANG); } else { + // vacuum + waitpid(WAIT_ANY, &waitpid_status, WNOHANG); diedpid = 0; } if (diedpid < 0) { @@ -408,21 +466,23 @@ reconnect: while (ui_current->ui_next) { ui_current = ui_current->ui_next; if (ui_current->status == 1) { + if (ui_current->config) free(ui_current->config); emperor_del(ui_current); break; } - else if (stat(ui_current->name, &st)) { + else if (!ui_current->use_config && stat(ui_current->name, &st)) { emperor_stop(ui_current); } else if (ui_current->pid == diedpid) { if (ui_current->status == 0) { // respawn an accidentally dead instance - emperor_add(ui_current->name, ui_current->last_mod); + emperor_add(ui_current->name, ui_current->last_mod, ui_current->config, ui_current->config_len); emperor_del(ui_current); break; } else if (ui_current->status == 1) { // remove 'marked for dead' instance + if (ui_current->config) free(ui_current->config); emperor_del(ui_current); break; } diff --git a/lib/amqp.c b/lib/amqp.c index 90e22584..42982754 100644 --- a/lib/amqp.c +++ b/lib/amqp.c @@ -142,7 +142,7 @@ static int amqp_send_ack(int fd, uint64_t delivery_tag) { return 0; } -char *uwsgi_amqp_consume(int fd, uint64_t *msgsize) { +char *uwsgi_amqp_consume(int fd, uint64_t *msgsize, char **routing_key) { uint32_t size; struct amqp_frame_header fh; @@ -158,9 +158,31 @@ char *uwsgi_amqp_consume(int fd, uint64_t *msgsize) { ptr = frame+4; watermark = frame+size; + // consumer_tag ptr = amqp_get_str(ptr, watermark); if (!ptr) goto clear; + // delivery_tag (needed for ack) ptr = amqp_get_longlong(ptr, watermark, &delivery_tag); if (!ptr) goto clear; + // redelivered + if (ptr+1 > watermark) goto clear; + ptr++; + // exchange + ptr = amqp_get_str(ptr, watermark); if (!ptr) goto clear; + // routing_key + if (ptr+1 > watermark) goto clear; + uint8_t rk_size = (uint8_t) *ptr; + ptr++; + if (ptr+rk_size > watermark) goto clear; + if (rk_size > 0) { + char *rkey = uwsgi_concat2n(ptr, rk_size, "", 0); + ptr+=rk_size; + + *routing_key = rkey; + } + else { + *routing_key = NULL; + } + char *header = amqp_simple_get_frame(fd, &fh); if (!header) goto clear; diff --git a/utils.c b/utils.c index aee7c7dc..1651a0c5 100644 --- a/utils.c +++ b/utils.c @@ -1403,6 +1403,12 @@ char *uwsgi_open_and_read(char *url, int *size, int add_zero, char *magic_table[ } close(uwsgi.emperor_fd_config); uwsgi.emperor_fd_config = -1; + + if (add_zero) { + *size = *size + 1; + buffer = realloc(buffer, *size); + buffer[*size-1] = 0; + } } // fallback to file else { diff --git a/uwsgi.c b/uwsgi.c index 16204281..6cbeaf41 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -558,6 +558,10 @@ int main(int argc, char *argv[], char *envp[]) uwsgi.master_process = 1; uwsgi.no_orphans = 1; uwsgi_log("*** has_emperor mode detected (fd: %d) ***\n", uwsgi.emperor_fd); + + if (getenv("UWSGI_EMPEROR_FD_CONFIG")) { + uwsgi.emperor_fd_config = atoi(getenv("UWSGI_EMPEROR_FD_CONFIG")); + } } env_reloads = getenv("UWSGI_RELOADS"); diff --git a/uwsgi.h b/uwsgi.h index fbbbe234..eca82b31 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -1707,4 +1707,4 @@ int uwsgi_netlink_del(char *); int uwsgi_amqp_consume_queue(int, char *, char *, char *); -char *uwsgi_amqp_consume(int, uint64_t *); +char *uwsgi_amqp_consume(int, uint64_t *, char **);