diff --git a/buildconf/base.ini b/buildconf/base.ini index 69c49bd3..226eafa9 100644 --- a/buildconf/base.ini +++ b/buildconf/base.ini @@ -28,7 +28,7 @@ plugins = bin_name = uwsgi append_version = plugin_dir = . -embedded_plugins = %(main_plugin)s, ping, cache, nagios, rrdtool, carbon, rpc, fastrouter, http, ugreen, signal, syslog, rsyslog, logsocket, router_uwsgi, router_redirect, router_basicauth, zergpool, redislog +embedded_plugins = %(main_plugin)s, ping, cache, nagios, rrdtool, carbon, rpc, corerouter, fastrouter, http, ugreen, signal, syslog, rsyslog, logsocket, router_uwsgi, router_redirect, router_basicauth, zergpool, redislog as_shared_library = false locking = auto diff --git a/gateway.c b/gateway.c index f6cbd296..bb1fb171 100644 --- a/gateway.c +++ b/gateway.c @@ -2,7 +2,7 @@ extern struct uwsgi_server uwsgi; -struct uwsgi_gateway *register_gateway(char *name, void (*loop)(int)) { +struct uwsgi_gateway *register_gateway(char *name, void (*loop)(int, void *), void *data) { struct uwsgi_gateway *ug; int num=1,i; @@ -26,6 +26,7 @@ struct uwsgi_gateway *register_gateway(char *name, void (*loop)(int)) { ug->loop = loop; ug->num = num; ug->fullname = fullname; + ug->data = data; if (socketpair(AF_UNIX, SOCK_DGRAM, 0, ug->internal_subscription_pipe)) { uwsgi_error("socketpair()"); @@ -71,7 +72,7 @@ void gateway_respawn(int id) { signal(SIGUSR2, SIG_IGN); signal(SIGPIPE, SIG_IGN); - ug->loop(id); + ug->loop(id, ug->data); // never here !!! (i hope) exit(1); } diff --git a/plugins/fastrouter/fastrouter.c b/plugins/fastrouter/fastrouter.c index cd2da482..c2d9cdf5 100644 --- a/plugins/fastrouter/fastrouter.c +++ b/plugins/fastrouter/fastrouter.c @@ -14,811 +14,67 @@ extern struct uwsgi_server uwsgi; -#include "../../lib/corerouter.h" - - -#ifdef UWSGI_SCTP -extern struct uwsgi_fr_sctp_node **uwsgi_fastrouter_sctp_nodes; -extern struct uwsgi_fr_sctp_node **uwsgi_fastrouter_sctp_nodes_current; -#endif - -void fastrouter_send_stats(int); - #include "fr.h" struct uwsgi_fastrouter ufr; -void uwsgi_opt_fastrouter(char *opt, char *value, void *foobar) { - uwsgi_new_gateway_socket(value, "uWSGI fastrouter"); - ufr.has_sockets++; -} - -void uwsgi_opt_fastrouter_use_socket(char *opt, char *value, void *foobar) { - ufr.use_socket = 1; - - if (value) { - ufr.socket_num = atoi(value); - } -} - -void uwsgi_opt_fastrouter_zerg(char *opt, char *value, void *foobar) { - - int j; - int count = 8; - struct uwsgi_gateway_socket *ugs; - - int zerg_fd = uwsgi_connect(value, 30, 0); - if (zerg_fd < 0) { - uwsgi_log("--- unable to connect to zerg server ---\n"); - exit(1); - } - - int last_count = count; - int *zerg = uwsgi_attach_fd(zerg_fd, &count, "uwsgi-zerg", 10); - if (zerg == NULL) { - if (last_count != count) { - close(zerg_fd); - zerg_fd = uwsgi_connect(value, 30, 0); - if (zerg_fd < 0) { - uwsgi_log("--- unable to connect to zerg server ---\n"); - exit(1); - } - zerg = uwsgi_attach_fd(zerg_fd, &count, "uwsgi-zerg", 10); - } - else { - uwsgi_log("--- invalid data received from zerg-server ---\n"); - exit(1); - } - } - - if (zerg == NULL) { - uwsgi_log("--- invalid data received from zerg-server ---\n"); - exit(1); - } - - - close(zerg_fd); - - for(j=0;jzerg = optarg; - } -} - -void uwsgi_opt_fastrouter_cs(char *opt, char *value, void *foobar) { - - char *cs = uwsgi_str(value); - char *cs_code = strchr(cs, ':'); - if (!cs_code) { - uwsgi_log("invalid code_string option\n"); - exit(1); - } - cs_code[0] = 0; - char *cs_func = strchr(cs_code + 1, ':'); - if (!cs_func) { - uwsgi_log("invalid code_string option\n"); - exit(1); - } - cs_func[0] = 0; - ufr.code_string_modifier1 = atoi(cs); - ufr.code_string_code = cs_code + 1; - ufr.code_string_function = cs_func + 1; - -} - -void uwsgi_opt_fastrouter_ss(char *opt, char *value, void *foobar) { - - struct uwsgi_gateway_socket *ugs = uwsgi_new_gateway_socket(value, "uWSGI fastrouter"); - ugs->subscription = 1; - ufr.has_subscription_sockets++; - -} - -void uwsgi_opt_fastrouter_use_base(char *opt, char *value, void *foobar) { - ufr.base = value; - ufr.base_len = strlen(ufr.base); -} - -void uwsgi_opt_fastrouter_use_pattern(char *opt, char *value, void *foobar) { - ufr.pattern = value; - ufr.pattern_len = strlen(ufr.pattern); -} struct uwsgi_option fastrouter_options[] = { - {"fastrouter", required_argument, 0, "run the fastrouter on the specified port", uwsgi_opt_fastrouter, NULL, 0}, - {"fastrouter-processes", required_argument, 0, "prefork the specified number of fastrouter processes", uwsgi_opt_set_int, &ufr.processes, 0}, - {"fastrouter-workers", required_argument, 0, "prefork the specified number of fastrouter processes", uwsgi_opt_set_int, &ufr.processes, 0}, - {"fastrouter-zerg", required_argument, 0, "attach the fastrouter to a zerg server", uwsgi_opt_fastrouter_zerg, NULL, 0 }, - {"fastrouter-use-cache", no_argument, 0, "use uWSGI cache as hostname->server mapper for the fastrouter", uwsgi_opt_true, &ufr.use_cache, 0}, + {"fastrouter", required_argument, 0, "run the fastrouter on the specified port", uwsgi_opt_corerouter, &ufr, 0}, + {"fastrouter-processes", required_argument, 0, "prefork the specified number of fastrouter processes", uwsgi_opt_set_int, &ufr.cr.processes, 0}, + {"fastrouter-workers", required_argument, 0, "prefork the specified number of fastrouter processes", uwsgi_opt_set_int, &ufr.cr.processes, 0}, + {"fastrouter-zerg", required_argument, 0, "attach the fastrouter to a zerg server", uwsgi_opt_corerouter_zerg, &ufr, 0 }, + {"fastrouter-use-cache", no_argument, 0, "use uWSGI cache as hostname->server mapper for the fastrouter", uwsgi_opt_true, &ufr.cr.use_cache, 0}, - {"fastrouter-use-pattern", required_argument, 0, "use a pattern for fastrouter hostname->server mapping", uwsgi_opt_fastrouter_use_pattern, NULL, 0}, - {"fastrouter-use-base", required_argument, 0, "use a base dir for fastrouter hostname->server mapping", uwsgi_opt_fastrouter_use_base, NULL, 0}, + {"fastrouter-use-pattern", required_argument, 0, "use a pattern for fastrouter hostname->server mapping", uwsgi_opt_corerouter_use_pattern, &ufr, 0}, + {"fastrouter-use-base", required_argument, 0, "use a base dir for fastrouter hostname->server mapping", uwsgi_opt_corerouter_use_base, &ufr, 0}, - {"fastrouter-fallback", required_argument, 0, "fallback to the specified node in case of error", uwsgi_opt_add_string_list, &ufr.fallback, 0}, + {"fastrouter-fallback", required_argument, 0, "fallback to the specified node in case of error", uwsgi_opt_add_string_list, &ufr.cr.fallback, 0}, - {"fastrouter-use-code-string", required_argument, 0, "use code string as hostname->server mapper for the fastrouter", uwsgi_opt_fastrouter_cs, NULL, 0}, - {"fastrouter-use-socket", optional_argument, 0, "forward request to the specified uwsgi socket", uwsgi_opt_fastrouter_use_socket, NULL, 0}, - {"fastrouter-to", required_argument, 0, "forward requests to the specified uwsgi server (you can specify it multiple times for load balancing)", uwsgi_opt_add_string_list, &ufr.static_nodes, 0}, - {"fastrouter-gracetime", required_argument, 0, "retry connections to dead static nodes after the specified amount of seconds", uwsgi_opt_set_int, &ufr.static_node_gracetime, 0}, - {"fastrouter-events", required_argument, 0, "set the maximum number of concurrent events", uwsgi_opt_set_int, &ufr.nevents, 0}, - {"fastrouter-quiet", required_argument, 0, "do not report failed connections to instances", uwsgi_opt_true, &ufr.quiet, 0}, - {"fastrouter-cheap", no_argument, 0, "run the fastrouter in cheap mode", uwsgi_opt_true, &ufr.cheap, 0}, - {"fastrouter-subscription-server", required_argument, 0, "run the fastrouter subscription server on the spcified address", uwsgi_opt_fastrouter_ss, NULL, 0}, + {"fastrouter-use-code-string", required_argument, 0, "use code string as hostname->server mapper for the fastrouter", uwsgi_opt_corerouter_cs, &ufr, 0}, + {"fastrouter-use-socket", optional_argument, 0, "forward request to the specified uwsgi socket", uwsgi_opt_corerouter_use_socket, NULL, 0}, + {"fastrouter-to", required_argument, 0, "forward requests to the specified uwsgi server (you can specify it multiple times for load balancing)", uwsgi_opt_add_string_list, &ufr.cr.static_nodes, 0}, + {"fastrouter-gracetime", required_argument, 0, "retry connections to dead static nodes after the specified amount of seconds", uwsgi_opt_set_int, &ufr.cr.static_node_gracetime, 0}, + {"fastrouter-events", required_argument, 0, "set the maximum number of concurrent events", uwsgi_opt_set_int, &ufr.cr.nevents, 0}, + {"fastrouter-quiet", required_argument, 0, "do not report failed connections to instances", uwsgi_opt_true, &ufr.cr.quiet, 0}, + {"fastrouter-cheap", no_argument, 0, "run the fastrouter in cheap mode", uwsgi_opt_true, &ufr.cr.cheap, 0}, + {"fastrouter-subscription-server", required_argument, 0, "run the fastrouter subscription server on the spcified address", uwsgi_opt_corerouter_ss, &ufr, 0}, {"fastrouter-subscription-slot", required_argument, 0, "*** deprecated ***", uwsgi_opt_deprecated, (void *) "useless thanks to the new implementation", 0}, - {"fastrouter-subscription-use-regexp", no_argument, 0, "enable regexp for subscription system", uwsgi_opt_true, &ufr.subscription_regexp, 0}, + {"fastrouter-subscription-use-regexp", no_argument, 0, "enable regexp for subscription system", uwsgi_opt_true, &ufr.cr.subscription_regexp, 0}, #ifdef UWSGI_SCTP - {"fastrouter-sctp", required_argument, 0, "run the fastrouter SCTP server on the specified address", uwsgi_opt_fastrouter_sctp, NULL, 0}, + {"fastrouter-sctp", required_argument, 0, "run the fastrouter SCTP server on the specified address", uwsgi_opt_corerouter_sctp, ufr, 0}, #endif - {"fastrouter-timeout", required_argument, 0, "set fastrouter timeout", uwsgi_opt_set_int, &ufr.socket_timeout, 0}, - {"fastrouter-post-buffering", required_argument, 0, "enable fastrouter post buffering", uwsgi_opt_set_64bit, &ufr.post_buffering, 0}, - {"fastrouter-post-buffering-dir", required_argument, 0, "put fastrouter buffered files to the specified directory", uwsgi_opt_set_str, &ufr.pb_base_dir, 0}, + {"fastrouter-timeout", required_argument, 0, "set fastrouter timeout", uwsgi_opt_set_int, &ufr.cr.socket_timeout, 0}, + {"fastrouter-post-buffering", required_argument, 0, "enable fastrouter post buffering", uwsgi_opt_set_64bit, &ufr.cr.post_buffering, 0}, + {"fastrouter-post-buffering-dir", required_argument, 0, "put fastrouter buffered files to the specified directory", uwsgi_opt_set_str, &ufr.cr.pb_base_dir, 0}, - {"fastrouter-stats", required_argument, 0, "run the fastrouter stats server", uwsgi_opt_set_str, &ufr.stats_server, 0}, - {"fastrouter-stats-server", required_argument, 0, "run the fastrouter stats server", uwsgi_opt_set_str, &ufr.stats_server, 0}, - {"fastrouter-ss", required_argument, 0, "run the fastrouter stats server", uwsgi_opt_set_str, &ufr.stats_server, 0}, - {"fastrouter-harakiri", required_argument, 0, "enable fastrouter harakiri", uwsgi_opt_set_int, &ufr.harakiri, 0 }, + {"fastrouter-stats", required_argument, 0, "run the fastrouter stats server", uwsgi_opt_set_str, &ufr.cr.stats_server, 0}, + {"fastrouter-stats-server", required_argument, 0, "run the fastrouter stats server", uwsgi_opt_set_str, &ufr.cr.stats_server, 0}, + {"fastrouter-ss", required_argument, 0, "run the fastrouter stats server", uwsgi_opt_set_str, &ufr.cr.stats_server, 0}, + {"fastrouter-harakiri", required_argument, 0, "enable fastrouter harakiri", uwsgi_opt_set_int, &ufr.cr.harakiri, 0 }, {0, 0, 0, 0, 0, 0, 0}, }; -extern struct uwsgi_server uwsgi; - -void fastrouter_manage_subscription(char *key, uint16_t keylen, char *val, uint16_t vallen, void *data) { - - struct uwsgi_subscribe_req *usr = (struct uwsgi_subscribe_req *) data; - - if (!uwsgi_strncmp("key", 3, key, keylen)) { - usr->key = val; - usr->keylen = vallen; - } - else if (!uwsgi_strncmp("address", 7, key, keylen)) { - usr->address = val; - usr->address_len = vallen; - } - else if (!uwsgi_strncmp("modifier1", 9, key, keylen)) { - usr->modifier1 = uwsgi_str_num(val, vallen); - } - else if (!uwsgi_strncmp("cores", 5, key, keylen)) { - usr->cores = uwsgi_str_num(val, vallen); - } - else if (!uwsgi_strncmp("load", 4, key, keylen)) { - usr->load = uwsgi_str_num(val, vallen); - } - else if (!uwsgi_strncmp("weight", 5, key, keylen)) { - usr->weight = uwsgi_str_num(val, vallen); - } -} - -static struct uwsgi_rb_timer *reset_timeout(struct fastrouter_session *); - -void close_session(struct fastrouter_session *fr_session) { - - - if (fr_session->instance_fd != -1) { -#ifdef UWSGI_SCTP - if (!ufr.fr_table[fr_session->instance_fd]->persistent) { -#endif - close(fr_session->instance_fd); - ufr.fr_table[fr_session->instance_fd] = NULL; -#ifdef UWSGI_SCTP - } -#endif - } - - if (ufr.subscriptions && fr_session->un && fr_session->un->len > 0) { - // decrease reference count -#ifdef UWSGI_DEBUG - uwsgi_log("[1] node %.*s refcnt: %llu\n", fr_session->un->len, fr_session->un->name, fr_session->un->reference); -#endif - fr_session->un->reference--; -#ifdef UWSGI_DEBUG - uwsgi_log("[2] node %.*s refcnt: %llu\n", fr_session->un->len, fr_session->un->name, fr_session->un->reference); -#endif - } - - - if (fr_session->instance_failed) { - - if (fr_session->soopt) { - if (!ufr.quiet) - uwsgi_log("unable to connect() to uwsgi instance \"%.*s\": %s\n", (int) fr_session->instance_address_len, fr_session->instance_address, strerror(fr_session->soopt)); - } - else if (fr_session->timed_out) { - if (fr_session->instance_address_len > 0) { - if (fr_session->status == FASTROUTER_STATUS_CONNECTING) { - if (!ufr.quiet) - uwsgi_log("unable to connect() to uwsgi instance \"%.*s\": timeout\n", (int) fr_session->instance_address_len, fr_session->instance_address); - } - else if (fr_session->status == FASTROUTER_STATUS_RESPONSE) { - uwsgi_log("timeout waiting for instance \"%.*s\"\n", (int) fr_session->instance_address_len, fr_session->instance_address); - } - } - } - - // now check for dead nodes - if (ufr.subscriptions && fr_session->un && fr_session->un->len > 0) { - - if (fr_session->un->death_mark == 0) - uwsgi_log("[uwsgi-fastrouter] %.*s => marking %.*s as failed\n", (int) fr_session->hostname_len, fr_session->hostname, (int) fr_session->instance_address_len, fr_session->instance_address); - - fr_session->un->failcnt++; - fr_session->un->death_mark = 1; - // check if i can remove the node - if (fr_session->un->reference == 0) { - uwsgi_remove_subscribe_node(&ufr.subscriptions, fr_session->un); - } - if (ufr.subscriptions == NULL && ufr.cheap && !ufr.i_am_cheap && !ufr.fallback) { - uwsgi_gateway_go_cheap("uWSGI fastrouter", ufr.queue, &ufr.i_am_cheap); - } - - } - else if (fr_session->static_node) { - fr_session->static_node->custom = uwsgi_now(); - uwsgi_log("[uwsgi-fastrouter] %.*s => marking %.*s as failed\n", (int) fr_session->hostname_len, fr_session->hostname, (int) fr_session->instance_address_len, fr_session->instance_address); - } - - - if (fr_session->tmp_socket_name) { - free(fr_session->tmp_socket_name); - fr_session->tmp_socket_name = NULL; - } - - if (ufr.fallback) { - // ok let's try with the fallback nodes - if (!fr_session->fallback) { - fr_session->fallback = ufr.fallback; - } - else { - fr_session->fallback = fr_session->fallback->next; - if (!fr_session->fallback) goto end; - } - - fr_session->instance_address = fr_session->fallback->value; - fr_session->instance_address_len = fr_session->fallback->len; - - // reset error and timeout - fr_session->timeout = reset_timeout(fr_session); - fr_session->timed_out = 0; - fr_session->soopt = 0; - - // reset nodes - fr_session->un = NULL; - fr_session->static_node = NULL; - - fr_session->pass_fd = is_unix(fr_session->instance_address, fr_session->instance_address_len); - - - fr_session->instance_fd = uwsgi_connectn(fr_session->instance_address, fr_session->instance_address_len, 0, 1); - - if (fr_session->instance_fd < 0) { - fr_session->instance_failed = 1; - fr_session->soopt = errno; - close_session(fr_session); - return; - } - - ufr.fr_table[fr_session->instance_fd] = fr_session; - - fr_session->status = FASTROUTER_STATUS_CONNECTING; - ufr.fr_table[fr_session->instance_fd] = fr_session; - event_queue_add_fd_write(ufr.queue, fr_session->instance_fd); - return; - - } - } - -end: - - if (fr_session->tmp_socket_name) { - free(fr_session->tmp_socket_name); - } - - if (fr_session->buf_file) - fclose(fr_session->buf_file); - - if (fr_session->buf_file_name) { - if (unlink(fr_session->buf_file_name)) { - uwsgi_error("unlink()"); - } - free(fr_session->buf_file_name); - } - - close(fr_session->fd); - ufr.fr_table[fr_session->fd] = NULL; - - del_timeout(fr_session); - free(fr_session); -} - -static struct uwsgi_rb_timer *reset_timeout(struct fastrouter_session *fr_session) { - del_timeout(fr_session); - return add_timeout(fr_session); -} - -static void expire_timeouts() { - - time_t current = time(NULL); - struct uwsgi_rb_timer *urbt; - struct fastrouter_session *fr_session; - - for (;;) { - urbt = uwsgi_min_rb_timer(ufr.timeouts); - if (urbt == NULL) - return; - - if (urbt->key <= current) { - fr_session = (struct fastrouter_session *) urbt->data; - fr_session->timed_out = 1; - if (fr_session->retry) { - fr_session->retry = 0; - uwsgi_fastrouter_switch_events(fr_session, -1, ufr.magic_table); - if (fr_session->retry) { - del_timeout(fr_session); - fr_session->timeout = add_fake_timeout(fr_session); - } - else { - fr_session->timeout = reset_timeout(fr_session); - } - } - else { - close_session(fr_session); - } - continue; - } - - break; - } -} - -void fr_get_hostname(char *key, uint16_t keylen, char *val, uint16_t vallen, void *data) { - - struct fastrouter_session *fr_session = (struct fastrouter_session *) data; - - //uwsgi_log("%.*s = %.*s\n", keylen, key, vallen, val); - if (!uwsgi_strncmp("SERVER_NAME", 11, key, keylen) && !fr_session->hostname_len) { - fr_session->hostname = val; - fr_session->hostname_len = vallen; - return; - } - - if (!uwsgi_strncmp("HTTP_HOST", 9, key, keylen) && !fr_session->has_key) { - fr_session->hostname = val; - fr_session->hostname_len = vallen; - return; - } - - if (!uwsgi_strncmp("UWSGI_FASTROUTER_KEY", 20, key, keylen)) { - fr_session->has_key = 1; - fr_session->hostname = val; - fr_session->hostname_len = vallen; - return; - } - - if (ufr.post_buffering > 0) { - if (!uwsgi_strncmp("CONTENT_LENGTH", 14, key, keylen)) { - fr_session->post_cl = uwsgi_str_num(val, vallen); - return; - } - } -} - -struct fastrouter_session *alloc_fr_session() { - - return uwsgi_calloc(sizeof(struct fastrouter_session)); -} - -void fastrouter_loop(int id) { - - int i; - - ufr.fr_stats_server = -1; - - ufr.fr_table = uwsgi_malloc(sizeof(struct fastrouter_session *) * uwsgi.max_fd); - - for (i = 0; i < (int) uwsgi.max_fd; i++) { - ufr.fr_table[i] = NULL; - } - - ufr.i_am_cheap = ufr.cheap; - - 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, ushared->gateways[id].internal_subscription_pipe[1]); - - - if (!ufr.socket_timeout) - ufr.socket_timeout = 30; - - if (!ufr.static_node_gracetime) - ufr.static_node_gracetime = 30; - - int i_am_the_first = 1; - for(i=0;igateways[i].name, "uWSGI fastrouter")) { - i_am_the_first = 0; - break; - } - } - - if (ufr.stats_server && i_am_the_first) { - char *tcp_port = strchr(ufr.stats_server, ':'); - if (tcp_port) { - // disable deferred accept for this socket - int current_defer_accept = uwsgi.no_defer_accept; - uwsgi.no_defer_accept = 1; - ufr.fr_stats_server = bind_to_tcp(ufr.stats_server, uwsgi.listen_queue, tcp_port); - uwsgi.no_defer_accept = current_defer_accept; - } - else { - ufr.fr_stats_server = bind_to_unix(ufr.stats_server, uwsgi.listen_queue, uwsgi.chmod_socket, uwsgi.abstract_socket); - } - - event_queue_add_fd_read(ufr.queue, ufr.fr_stats_server); - uwsgi_log("*** FastRouter stats server enabled on %s fd: %d ***\n", ufr.stats_server, ufr.fr_stats_server); - } - - - if (ufr.use_socket) { - ufr.to_socket = uwsgi_get_socket_by_num(ufr.socket_num); - if (ufr.to_socket) { - // fix socket name_len - if (ufr.to_socket->name_len == 0 && ufr.to_socket->name) { - ufr.to_socket->name_len = strlen(ufr.to_socket->name); - } - } - } - - if (!ufr.pb_base_dir) { - ufr.pb_base_dir = getenv("TMPDIR"); - if (!ufr.pb_base_dir) - ufr.pb_base_dir = "/tmp"; - } - - int nevents; - - time_t delta; - - struct uwsgi_rb_timer *min_timeout; - - int interesting_fd; - int new_connection; - - - if (ufr.pattern) { - init_magic_table(ufr.magic_table); - } - -#ifdef UWSGI_SCTP - uwsgi_fastrouter_sctp_nodes = uwsgi_calloc(sizeof(struct uwsgi_fastrouter_sctp_nodes*)); - uwsgi_fastrouter_sctp_nodes_current = uwsgi_calloc(sizeof(struct uwsgi_fastrouter_sctp_nodes*)); -#endif - - struct sockaddr_un fr_addr; - socklen_t fr_addr_len = sizeof(struct sockaddr_un); - - struct fastrouter_session *fr_session; - - ufr.mapper = uwsgi_fr_map_use_void; - - if (ufr.use_cache) { - ufr.mapper = uwsgi_fr_map_use_cache; - } - else if (ufr.pattern) { - ufr.mapper = uwsgi_fr_map_use_pattern; - } - else if (ufr.has_subscription_sockets) { - ufr.mapper = uwsgi_fr_map_use_subscription; - } - else if (ufr.base) { - ufr.mapper = uwsgi_fr_map_use_base; - } - else if (ufr.code_string_code && ufr.code_string_function) { - ufr.mapper = uwsgi_fr_map_use_cs; - } - else if (ufr.to_socket) { - ufr.mapper = uwsgi_fr_map_use_to; - } - else if (ufr.static_nodes) { - ufr.mapper = uwsgi_fr_map_use_static_nodes; - } -#ifdef UWSGI_SCTP - else if (ufr.has_sctp_sockets > 0) { - ufr.mapper = uwsgi_fr_map_use_sctp; - } -#endif - - - - ufr.timeouts = uwsgi_init_rb_timer(); - - for (;;) { - - min_timeout = uwsgi_min_rb_timer(ufr.timeouts); - if (min_timeout == NULL) { - delta = -1; - } - else { - delta = min_timeout->key - time(NULL); - if (delta <= 0) { - expire_timeouts(); - delta = 0; - } - } - - if (uwsgi.master_process && ufr.harakiri > 0) { - ushared->gateways_harakiri[id] = 0; - } - - nevents = event_queue_wait_multi(ufr.queue, delta, events, ufr.nevents); - - if (uwsgi.master_process && ufr.harakiri > 0) { - ushared->gateways_harakiri[id] = time(NULL) + ufr.harakiri; - } - - if (nevents == 0) { - expire_timeouts(); - } - - for (i = 0; i < nevents; i++) { - - interesting_fd = event_queue_interesting_fd(events, i); - - struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets; - int taken = 0; - while (ugs) { - if (ugs->gateway == &ushared->gateways[id] && interesting_fd == ugs->fd) { -#ifdef UWSGI_SCTP - if (!ugs->subscription && !ugs->sctp) { -#else - if (!ugs->subscription) { -#endif - - new_connection = accept(interesting_fd, (struct sockaddr *) &fr_addr, &fr_addr_len); - if (new_connection < 0) { - taken = 1; - break; - } - -#ifndef __linux__ - uwsgi_socket_b(new_connection); -#endif - - ufr.fr_table[new_connection] = alloc_fr_session(); - ufr.fr_table[new_connection]->fd = new_connection; - ufr.fr_table[new_connection]->instance_fd = -1; - ufr.fr_table[new_connection]->status = FASTROUTER_STATUS_RECV_HDR; - - ufr.fr_table[new_connection]->timeout = add_timeout(ufr.fr_table[new_connection]); - - event_queue_add_fd_read(ufr.queue, new_connection); - } - else if (ugs->subscription) { - uwsgi_corerouter_manage_subscription("uWSGI fastrouter", id, ugs, ufr.queue, &ufr.subscriptions, - ufr.subscription_regexp, fastrouter_manage_subscription, ufr.cheap, &ufr.i_am_cheap); - } -#ifdef UWSGI_SCTP - else if (ugs->sctp) { - new_connection = accept(interesting_fd, (struct sockaddr *) &fr_addr, &fr_addr_len); - if (new_connection < 0) { - taken = 1; - break; - } - struct uwsgi_fr_sctp_node *sctp_node = uwsgi_fr_sctp_add_node(new_connection); - snprintf(sctp_node->name, 64, "%s:%d", inet_ntoa(((struct sockaddr_in *)&fr_addr)->sin_addr), ntohs(((struct sockaddr_in *) &fr_addr)->sin_port)); - uwsgi_log("new SCTP peer: %s:%d\n", inet_ntoa(((struct sockaddr_in *)&fr_addr)->sin_addr), ntohs(((struct sockaddr_in *) &fr_addr)->sin_port)); - - ufr.fr_table[new_connection] = alloc_fr_session(); - ufr.fr_table[new_connection]->instance_fd = new_connection; - ufr.fr_table[new_connection]->fd = -1; - ufr.fr_table[new_connection]->persistent = 1; - ufr.fr_table[new_connection]->status = FASTROUTER_STATUS_SCTP_NODE_FREE; - - struct sctp_event_subscribe events; - memset(&events, 0, sizeof(events) ); - events.sctp_data_io_event = 1; - // check for errors - setsockopt(new_connection, SOL_SCTP, SCTP_EVENTS, &events, sizeof(events) ); - - event_queue_add_fd_read(ufr.queue, new_connection); - } -#endif - - taken = 1; - break; - } - - - ugs = ugs->next; - } - - if (taken) { - continue; - } - - if (interesting_fd == ushared->gateways[id].internal_subscription_pipe[1]) { - 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); - } - else { - fr_session = ufr.fr_table[interesting_fd]; - - // something is going wrong... - if (fr_session == NULL) - continue; - - if (event_queue_interesting_fd_has_error(events, i)) { -#ifdef UWSGI_SCTP - if (!fr_session->persistent) { -#endif - close_session(fr_session); - continue; -#ifdef UWSGI_SCTP - } -#endif - } - -#ifdef UWSGI_SCTP - if (!fr_session->persistent) { -#endif - fr_session->timeout = reset_timeout(fr_session); -#ifdef UWSGI_SCTP - } -#endif - - uwsgi_fastrouter_switch_events(fr_session, interesting_fd, ufr.magic_table); - - - } - } - } - -} - int fastrouter_init() { - int i; - - if (ufr.has_sockets) { - - if (ufr.use_cache && !uwsgi.cache_max_items) { - uwsgi_log("you need to create a uwsgi cache to use the fastrouter (add --cache )\n"); - exit(1); - } - - if (!ufr.nevents) - ufr.nevents = 64; - - 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"); - exit(1); - } - } - } + ufr.cr.session_size = sizeof(struct fastrouter_session); + ufr.cr.switch_events = uwsgi_fastrouter_switch_events; + uwsgi_corerouter_init((struct uwsgi_corerouter *) &ufr); return 0; } +void fastrouter_setup() { + ufr.cr.name = uwsgi_str("uWSGI fastrouter"); +} + struct uwsgi_plugin fastrouter_plugin = { .name = "fastrouter", .options = fastrouter_options, .init = fastrouter_init, + .on_load = fastrouter_setup }; - - -#define stats_send_llu(x, y) fprintf(output, x, (long long unsigned int) y) -#define stats_send(x, y) fprintf(output, x, y) - -void fastrouter_send_stats(int fd) { - - struct sockaddr_un client_src; - socklen_t client_src_len = 0; - int client_fd = accept(fd, (struct sockaddr *) &client_src, &client_src_len); - if (client_fd < 0) { - uwsgi_error("accept()"); - return; - } - - FILE *output = fdopen(client_fd, "w"); - if (!output) { - uwsgi_error("fdopen()"); - close(client_fd); - return; - } - - stats_send("{ \"version\": \"%s\",\n", UWSGI_VERSION); - - fprintf(output, "\"pid\": %d,\n", (int) (getpid())); - fprintf(output, "\"uid\": %d,\n", (int) (getuid())); - fprintf(output, "\"gid\": %d,\n", (int) (getgid())); - - char *cwd = uwsgi_get_cwd(); - stats_send("\"cwd\": \"%s\",\n", cwd); - free(cwd); - - fprintf(output, "\"fastrouter\": ["); - struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets; - while (ugs) { - if (!strcmp(ugs->owner, "uWSGI fastrouter")) { - if (ugs->next) { - stats_send("\"%s\",", ugs->name); - } - else { - stats_send("\"%s\"", ugs->name); - } - } - ugs = ugs->next; - } - fprintf(output, "],\n"); - - if (ufr.has_subscription_sockets) { - fprintf(output, "\"subscriptions\": [\n"); - struct uwsgi_subscribe_slot *s_slot = ufr.subscriptions; - while (s_slot) { - fprintf(output, "\t{ \"key\": \"%.*s\",\n", s_slot->keylen, s_slot->key); - fprintf(output, "\t\t\"hits\": %llu,\n", (unsigned long long) s_slot->hits); - fprintf(output, "\t\t\"nodes\": [\n"); - struct uwsgi_subscribe_node *s_node = s_slot->nodes; - while (s_node) { - fprintf(output, "\t\t\t{\"name\": \"%.*s\", \"modifier1\": %d, \"modifier2\": %d, \"last_check\": %llu, \"requests\": %llu, \"tx\": %llu, \"cores\": %llu, \"load\": %llu, \"weight\": %llu, \"wrr\": %llu, \"ref\": %llu, \"failcnt\": %llu, \"death_mark\": %d}", s_node->len, s_node->name, s_node->modifier1, s_node->modifier2, (unsigned long long) s_node->last_check, (unsigned long long) s_node->requests, (unsigned long long) s_node->transferred, (unsigned long long) s_node->cores, (unsigned long long) s_node->load, (unsigned long long) s_node->weight, (unsigned long long) s_node->wrr, (unsigned long long) s_node->reference, (unsigned long long) s_node->failcnt, s_node->death_mark); - if (s_node->next) { - fprintf(output, ",\n"); - } - else { - fprintf(output, "\n"); - } - s_node = s_node->next; - } - fprintf(output, "\t\t]\n"); - if (s_slot->next) { - fprintf(output, "\t},\n"); - } - else { - fprintf(output, "\t}\n"); - } - s_slot = s_slot->next; - // check for loopy optimization - if (s_slot == ufr.subscriptions) - break; - } - fprintf(output, "],\n"); - } - -#ifdef UWSGI_SCTP - if (ufr.has_sctp_sockets > 0) { - fprintf(output, "\"sctp_nodes\": [\n"); - struct uwsgi_fr_sctp_node *sctp_node = *uwsgi_fastrouter_sctp_nodes; - while(sctp_node) { - fprintf(output, "\t{ \"node\": \"%s\", \"requests\": %llu }", sctp_node->name, (unsigned long long) sctp_node->requests); - if (sctp_node->next == *uwsgi_fastrouter_sctp_nodes) { - fprintf(output, "\n"); - break; - } - sctp_node = sctp_node->next; - fprintf(output, ",\n"); - } - fprintf(output, "],\n"); - } -#endif - - fprintf(output, "\"cheap\": %d\n", ufr.i_am_cheap); - - fprintf(output, "}\n"); - fclose(output); - -} diff --git a/plugins/fastrouter/fr.h b/plugins/fastrouter/fr.h index f8ea1854..ccd50269 100644 --- a/plugins/fastrouter/fr.h +++ b/plugins/fastrouter/fr.h @@ -1,164 +1,19 @@ -#define FASTROUTER_STATUS_FREE 0 -#define FASTROUTER_STATUS_CONNECTING 1 -#define FASTROUTER_STATUS_RECV_HDR 2 -#define FASTROUTER_STATUS_RECV_VARS 3 -#define FASTROUTER_STATUS_RESPONSE 4 -#define FASTROUTER_STATUS_BUFFERING 5 +#include "../corerouter/cr.h" -#ifdef UWSGI_SCTP -#define FASTROUTER_STATUS_SCTP_NODE_FREE 6 -#define FASTROUTER_STATUS_SCTP_RESPONSE 7 -#endif - -#define add_timeout(x) uwsgi_add_rb_timer(ufr.timeouts, time(NULL)+ufr.socket_timeout, x) -#define add_fake_timeout(x) uwsgi_add_rb_timer(ufr.timeouts, time(NULL)+1, x) -#define add_check_timeout(x) uwsgi_add_rb_timer(timeouts, time(NULL)+x, NULL) -#define del_check_timeout(x) rb_erase(&x->rbt, timeouts); -#define del_timeout(x) rb_erase(&x->timeout->rbt, ufr.timeouts); free(x->timeout); - -struct fastrouter_session; +#define FASTROUTER_STATUS_RECV_VARS 10 +#define FASTROUTER_STATUS_BUFFERING 11 struct uwsgi_fastrouter { - int (*mapper)(struct fastrouter_session *, char **); - - int has_sockets; - int has_subscription_sockets; -#ifdef UWSGI_SCTP - int has_sctp_sockets; -#endif - - int processes; - int quiet; - - struct rb_root *timeouts; - - int use_cache; - int nevents; - - char *magic_table[0xff]; - - int queue; - - char *pattern; - int pattern_len; - - char *base; - int base_len; - - size_t post_buffering; - char *pb_base_dir; - - struct uwsgi_string_list *static_nodes; - struct uwsgi_string_list *current_static_node; - int static_node_gracetime; - - char *stats_server; - int fr_stats_server; - - int use_socket; - int socket_num; - struct uwsgi_socket *to_socket; - - struct uwsgi_subscribe_slot *subscriptions; - int subscription_regexp; - - struct uwsgi_string_list *fallback; - - int socket_timeout; - - uint8_t code_string_modifier1; - char *code_string_code; - char *code_string_function; - - - struct uwsgi_rb_timer *subscriptions_check; - - int cheap; - int i_am_cheap; - - int tolerance; - int harakiri; - - struct fastrouter_session **fr_table; + struct uwsgi_corerouter cr; }; -#ifdef UWSGI_SCTP -struct uwsgi_fr_sctp_node { - - int fd; - char name[64]; - uint64_t requests; - struct uwsgi_fr_sctp_node *prev; - struct uwsgi_fr_sctp_node *next; -}; - -struct uwsgi_fr_sctp_node *uwsgi_fr_sctp_add_node(int); -void uwsgi_fr_sctp_del_node(int); -void uwsgi_opt_fastrouter_sctp(char *, char *, void *); - -#endif - struct fastrouter_session { - int fd; - int instance_fd; - int status; - struct uwsgi_header uh; - uint8_t h_pos; - uint16_t pos; - - char *hostname; - uint16_t hostname_len; - - int has_key; - int retry; -#ifdef UWSGI_SCTP - int persistent; -#endif - - char *instance_address; - uint64_t instance_address_len; - - struct uwsgi_subscribe_node *un; - struct uwsgi_string_list *static_node; - int pass_fd; - int soopt; - int timed_out; - - struct uwsgi_rb_timer *timeout; - int instance_failed; - - size_t post_cl; - size_t post_remains; - - struct uwsgi_string_list *fallback; - - char *buf_file_name; - FILE *buf_file; - - uint8_t modifier1; - uint8_t modifier2; - - char *tmp_socket_name; - - char buffer[0xffff]; + struct corerouter_session crs; + char buffer[UMAX16]; }; +void uwsgi_fastrouter_switch_events(struct uwsgi_corerouter *, struct corerouter_session *, int interesting_fd); -void uwsgi_fastrouter_switch_events(struct fastrouter_session *, int intersting_fd, char **); -void close_session(struct fastrouter_session *); -void fr_get_hostname(char *, uint16_t, char *, uint16_t, void *); - -int uwsgi_fr_map_use_void(struct fastrouter_session *, char **); -int uwsgi_fr_map_use_cache(struct fastrouter_session *, char **); -int uwsgi_fr_map_use_pattern(struct fastrouter_session *, char **); -int uwsgi_fr_map_use_subscription(struct fastrouter_session *, char **); -int uwsgi_fr_map_use_base(struct fastrouter_session *, char **); -int uwsgi_fr_map_use_cs(struct fastrouter_session *, char **); -int uwsgi_fr_map_use_to(struct fastrouter_session *, char **); -int uwsgi_fr_map_use_static_nodes(struct fastrouter_session *, char **); -#ifdef UWSGI_SCTP -int uwsgi_fr_map_use_sctp(struct fastrouter_session *, char **); -#endif diff --git a/plugins/fastrouter/fr_events.c b/plugins/fastrouter/fr_events.c index 258e69f0..52a476c4 100644 --- a/plugins/fastrouter/fr_events.c +++ b/plugins/fastrouter/fr_events.c @@ -5,7 +5,42 @@ extern struct uwsgi_server uwsgi; extern struct uwsgi_fastrouter ufr; -void uwsgi_fastrouter_switch_events(struct fastrouter_session *fr_session, int interesting_fd, char **magic_table) { +void fr_get_hostname(char *key, uint16_t keylen, char *val, uint16_t vallen, void *data) { + + // here i use directly corerouter_session + struct corerouter_session *fr_session = (struct corerouter_session *) data; + + //uwsgi_log("%.*s = %.*s\n", keylen, key, vallen, val); + if (!uwsgi_strncmp("SERVER_NAME", 11, key, keylen) && !fr_session->hostname_len) { + fr_session->hostname = val; + fr_session->hostname_len = vallen; + return; + } + + if (!uwsgi_strncmp("HTTP_HOST", 9, key, keylen) && !fr_session->has_key) { + fr_session->hostname = val; + fr_session->hostname_len = vallen; + return; + } + + if (!uwsgi_strncmp("UWSGI_FASTROUTER_KEY", 20, key, keylen)) { + fr_session->has_key = 1; + fr_session->hostname = val; + fr_session->hostname_len = vallen; + return; + } + + if (ufr.cr.post_buffering > 0) { + if (!uwsgi_strncmp("CONTENT_LENGTH", 14, key, keylen)) { + fr_session->post_cl = uwsgi_str_num(val, vallen); + return; + } + } + } + +void uwsgi_fastrouter_switch_events(struct uwsgi_corerouter *ucr, struct corerouter_session *cs, int interesting_fd) { + + struct fastrouter_session *fr_session = (struct fastrouter_session *) cs; socklen_t solen = sizeof(int); struct iovec iov[2]; @@ -18,24 +53,24 @@ void uwsgi_fastrouter_switch_events(struct fastrouter_session *fr_session, int i struct cmsghdr *cmsg; ssize_t len; - char *post_tmp_buf[0xffff]; + char *post_tmp_buf[UMAX16]; - switch (fr_session->status) { + switch (cs->status) { - case FASTROUTER_STATUS_RECV_HDR: - len = recv(fr_session->fd, (char *) (&fr_session->uh) + fr_session->h_pos, 4 - fr_session->h_pos, 0); + case COREROUTER_STATUS_RECV_HDR: + len = recv(cs->fd, (char *) (&cs->uh) + cs->h_pos, 4 - cs->h_pos, 0); if (len <= 0) { if (len < 0) uwsgi_error("recv()"); - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } - fr_session->h_pos += len; - if (fr_session->h_pos == 4) { + cs->h_pos += len; + if (cs->h_pos == 4) { #ifdef UWSGI_DEBUG - uwsgi_log("modifier1: %d pktsize: %d modifier2: %d\n", fr_session->uh.modifier1, fr_session->uh.pktsize, fr_session->uh.modifier2); + uwsgi_log("modifier1: %d pktsize: %d modifier2: %d\n", cs->uh.modifier1, cs->uh.pktsize, cs->uh.modifier2); #endif - fr_session->status = FASTROUTER_STATUS_RECV_VARS; + cs->status = FASTROUTER_STATUS_RECV_VARS; } break; @@ -46,76 +81,76 @@ void uwsgi_fastrouter_switch_events(struct fastrouter_session *fr_session, int i goto choose_node; } - len = recv(fr_session->fd, fr_session->buffer + fr_session->pos, fr_session->uh.pktsize - fr_session->pos, 0); + len = recv(cs->fd, fr_session->buffer + cs->pos, cs->uh.pktsize - cs->pos, 0); if (len <= 0) { uwsgi_error("recv()"); - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } - fr_session->pos += len; - if (fr_session->pos == fr_session->uh.pktsize) { - if (uwsgi_hooked_parse(fr_session->buffer, fr_session->uh.pktsize, fr_get_hostname, (void *) fr_session)) { - close_session(fr_session); + cs->pos += len; + if (cs->pos == cs->uh.pktsize) { + if (uwsgi_hooked_parse(fr_session->buffer, cs->uh.pktsize, fr_get_hostname, (void *) fr_session)) { + corerouter_close_session(ucr, cs); break; } - if (fr_session->hostname_len == 0) { - close_session(fr_session); + if (cs->hostname_len == 0) { + corerouter_close_session(ucr, cs); break; } #ifdef UWSGI_DEBUG - //uwsgi_log("requested domain %.*s\n", fr_session->hostname_len, fr_session->hostname); + //uwsgi_log("requested domain %.*s\n", cs->hostname_len, cs->hostname); #endif // the mapper hook choose_node: - if (ufr.mapper(fr_session, magic_table)) + if (ucr->mapper(ucr, cs)) break; // no address found - if (!fr_session->instance_address_len) { + if (!cs->instance_address_len) { // if fallback nodes are configured, trigger them - if (ufr.fallback) { - fr_session->instance_failed = 1; + if (ucr->fallback) { + cs->instance_failed = 1; } - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } - if (ufr.post_buffering > 0 && fr_session->post_cl > ufr.post_buffering) { - fr_session->status = FASTROUTER_STATUS_BUFFERING; - fr_session->buf_file_name = uwsgi_tmpname(ufr.pb_base_dir, "uwsgiXXXXX"); - if (!fr_session->buf_file_name) { + if (ucr->post_buffering > 0 && cs->post_cl > ucr->post_buffering) { + cs->status = FASTROUTER_STATUS_BUFFERING; + cs->buf_file_name = uwsgi_tmpname(ucr->pb_base_dir, "uwsgiXXXXX"); + if (!cs->buf_file_name) { uwsgi_error("tempnam()"); - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } - fr_session->post_remains = fr_session->post_cl; + cs->post_remains = cs->post_cl; - // 2 + UWSGI_POSTFILE + 2 + fr_session->buf_file_name - if (fr_session->uh.pktsize + (2 + 14 + 2 + strlen(fr_session->buf_file_name)) > 0xffff) { - uwsgi_log("unable to buffer request body to file %s: not enough space\n", fr_session->buf_file_name); - close_session(fr_session); + // 2 + UWSGI_POSTFILE + 2 + cs->buf_file_name + if (cs->uh.pktsize + (2 + 14 + 2 + strlen(cs->buf_file_name)) > UMAX16) { + uwsgi_log("unable to buffer request body to file %s: not enough space\n", cs->buf_file_name); + corerouter_close_session(ucr, cs); break; } - char *ptr = fr_session->buffer + fr_session->uh.pktsize; - uint16_t bfn_len = strlen(fr_session->buf_file_name); + char *ptr = fr_session->buffer + cs->uh.pktsize; + uint16_t bfn_len = strlen(cs->buf_file_name); *ptr++ = 14; *ptr++ = 0; memcpy(ptr, "UWSGI_POSTFILE", 14); ptr += 14; *ptr++ = (char) (bfn_len & 0xff); *ptr++ = (char) ((bfn_len >> 8) & 0xff); - memcpy(ptr, fr_session->buf_file_name, bfn_len); - fr_session->uh.pktsize += 2 + 14 + 2 + bfn_len; + memcpy(ptr, cs->buf_file_name, bfn_len); + cs->uh.pktsize += 2 + 14 + 2 + bfn_len; - fr_session->buf_file = fopen(fr_session->buf_file_name, "w"); - if (!fr_session->buf_file) { - uwsgi_error_open(fr_session->buf_file_name); - close_session(fr_session); + cs->buf_file = fopen(cs->buf_file_name, "w"); + if (!cs->buf_file) { + uwsgi_error_open(cs->buf_file_name); + corerouter_close_session(ucr, cs); break; } @@ -123,57 +158,57 @@ void uwsgi_fastrouter_switch_events(struct fastrouter_session *fr_session, int i else { - fr_session->pass_fd = is_unix(fr_session->instance_address, fr_session->instance_address_len); + cs->pass_fd = is_unix(cs->instance_address, cs->instance_address_len); - fr_session->instance_fd = uwsgi_connectn(fr_session->instance_address, fr_session->instance_address_len, 0, 1); + cs->instance_fd = uwsgi_connectn(cs->instance_address, cs->instance_address_len, 0, 1); - if (fr_session->instance_fd < 0) { - fr_session->instance_failed = 1; - fr_session->soopt = errno; - close_session(fr_session); + if (cs->instance_fd < 0) { + cs->instance_failed = 1; + cs->soopt = errno; + corerouter_close_session(ucr, cs); break; } - fr_session->status = FASTROUTER_STATUS_CONNECTING; - ufr.fr_table[fr_session->instance_fd] = fr_session; - event_queue_add_fd_write(ufr.queue, fr_session->instance_fd); + cs->status = COREROUTER_STATUS_CONNECTING; + ucr->cr_table[cs->instance_fd] = cs; + event_queue_add_fd_write(ucr->queue, cs->instance_fd); } } break; - case FASTROUTER_STATUS_CONNECTING: + case COREROUTER_STATUS_CONNECTING: - if (interesting_fd == fr_session->instance_fd) { + if (interesting_fd == cs->instance_fd) { - if (getsockopt(fr_session->instance_fd, SOL_SOCKET, SO_ERROR, (void *) (&fr_session->soopt), &solen) < 0) { + if (getsockopt(cs->instance_fd, SOL_SOCKET, SO_ERROR, (void *) (&cs->soopt), &solen) < 0) { uwsgi_error("getsockopt()"); - fr_session->instance_failed = 1; - close_session(fr_session); + cs->instance_failed = 1; + corerouter_close_session(ucr, cs); break; } - if (fr_session->soopt) { - fr_session->instance_failed = 1; - close_session(fr_session); + if (cs->soopt) { + cs->instance_failed = 1; + corerouter_close_session(ucr, cs); break; } - fr_session->uh.modifier1 = fr_session->modifier1; + cs->uh.modifier1 = cs->modifier1; - iov[0].iov_base = &fr_session->uh; + iov[0].iov_base = &cs->uh; iov[0].iov_len = 4; iov[1].iov_base = fr_session->buffer; - iov[1].iov_len = fr_session->uh.pktsize; + iov[1].iov_len = cs->uh.pktsize; // increment node requests counter - if (fr_session->un) - fr_session->un->requests++; + if (cs->un) + cs->un->requests++; // fd passing: PERFORMANCE EXTREME BOOST !!! - if (fr_session->pass_fd && !uwsgi.no_fd_passing) { + if (cs->pass_fd && !uwsgi.no_fd_passing) { msg.msg_name = NULL; msg.msg_namelen = 0; msg.msg_iov = iov; @@ -187,24 +222,24 @@ void uwsgi_fastrouter_switch_events(struct fastrouter_session *fr_session, int i cmsg->cmsg_level = SOL_SOCKET; cmsg->cmsg_type = SCM_RIGHTS; - memcpy(CMSG_DATA(cmsg), &fr_session->fd, sizeof(int)); + memcpy(CMSG_DATA(cmsg), &cs->fd, sizeof(int)); - if (sendmsg(fr_session->instance_fd, &msg, 0) < 0) { + if (sendmsg(cs->instance_fd, &msg, 0) < 0) { uwsgi_error("sendmsg()"); } - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } - if (writev(fr_session->instance_fd, iov, 2) < 0) { + if (writev(cs->instance_fd, iov, 2) < 0) { uwsgi_error("writev()"); - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } - event_queue_fd_write_to_read(ufr.queue, fr_session->instance_fd); - fr_session->status = FASTROUTER_STATUS_RESPONSE; + event_queue_fd_write_to_read(ucr->queue, cs->instance_fd); + cs->status = COREROUTER_STATUS_RESPONSE; } break; @@ -216,15 +251,15 @@ void uwsgi_fastrouter_switch_events(struct fastrouter_session *fr_session, int i int msg_flags = 0; memset(&sinfo, 0, sizeof(struct sctp_sndrcvinfo)); - len = sctp_recvmsg(interesting_fd, fr_session->buffer, 0xffff, NULL, NULL, &sinfo, &msg_flags); + len = sctp_recvmsg(interesting_fd, fr_session->buffer, UMAX16, NULL, NULL, &sinfo, &msg_flags); // remove the SCTP node uwsgi_log("[0] removing SCTP node %d flags = %d len = %d\n", interesting_fd, msg_flags, len); uwsgi_fr_sctp_del_node(interesting_fd); - if (ufr.fr_table[interesting_fd]->timeout) { - del_timeout(ufr.fr_table[interesting_fd]); + if (ucr->fr_table[interesting_fd]->timeout) { + del_timeout(ucr->fr_table[interesting_fd]); } - free(ufr.fr_table[interesting_fd]); - ufr.fr_table[interesting_fd] = NULL; + free(ucr->fr_table[interesting_fd]); + ucr->fr_table[interesting_fd] = NULL; close(interesting_fd); } @@ -232,33 +267,33 @@ void uwsgi_fastrouter_switch_events(struct fastrouter_session *fr_session, int i case FASTROUTER_STATUS_SCTP_RESPONSE: // data from instance - if (interesting_fd == fr_session->instance_fd) { + if (interesting_fd == cs->instance_fd) { struct sctp_sndrcvinfo sinfo; struct uwsgi_header *uh; int msg_flags = 0; memset(&sinfo, 0, sizeof(struct sctp_sndrcvinfo)); - len = sctp_recvmsg(fr_session->instance_fd, fr_session->buffer, 0xffff, NULL, NULL, &sinfo, &msg_flags); + len = sctp_recvmsg(cs->instance_fd, cs->buffer, UMAX16, NULL, NULL, &sinfo, &msg_flags); if (len <= 0) { if (len < 0) uwsgi_error("recv()"); - close_session(ufr.fr_table[fr_session->fd]); + close_session(ucr->fr_table[cs->fd]); // REMOVE THE NODE uwsgi_log("[1] removing SCTP node %d flags = %d len = %d\n", interesting_fd, msg_flags, len); uwsgi_fr_sctp_del_node(interesting_fd); - if (ufr.fr_table[interesting_fd]->timeout) { - del_timeout(ufr.fr_table[interesting_fd]); + if (ucr->fr_table[interesting_fd]->timeout) { + del_timeout(ucr->fr_table[interesting_fd]); } - free(ufr.fr_table[interesting_fd]); - ufr.fr_table[interesting_fd] = NULL; + free(ucr->fr_table[interesting_fd]); + ucr->fr_table[interesting_fd] = NULL; close(interesting_fd); break; } - if (fr_session->fd != -1 && sinfo.sinfo_stream != fr_session->fd) { - if (fr_session->fd != -1) { + if (cs->fd != -1 && sinfo.sinfo_stream != cs->fd) { + if (cs->fd != -1) { uwsgi_log("INVALID SCTP STREAM !!!\n"); - close_session(ufr.fr_table[fr_session->fd]); + close_session(ucr->fr_table[cs->fd]); } break; } @@ -267,40 +302,40 @@ void uwsgi_fastrouter_switch_events(struct fastrouter_session *fr_session, int i // check for close packet if (uh->modifier1 == 200) { - fr_session->status = FASTROUTER_STATUS_SCTP_NODE_FREE; - if (fr_session->fd != -1) { - close_session(ufr.fr_table[fr_session->fd]); + cs->status = FASTROUTER_STATUS_SCTP_NODE_FREE; + if (cs->fd != -1) { + close_session(ucr->fr_table[cs->fd]); } break; } - if (fr_session->fd == -1) { + if (cs->fd == -1) { break; } - len = send(fr_session->fd, fr_session->buffer, len, 0); + len = send(cs->fd, cs->buffer, len, 0); if (len <= 0) { if (len < 0) uwsgi_error("send()"); - close_session(ufr.fr_table[fr_session->fd]); + close_session(ucr->fr_table[cs->fd]); break; } // update transfer statistics - if (fr_session->un) - fr_session->un->transferred += len; + if (cs->un) + cs->un->transferred += len; } // body from client - else if (interesting_fd == fr_session->fd) { + else if (interesting_fd == cs->fd) { - len = recv(fr_session->fd, fr_session->buffer, 0xffff, 0); + len = recv(cs->fd, fr_session->buffer, UMAX16, 0); if (len <= 0) { if (len < 0) uwsgi_error("recv()"); // mark session as broken - ufr.fr_table[fr_session->instance_fd]->fd = -1; + ucr->fr_table[cs->instance_fd]->fd = -1; close_session(fr_session); break; } @@ -313,9 +348,9 @@ void uwsgi_fastrouter_switch_events(struct fastrouter_session *fr_session, int i uh.pktsize = 0; uh.modifier2 = 0; memcpy(&sinfo.sinfo_ppid, &uh, sizeof(uint32_t)); - sinfo.sinfo_stream = fr_session->fd; + sinfo.sinfo_stream = cs->fd; - len = sctp_send(fr_session->instance_fd, fr_session->buffer, len, &sinfo, 0); + len = sctp_send(cs->instance_fd, cs->buffer, len, &sinfo, 0); if (len <= 0) { if (len < 0) @@ -327,50 +362,50 @@ void uwsgi_fastrouter_switch_events(struct fastrouter_session *fr_session, int i break; #endif - case FASTROUTER_STATUS_RESPONSE: + case COREROUTER_STATUS_RESPONSE: // data from instance - if (interesting_fd == fr_session->instance_fd) { - len = recv(fr_session->instance_fd, fr_session->buffer, 0xffff, 0); + if (interesting_fd == cs->instance_fd) { + len = recv(cs->instance_fd, fr_session->buffer, UMAX16, 0); if (len <= 0) { if (len < 0) uwsgi_error("recv()"); - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } - len = send(fr_session->fd, fr_session->buffer, len, 0); + len = send(cs->fd, fr_session->buffer, len, 0); if (len <= 0) { if (len < 0) uwsgi_error("send()"); - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } // update transfer statistics - if (fr_session->un) - fr_session->un->transferred += len; + if (cs->un) + cs->un->transferred += len; } // body from client - else if (interesting_fd == fr_session->fd) { + else if (interesting_fd == cs->fd) { //uwsgi_log("receiving body...\n"); - len = recv(fr_session->fd, fr_session->buffer, 0xffff, 0); + len = recv(cs->fd, fr_session->buffer, UMAX16, 0); if (len <= 0) { if (len < 0) uwsgi_error("recv()"); - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } - len = send(fr_session->instance_fd, fr_session->buffer, len, 0); + len = send(cs->instance_fd, fr_session->buffer, len, 0); if (len <= 0) { if (len < 0) uwsgi_error("send()"); - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } } @@ -378,40 +413,40 @@ void uwsgi_fastrouter_switch_events(struct fastrouter_session *fr_session, int i break; case FASTROUTER_STATUS_BUFFERING: - len = recv(fr_session->fd, post_tmp_buf, UMIN(0xffff, fr_session->post_remains), 0); + len = recv(cs->fd, post_tmp_buf, UMIN(UMAX16, cs->post_remains), 0); if (len <= 0) { if (len < 0) uwsgi_error("recv()"); - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } - if (fwrite(post_tmp_buf, len, 1, fr_session->buf_file) != 1) { + if (fwrite(post_tmp_buf, len, 1, cs->buf_file) != 1) { uwsgi_error("fwrite()"); - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } - fr_session->post_remains -= len; + cs->post_remains -= len; - if (fr_session->post_remains == 0) { + if (cs->post_remains == 0) { // close the buf_file ASAP - fclose(fr_session->buf_file); - fr_session->buf_file = NULL; + fclose(cs->buf_file); + cs->buf_file = NULL; - fr_session->pass_fd = is_unix(fr_session->instance_address, fr_session->instance_address_len); + cs->pass_fd = is_unix(cs->instance_address, cs->instance_address_len); - fr_session->instance_fd = uwsgi_connectn(fr_session->instance_address, fr_session->instance_address_len, 0, 1); + cs->instance_fd = uwsgi_connectn(cs->instance_address, cs->instance_address_len, 0, 1); - if (fr_session->instance_fd < 0) { - fr_session->instance_failed = 1; - close_session(fr_session); + if (cs->instance_fd < 0) { + cs->instance_failed = 1; + corerouter_close_session(ucr, cs); break; } - fr_session->status = FASTROUTER_STATUS_CONNECTING; - ufr.fr_table[fr_session->instance_fd] = fr_session; - event_queue_add_fd_write(ufr.queue, fr_session->instance_fd); + cs->status = COREROUTER_STATUS_CONNECTING; + ucr->cr_table[cs->instance_fd] = cs; + event_queue_add_fd_write(ucr->queue, cs->instance_fd); } break; @@ -421,7 +456,7 @@ void uwsgi_fastrouter_switch_events(struct fastrouter_session *fr_session, int i // fallback to destroy !!! default: uwsgi_log("unknown event: closing session\n"); - close_session(fr_session); + corerouter_close_session(ucr, cs); break; } diff --git a/plugins/fastrouter/uwsgiplugin.py b/plugins/fastrouter/uwsgiplugin.py index b794a215..32ca4033 100644 --- a/plugins/fastrouter/uwsgiplugin.py +++ b/plugins/fastrouter/uwsgiplugin.py @@ -4,4 +4,4 @@ CFLAGS = [] LDFLAGS = [] LIBS = [] -GCC_LIST = ['fr_sctp', 'fr_map', 'fastrouter', 'fr_events'] +GCC_LIST = ['fastrouter', 'fr_events'] diff --git a/plugins/http/http.c b/plugins/http/http.c index 8ba34220..c1fb85e1 100644 --- a/plugins/http/http.c +++ b/plugins/http/http.c @@ -14,234 +14,66 @@ extern struct uwsgi_server uwsgi; -#include "../../lib/corerouter.h" +#include "../corerouter/cr.h" #define MAX_HTTP_VEC 128 -#define HTTP_STATUS_FREE 0 -#define HTTP_STATUS_CONNECTING 1 -#define HTTP_STATUS_RECV 2 -#define HTTP_STATUS_RESPONSE 4 - -#define add_timeout(x) uwsgi_add_rb_timer(uhttp.timeouts, time(NULL)+uhttp.socket_timeout, x) -#define del_timeout(x) rb_erase(&x->timeout->rbt, uhttp.timeouts); free(x->timeout); - struct uwsgi_http { - int has_sockets; - - int processes; - - int use_cache; - int use_cluster; - int nevents; - - int cheap; - int i_am_cheap; - - int has_subscription_sockets; - int subscription_regexp; - - char *pattern; - int pattern_len; - - char *base; - int base_len; - - char *to; - int to_len; + struct uwsgi_corerouter cr; + uint16_t modifier1; struct uwsgi_string_list *http_vars; - - uint8_t modifier1; - int load; - - int socket_timeout; - - struct uwsgi_subscribe_slot *subscriptions; - int manage_expect; - struct rb_root *timeouts; } uhttp; -void uwsgi_opt_http(char *opt, char *value, void *foobar) { - - uwsgi_new_gateway_socket(value, "uWSGI http"); - uhttp.has_sockets++; -} - -void uwsgi_opt_http_ss(char *opt, char *value, void *foobar) { - - struct uwsgi_gateway_socket *ugs = uwsgi_new_gateway_socket(value, "uWSGI http"); - ugs->subscription = 1; - uhttp.has_subscription_sockets++; - -} - -void uwsgi_opt_http_use_pattern(char *opt, char *value, void *foobar) { - uhttp.pattern = value; - // optimization - uhttp.pattern_len = strlen(uhttp.pattern); -} - -void uwsgi_opt_http_use_base(char *opt, char *value, void *foobar) { - uhttp.base = value; - // optimization - uhttp.base_len = strlen(uhttp.base); -} - -void uwsgi_opt_http_use_to(char *opt, char *value, void *foobar) { - uhttp.to = value; - // optimization - uhttp.to_len = strlen(uhttp.to); -} - - struct uwsgi_option http_options[] = { - {"http", required_argument, 0, "add an http router/server on the specified address", uwsgi_opt_http, NULL, 0}, - {"http-processes", required_argument, 0, "set the number of http processes to spawn", uwsgi_opt_set_int, &uhttp.processes, 0}, - {"http-workers", required_argument, 0, "set the number of http processes to spawn", uwsgi_opt_set_int, &uhttp.processes, 0}, + {"http", required_argument, 0, "add an http router/server on the specified address", uwsgi_opt_corerouter, NULL, 0}, + {"http-processes", required_argument, 0, "set the number of http processes to spawn", uwsgi_opt_set_int, &uhttp.cr.processes, 0}, + {"http-workers", required_argument, 0, "set the number of http processes to spawn", uwsgi_opt_set_int, &uhttp.cr.processes, 0}, {"http-var", required_argument, 0, "add a key=value item to the generated uwsgi packet", uwsgi_opt_add_string_list, &uhttp.http_vars, 0}, - {"http-to", required_argument, 0, "forward requests to the specified node", uwsgi_opt_http_use_to, NULL, 0 }, + //{"http-to", required_argument, 0, "forward requests to the specified node", uwsgi_opt_corerouter_use_to, NULL, 0 }, {"http-modifier1", required_argument, 0, "set uwsgi protocol modifier1", uwsgi_opt_set_int, &uhttp.modifier1, 0}, - {"http-use-cache", no_argument, 0, "use uWSGI cache as key->value virtualhost mapper", uwsgi_opt_true, &uhttp.use_cache, 0}, - {"http-use-pattern", required_argument, 0, "use the specified pattern for mapping requests to unix sockets", uwsgi_opt_http_use_pattern, NULL, 0}, - {"http-use-base", required_argument, 0, "use the specified base for mapping requests to unix sockets", uwsgi_opt_http_use_base, NULL, 0}, - {"http-use-cluster", no_argument, 0, "load balance to nodes subscribed to the cluster", uwsgi_opt_true, &uhttp.use_cluster, 0}, - {"http-events", required_argument, 0, "set the number of concurrent http async events", uwsgi_opt_set_int, &uhttp.nevents, 0}, - {"http-subscription-server", required_argument, 0, "enable the subscription server", uwsgi_opt_http_ss, NULL, 0}, - {"http-subscription-use-regexp", no_argument, 0, "enable regexp usage in subscription system", uwsgi_opt_true, &uhttp.subscription_regexp, 0}, - {"http-timeout", required_argument, 0, "set internal http socket timeout", uwsgi_opt_set_int, &uhttp.socket_timeout, 0}, + {"http-use-cache", no_argument, 0, "use uWSGI cache as key->value virtualhost mapper", uwsgi_opt_true, &uhttp.cr.use_cache, 0}, + {"http-use-pattern", required_argument, 0, "use the specified pattern for mapping requests to unix sockets", uwsgi_opt_corerouter_use_pattern, NULL, 0}, + {"http-use-base", required_argument, 0, "use the specified base for mapping requests to unix sockets", uwsgi_opt_corerouter_use_base, NULL, 0}, + //{"http-use-cluster", no_argument, 0, "load balance to nodes subscribed to the cluster", uwsgi_opt_true, &uhttp.cr.use_cluster, 0}, + {"http-events", required_argument, 0, "set the number of concurrent http async events", uwsgi_opt_set_int, &uhttp.cr.nevents, 0}, + {"http-subscription-server", required_argument, 0, "enable the subscription server", uwsgi_opt_corerouter_ss, NULL, 0}, + {"http-subscription-use-regexp", no_argument, 0, "enable regexp usage in subscription system", uwsgi_opt_true, &uhttp.cr.subscription_regexp, 0}, + {"http-timeout", required_argument, 0, "set internal http socket timeout", uwsgi_opt_set_int, &uhttp.cr.socket_timeout, 0}, {"http-manage-expect", no_argument, 0, "manage the Expect HTTP request header", uwsgi_opt_true, &uhttp.manage_expect, 0}, {0, 0, 0, 0, 0, 0, 0}, }; -extern struct uwsgi_server uwsgi; - - -void http_manage_subscription(char *key, uint16_t keylen, char *val, uint16_t vallen, void *data) { - - struct uwsgi_subscribe_req *usr = (struct uwsgi_subscribe_req *) data; - - if (!uwsgi_strncmp("key", 3, key, keylen)) { - usr->key = val; - usr->keylen = vallen; - } - - else if (!uwsgi_strncmp("auth", 4, key, keylen)) { - usr->auth = val; - usr->auth_len = vallen; - } - - else if (!uwsgi_strncmp("address", 7, key, keylen)) { - usr->address = val; - usr->address_len = vallen; - } - -} - - struct http_session { - int fd; - int instance_fd; - int status; + struct corerouter_session crs; + struct uwsgi_header uh; - uint8_t h_pos; - uint16_t pos; - uint16_t parse_pos; - char *ptr; int rnrn; - - char *hostname; - uint16_t hostname_len; - - char *instance_address; - uint64_t instance_address_len; - - int instance_failed; - - int pass_fd; - - size_t remains; + char *ptr; char *port; int port_len; struct iovec iov[MAX_HTTP_VEC]; int iov_len; - char uss[MAX_HTTP_VEC * 2]; char buffer[UMAX16]; char path_info[UMAX16]; uint16_t path_info_len; - size_t content_length; size_t received_body; - struct uwsgi_subscribe_node *un; - in_addr_t ip_addr; char ip[INET_ADDRSTRLEN]; - struct uwsgi_rb_timer *timeout; }; -static struct uwsgi_rb_timer *reset_timeout(struct http_session *uhttp_session) { - - del_timeout(uhttp_session); - return add_timeout(uhttp_session); -} - -static void close_session(struct http_session **uhttp_table, struct http_session *uhttp_session) { - - close(uhttp_session->fd); - uhttp_table[uhttp_session->fd] = NULL; - if (uhttp_session->instance_fd != -1) { - if (uhttp.subscriptions && (uhttp_session->instance_failed || uhttp_session->status == HTTP_STATUS_CONNECTING)) { - uwsgi_log("marking %.*s as failed\n", (int) uhttp_session->instance_address_len, uhttp_session->instance_address); - uwsgi_remove_subscribe_node(&uhttp.subscriptions, uhttp_session->un); - } - close(uhttp_session->instance_fd); - uhttp_table[uhttp_session->instance_fd] = NULL; - } - - uhttp.load--; - del_timeout(uhttp_session); - free(uhttp_session); - -} - -static void expire_timeouts(struct http_session **uhttp_table) { - - time_t current = time(NULL); - struct uwsgi_rb_timer *urbt; - - for (;;) { - - urbt = uwsgi_min_rb_timer(uhttp.timeouts); - - if (urbt == NULL) - return; - - if (urbt->key <= current) { - close_session(uhttp_table, (struct http_session *) urbt->data); - uwsgi_log("timeout !!!\n"); - continue; - } - - break; - } -} - - -struct http_session *alloc_uhttp_session() { - - return uwsgi_malloc(sizeof(struct http_session)); -} uint16_t http_add_uwsgi_header(struct http_session *h_session, struct iovec *iov, char *strsize1, char *strsize2, char *hh, uint16_t hhlen, int *c) { @@ -281,12 +113,12 @@ uint16_t http_add_uwsgi_header(struct http_session *h_session, struct iovec *iov return 0; if (!uwsgi_strncmp("HOST", 4, hh, keylen)) { - h_session->hostname = val; - h_session->hostname_len = vallen; + h_session->crs.hostname = val; + h_session->crs.hostname_len = vallen; } if (!uwsgi_strncmp("CONTENT_LENGTH", 14, hh, keylen)) { - h_session->content_length = uwsgi_str_num(val, vallen); + h_session->crs.post_cl = uwsgi_str_num(val, vallen); } if (uwsgi_strncmp("CONTENT_TYPE", 12, hh, keylen) && uwsgi_strncmp("CONTENT_LENGTH", 14, hh, keylen)) { @@ -427,13 +259,11 @@ int http_parse(struct http_session *h_session) { // SERVER_NAME h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "SERVER_NAME", 11, uwsgi.hostname, uwsgi.hostname_len, &c); - h_session->hostname = uwsgi.hostname; - h_session->hostname_len = uwsgi.hostname_len; + h_session->crs.hostname = uwsgi.hostname; + h_session->crs.hostname_len = uwsgi.hostname_len; // SERVER_PORT h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "SERVER_PORT", 11, h_session->port, h_session->port_len, &c); - h_session->hostname = uwsgi.hostname; - h_session->hostname_len = uwsgi.hostname_len; // UWSGI_ROUTER h_session->uh.pktsize += http_add_uwsgi_var(h_session->iov, h_session->uss + c, h_session->uss + c + 2, "UWSGI_ROUTER", 12, "http", 4, &c); @@ -466,8 +296,8 @@ int http_parse(struct http_session *h_session) { } if (uhttp.manage_expect) { if (!uwsgi_strncmp("Expect: 100-continue", 20, base, ptr - base)) { - if (send(h_session->fd, protocol, protocol_len, 0) == (ssize_t) protocol_len) { - if (send(h_session->fd, " 100 Continue\r\n\r\n", 17, 0) != 17) { + if (send(h_session->crs.fd, protocol, protocol_len, 0) == (ssize_t) protocol_len) { + if (send(h_session->crs.fd, " 100 Continue\r\n\r\n", 17, 0) != 17) { uwsgi_error("send()"); } } @@ -492,335 +322,152 @@ int http_parse(struct http_session *h_session) { hv = hv->next; } -#ifdef UWSGI_DEBUG - uwsgi_log("vec size: %d pkt size: %d load %d\n", c, h_session->uh.pktsize, uhttp.load); -#endif - return c; } -void http_loop(int id) { +void uwsgi_http_switch_events(struct uwsgi_corerouter *ucr, struct corerouter_session *cs, int interesting_fd) { - int nevents; - int interesting_fd; - int new_connection; ssize_t len; - int i, j; - - char *magic_table[0xff]; - - char bbuf[UMAX16]; - - time_t delta; - - struct uwsgi_rb_timer *min_timeout; + int j; + struct http_session *hs = (struct http_session *) cs; #ifndef __sun__ - struct msghdr msg; - union { - struct cmsghdr cmsg; - char control[CMSG_SPACE(sizeof(int))]; - } msg_control; - struct cmsghdr *cmsg; + struct msghdr msg; + union { + struct cmsghdr cmsg; + char control[CMSG_SPACE(sizeof(int))]; + } msg_control; + struct cmsghdr *cmsg; #endif - - union uwsgi_sockaddr uhttp_addr; - socklen_t uhttp_addr_len = sizeof(struct sockaddr_un); - - struct http_session *uhttp_session; - - struct http_session **uhttp_table; - - int uhttp_queue; - - int soopt; + char bbuf[UMAX16]; socklen_t solen = sizeof(int); - uhttp_table = uwsgi_malloc(sizeof(struct http_session *) * uwsgi.max_fd); - for (i = 0; i < (int) uwsgi.max_fd; i++) { - uhttp_table[i] = NULL; - } + switch (cs->status) { - 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, ushared->gateways[id].internal_subscription_pipe[1]); - - if (uhttp.pattern) { - init_magic_table(magic_table); - } - - uhttp.timeouts = uwsgi_init_rb_timer(); - if (!uhttp.socket_timeout) - uhttp.socket_timeout = 30; - - for (;;) { - - min_timeout = uwsgi_min_rb_timer(uhttp.timeouts); - if (min_timeout == NULL) { - delta = -1; - } - else { - delta = min_timeout->key - time(NULL); - if (delta <= 0) { - expire_timeouts(uhttp_table); - delta = 0; - } - } - nevents = event_queue_wait_multi(uhttp_queue, delta, events, uhttp.nevents); - - if (nevents == 0) { - // manage timeout - expire_timeouts(uhttp_table); - } - - for (i = 0; i < nevents; i++) { - - interesting_fd = event_queue_interesting_fd(events, i); - - struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets; - int taken = 0; - while (ugs) { - - if (ugs->gateway == &ushared->gateways[id] && interesting_fd == ugs->fd) { - if (!ugs->subscription) { - new_connection = accept(ugs->fd, (struct sockaddr *) &uhttp_addr, &uhttp_addr_len); + case COREROUTER_STATUS_RECV_HDR: + len = recv(cs->fd, hs->buffer + cs->h_pos, UMAX16 - cs->h_pos, 0); #ifdef UWSGI_EVENT_USE_PORT - event_queue_add_fd_read(uhttp_queue, ugs->fd); -#endif - if (new_connection < 0) { - taken = 1; - break; - } -#ifndef __linux__ - uwsgi_socket_b(new_connection); -#endif - - uhttp_table[new_connection] = alloc_uhttp_session(); - uhttp_table[new_connection]->fd = new_connection; - uhttp_table[new_connection]->instance_fd = -1; - uhttp_table[new_connection]->status = HTTP_STATUS_RECV; - uhttp_table[new_connection]->h_pos = 0; - uhttp_table[new_connection]->pos = 0; - uhttp_table[new_connection]->rnrn = 0; - uhttp_table[new_connection]->parse_pos = 0; - uhttp_table[new_connection]->pass_fd = 0; - uhttp_table[new_connection]->ptr = uhttp_table[new_connection]->buffer; - uhttp_table[new_connection]->instance_address_len = 0; - uhttp_table[new_connection]->uh.modifier1 = uhttp.modifier1; - uhttp_table[new_connection]->uh.pktsize = 0; - uhttp_table[new_connection]->uh.modifier2 = 0; - uhttp_table[new_connection]->ip_addr = ((struct sockaddr_in *) &uhttp_addr)->sin_addr.s_addr; - uhttp_table[new_connection]->instance_failed = 0; - - uhttp_table[new_connection]->content_length = 0; - uhttp_table[new_connection]->received_body = 0; - - uhttp_table[new_connection]->port = ugs->port; - uhttp_table[new_connection]->port_len = ugs->port_len; - - uhttp_table[new_connection]->timeout = add_timeout(uhttp_table[new_connection]); - - uhttp.load++; - - event_queue_add_fd_read(uhttp_queue, new_connection); - } - else { - 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; - break; - - } - - ugs = ugs->next; - } - - if (taken) - continue; - - if (interesting_fd == ushared->gateways[id].internal_subscription_pipe[1]) { - 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; - - uhttp_session = uhttp_table[interesting_fd]; - - // something is going wrong... - if (uhttp_session == NULL) - continue; - - if (event_queue_interesting_fd_has_error(events, i)) { - close_session(uhttp_table, uhttp_session); - continue; - } - - uhttp_session->timeout = reset_timeout(uhttp_session); - - switch (uhttp_session->status) { - - - case HTTP_STATUS_RECV: - len = recv(uhttp_session->fd, uhttp_session->buffer + uhttp_session->h_pos, UMAX16 - uhttp_session->h_pos, 0); -#ifdef UWSGI_EVENT_USE_PORT - event_queue_add_fd_read(uhttp_queue, uhttp_session->fd); + event_queue_add_fd_read(uhttp_queue, cs->fd); #endif if (len <= 0) { if (len < 0) uwsgi_error("recv()"); - close_session(uhttp_table, uhttp_session); + corerouter_close_session(ucr, cs); break; } - uhttp_session->h_pos += len; + cs->h_pos += len; for (j = 0; j < len; j++) { - //uwsgi_log("%d %d %d\n", j, *uhttp_session->ptr, uhttp_session->rnrn); - if (*uhttp_session->ptr == '\r' && (uhttp_session->rnrn == 0 || uhttp_session->rnrn == 2)) { - uhttp_session->rnrn++; + //uwsgi_log("%d %d %d\n", j, *cs->ptr, cs->rnrn); + if (*hs->ptr == '\r' && (hs->rnrn == 0 || hs->rnrn == 2)) { + hs->rnrn++; } - else if (*uhttp_session->ptr == '\r') { - uhttp_session->rnrn = 1; + else if (*hs->ptr == '\r') { + hs->rnrn = 1; } - else if (*uhttp_session->ptr == '\n' && uhttp_session->rnrn == 1) { - uhttp_session->rnrn = 2; + else if (*hs->ptr == '\n' && hs->rnrn == 1) { + hs->rnrn = 2; } - else if (*uhttp_session->ptr == '\n' && uhttp_session->rnrn == 3) { - uhttp_session->ptr++; - uhttp_session->remains = len - (j + 1); - uhttp_session->iov_len = http_parse(uhttp_session); + else if (*hs->ptr == '\n' && hs->rnrn == 3) { + hs->ptr++; + cs->post_remains = len - (j + 1); + hs->iov_len = http_parse(hs); - if (uhttp_session->iov_len == 0) { - close_session(uhttp_table, uhttp_session); + if (hs->iov_len == 0) { + corerouter_close_session(ucr, cs); break; } - if (uhttp.use_cluster) { - uhttp_session->instance_address = uwsgi_cluster_best_node(); - if (uhttp_session->instance_address) { - uhttp_session->instance_address_len = strlen(uhttp_session->instance_address); - } - } - else if (uhttp.use_cache) { - uhttp_session->instance_address = uwsgi_cache_get(uhttp_session->hostname, uhttp_session->hostname_len, &uhttp_session->instance_address_len); - } - else if (uhttp.base) { - uhttp_session->instance_address = uwsgi_concat2n(uhttp.base, uhttp.base_len, uhttp_session->hostname, uhttp_session->hostname_len); - uhttp_session->instance_address_len = uhttp.base_len + uhttp_session->hostname_len; - } - else if (uhttp.pattern) { - magic_table['s'] = uwsgi_concat2n(uhttp_session->hostname, uhttp_session->hostname_len, "", 0); - int tmp_addr_len = 0; - uhttp_session->instance_address = magic_sub(uhttp.pattern, uhttp.pattern_len, &tmp_addr_len, magic_table); - uhttp_session->instance_address_len = tmp_addr_len; - free(magic_table['s']); - } - else if (uhttp.to) { - uhttp_session->instance_address = uhttp.to; - uhttp_session->instance_address_len = uhttp.to_len; - } - else if (uhttp.has_subscription_sockets) { - uhttp_session->un = uwsgi_get_subscribe_node(&uhttp.subscriptions, uhttp_session->hostname, uhttp_session->hostname_len, uhttp.subscription_regexp); - if (uhttp_session->un && uhttp_session->un->len) { - uhttp_session->instance_address = uhttp_session->un->name; - uhttp_session->instance_address_len = uhttp_session->un->len; - } - } - else if (uwsgi.sockets) { - uhttp_session->instance_address = uwsgi.sockets->name; - uhttp_session->instance_address_len = strlen(uwsgi.sockets->name); - } + // call the mapper - if (!uhttp_session->instance_address_len) { - close_session(uhttp_table, uhttp_session); + if (!cs->instance_address_len) { + corerouter_close_session(ucr, cs); break; } - uhttp_session->pass_fd = is_unix(uhttp_session->instance_address, uhttp_session->instance_address_len); + cs->pass_fd = is_unix(cs->instance_address, cs->instance_address_len); - uhttp_session->instance_fd = uwsgi_connectn(uhttp_session->instance_address, uhttp_session->instance_address_len, 0, 1); + cs->instance_fd = uwsgi_connectn(cs->instance_address, cs->instance_address_len, 0, 1); #ifdef UWSGI_DEBUG - uwsgi_log("uwsgi backend: %.*s\n", (int) uhttp_session->instance_address_len, uhttp_session->instance_address); + uwsgi_log("uwsgi backend: %.*s\n", (int) cs->instance_address_len, cs->instance_address); #endif - if (uhttp.pattern || uhttp.base) { - free(uhttp_session->instance_address); + if (ucr->pattern || ucr->base) { + free(cs->instance_address); } - if (uhttp_session->instance_fd < 0) { - uhttp_session->instance_failed = 1; - close_session(uhttp_table, uhttp_session); + if (cs->instance_fd < 0) { + cs->instance_failed = 1; + corerouter_close_session(ucr, cs); break; } - uhttp_session->status = HTTP_STATUS_CONNECTING; - uhttp_table[uhttp_session->instance_fd] = uhttp_session; - event_queue_add_fd_write(uhttp_queue, uhttp_session->instance_fd); + cs->status = COREROUTER_STATUS_CONNECTING; + ucr->cr_table[cs->instance_fd] = cs; + event_queue_add_fd_write(ucr->queue, cs->instance_fd); break; } else { - uhttp_session->rnrn = 0; + hs->rnrn = 0; } - uhttp_session->ptr++; + hs->ptr++; } break; - case HTTP_STATUS_CONNECTING: + case COREROUTER_STATUS_CONNECTING: - if (interesting_fd == uhttp_session->instance_fd) { + if (interesting_fd == cs->instance_fd) { - if (getsockopt(uhttp_session->instance_fd, SOL_SOCKET, SO_ERROR, (void *) (&soopt), &solen) < 0) { + if (getsockopt(cs->instance_fd, SOL_SOCKET, SO_ERROR, (void *) (&cs->soopt), &solen) < 0) { uwsgi_error("getsockopt()"); - uhttp_session->instance_failed = 1; - close_session(uhttp_table, uhttp_session); + cs->instance_failed = 1; + corerouter_close_session(ucr, cs); break; } - if (soopt) { - uwsgi_log("unable to connect() to uwsgi instance: %s\n", strerror(soopt)); - uhttp_session->instance_failed = 1; - close_session(uhttp_table, uhttp_session); + if (cs->soopt) { + uwsgi_log("unable to connect() to uwsgi instance: %s\n", strerror(cs->soopt)); + cs->instance_failed = 1; + corerouter_close_session(ucr, cs); break; } #ifdef __BIG_ENDIAN__ - uhttp_session->uh.pktsize = uwsgi_swap16(uhttp_session->uh.pktsize); + cs->uh.pktsize = uwsgi_swap16(cs->uh.pktsize); #endif - uhttp_session->iov[0].iov_base = &uhttp_session->uh; - uhttp_session->iov[0].iov_len = 4; + hs->iov[0].iov_base = &cs->uh; + hs->iov[0].iov_len = 4; - if (uhttp_session->remains > 0) { - uhttp_session->iov[uhttp_session->iov_len].iov_base = uhttp_session->ptr; - if (uhttp_session->remains > uhttp_session->content_length) { - uhttp_session->remains = uhttp_session->content_length; + if (cs->post_remains > 0) { + hs->iov[hs->iov_len].iov_base = hs->ptr; + if (cs->post_remains > cs->post_cl) { + cs->post_remains = cs->post_cl; } - uhttp_session->iov[uhttp_session->iov_len].iov_len = uhttp_session->remains; - uhttp_session->received_body += uhttp_session->remains; - uhttp_session->iov_len++; + hs->iov[hs->iov_len].iov_len = cs->post_remains; + hs->received_body += cs->post_remains; + hs->iov_len++; } #ifndef __sun__ // fd passing: PERFORMANCE EXTREME BOOST !!! - if (uhttp_session->pass_fd && !uhttp_session->remains && !uwsgi.no_fd_passing) { + if (cs->pass_fd && !cs->post_remains && !uwsgi.no_fd_passing) { msg.msg_name = NULL; msg.msg_namelen = 0; - msg.msg_iov = uhttp_session->iov; - msg.msg_iovlen = uhttp_session->iov_len; + msg.msg_iov = hs->iov; + msg.msg_iovlen = hs->iov_len; msg.msg_flags = 0; msg.msg_control = &msg_control; msg.msg_controllen = sizeof(msg_control); @@ -830,26 +477,25 @@ void http_loop(int id) { cmsg->cmsg_level = SOL_SOCKET; cmsg->cmsg_type = SCM_RIGHTS; - memcpy(CMSG_DATA(cmsg), &uhttp_session->fd, sizeof(int)); + memcpy(CMSG_DATA(cmsg), &cs->fd, sizeof(int)); - if (sendmsg(uhttp_session->instance_fd, &msg, 0) < 0) { + if (sendmsg(cs->instance_fd, &msg, 0) < 0) { uwsgi_error("sendmsg()"); } - close(uhttp_session->fd); - close(uhttp_session->instance_fd); - uhttp_table[uhttp_session->fd] = NULL; - uhttp_table[uhttp_session->instance_fd] = NULL; - uhttp.load--; - del_timeout(uhttp_session); - free(uhttp_session); + close(cs->fd); + close(cs->instance_fd); + ucr->cr_table[cs->fd] = NULL; + ucr->cr_table[cs->instance_fd] = NULL; + cr_del_timeout(ucr, cs); + free(hs); break; } #endif #ifdef __sun__ - if (uhttp_session->iov_len > IOV_MAX) { - int remains = uhttp_session->iov_len; + if (cs->iov_len > IOV_MAX) { + int remains = cs->iov_len; int iov_len; while (remains) { if (remains > IOV_MAX) { @@ -858,96 +504,94 @@ void http_loop(int id) { else { iov_len = remains; } - if (writev(uhttp_session->instance_fd, uhttp_session->iov + (uhttp_session->iov_len - remains), iov_len) <= 0) { + if (writev(cs->instance_fd, cs->iov + (cs->iov_len - remains), iov_len) <= 0) { uwsgi_error("writev()"); - close_session(uhttp_table, uhttp_session); + corerouter_close_session(ucr, cs); break; } remains -= iov_len; } } #else - if (writev(uhttp_session->instance_fd, uhttp_session->iov, uhttp_session->iov_len) <= 0) { + if (writev(cs->instance_fd, hs->iov, hs->iov_len) <= 0) { uwsgi_error("writev()"); - close_session(uhttp_table, uhttp_session); + corerouter_close_session(ucr, cs); break; } #endif - event_queue_fd_write_to_read(uhttp_queue, uhttp_session->instance_fd); - uhttp_session->status = HTTP_STATUS_RESPONSE; + event_queue_fd_write_to_read(ucr->queue, cs->instance_fd); + cs->status = COREROUTER_STATUS_RESPONSE; } break; - case HTTP_STATUS_RESPONSE: + case COREROUTER_STATUS_RESPONSE: // data from instance - if (interesting_fd == uhttp_session->instance_fd) { - len = recv(uhttp_session->instance_fd, bbuf, UMAX16, 0); + if (interesting_fd == cs->instance_fd) { + len = recv(cs->instance_fd, bbuf, UMAX16, 0); #ifdef UWSGI_EVENT_USE_PORT - event_queue_add_fd_read(uhttp_queue, uhttp_session->instance_fd); + event_queue_add_fd_read(uhttp_queue, cs->instance_fd); #endif if (len <= 0) { if (len < 0) uwsgi_error("recv()"); - close_session(uhttp_table, uhttp_session); + corerouter_close_session(ucr, cs); break; } - ssize_t s_len = send(uhttp_session->fd, bbuf, len, 0); + ssize_t s_len = send(cs->fd, bbuf, len, 0); if (s_len <= 0) { if (s_len < 0) uwsgi_error("send()"); - close(uhttp_session->fd); - close(uhttp_session->instance_fd); - uhttp_table[uhttp_session->fd] = NULL; - uhttp_table[uhttp_session->instance_fd] = NULL; - uhttp.load--; - del_timeout(uhttp_session); - free(uhttp_session); + close(cs->fd); + close(cs->instance_fd); + ucr->cr_table[cs->fd] = NULL; + ucr->cr_table[cs->instance_fd] = NULL; + cr_del_timeout(ucr, cs); + free(hs); break; } } // body from client - else if (interesting_fd == uhttp_session->fd) { + else if (interesting_fd == cs->fd) { - len = recv(uhttp_session->fd, bbuf, UMAX16, 0); + len = recv(cs->fd, bbuf, UMAX16, 0); #ifdef UWSGI_EVENT_USE_PORT - event_queue_add_fd_read(uhttp_queue, uhttp_session->fd); + event_queue_add_fd_read(uhttp_queue, cs->fd); #endif if (len <= 0) { if (len < 0) uwsgi_error("recv()"); - close(uhttp_session->fd); - close(uhttp_session->instance_fd); - uhttp_table[uhttp_session->fd] = NULL; - uhttp_table[uhttp_session->instance_fd] = NULL; - uhttp.load--; - del_timeout(uhttp_session); - free(uhttp_session); + close(cs->fd); + close(cs->instance_fd); + ucr->cr_table[cs->fd] = NULL; + ucr->cr_table[cs->instance_fd] = NULL; + cr_del_timeout(ucr, cs); + free(hs); break; } - if (uhttp_session->received_body >= uhttp_session->content_length) { + if (hs->received_body >= cs->post_cl) { break; } - if (len + uhttp_session->received_body > uhttp_session->content_length) { - len = uhttp_session->content_length - uhttp_session->received_body; + if (len + hs->received_body > cs->post_cl) { + len = cs->post_cl - hs->received_body; } - len = send(uhttp_session->instance_fd, bbuf, len, 0); + len = send(cs->instance_fd, bbuf, len, 0); if (len <= 0) { if (len < 0) uwsgi_error("send()"); - close_session(uhttp_table, uhttp_session); + corerouter_close_session(ucr, cs); break; } - uhttp_session->received_body += len; + hs->received_body += len; } @@ -957,54 +601,22 @@ void http_loop(int id) { // fallback to destroy !!! default: - uwsgi_log("default action\n"); - close(uhttp_session->fd); - uhttp_table[uhttp_session->fd] = NULL; - if (uhttp_session->instance_fd != -1) { - close(uhttp_session->instance_fd); - uhttp_table[uhttp_session->instance_fd] = NULL; - } - uhttp.load--; - del_timeout(uhttp_session); - free(uhttp_session); - break; - - } - - } + uwsgi_log("unknown event: closing session\n"); + corerouter_close_session(ucr, cs); + break; } + +} + +void http_setup() { + uhttp.cr.name = uwsgi_str("uWSGI http"); } int http_init() { - int i; - - if (uhttp.has_sockets) { - - if (uhttp.use_cache && !uwsgi.cache_max_items) { - uwsgi_log("you need to create a uwsgi cache to use the http (add --cache )\n"); - exit(1); - } - - if (!uhttp.nevents) - uhttp.nevents = 64; - - if (!uhttp.base && !uhttp.use_cache && !uhttp.to && !uwsgi.sockets && !uhttp.has_subscription_sockets && !uhttp.use_cluster) { - uwsgi_new_socket(uwsgi_concat2("127.0.0.1:0", "")); - } - - uwsgi_corerouter_setup_sockets("uWSGI http"); - - if (uhttp.processes < 1) - uhttp.processes = 1; - for (i = 0; i < uhttp.processes; i++) { - if (register_gateway("uWSGI http", http_loop) == NULL) { - uwsgi_log("unable to register the http gateway\n"); - exit(1); - } - } - - } + uhttp.cr.session_size = sizeof(struct http_session); + uhttp.cr.switch_events = uwsgi_http_switch_events; + uwsgi_corerouter_init((struct uwsgi_corerouter *) &uhttp); return 0; } @@ -1015,4 +627,5 @@ struct uwsgi_plugin http_plugin = { .name = "http", .options = http_options, .init = http_init, + .on_load = http_setup, }; diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index e2fd0ddc..c182e806 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -794,6 +794,11 @@ PyObject *py_uwsgi_advanced_sendfile(PyObject * self, PyObject * args) { wsgi_req->sendfile_fd_chunk = chunk; wsgi_req->sendfile_fd_pos = pos; + // send the headers if not already sent + if (!wsgi_req->headers_sent && wsgi_req->headers_hvec > 0) { + uwsgi_python_do_send_headers(wsgi_req); + } + // do sendfile if (uwsgi.async > 1) { ssize_t sf_len = uwsgi_sendfile(wsgi_req); diff --git a/plugins/zergpool/zergpool.c b/plugins/zergpool/zergpool.c index 57b78413..88739896 100644 --- a/plugins/zergpool/zergpool.c +++ b/plugins/zergpool/zergpool.c @@ -28,7 +28,7 @@ struct zergpool_socket { struct zergpool_socket *zergpool_sockets; -void zergpool_loop(int id) { +void zergpool_loop(int id, void *foobar) { int i; @@ -137,7 +137,7 @@ int zergpool_init() { zpsn = zpsn->next; } - if (register_gateway("uWSGI zergpool", zergpool_loop) == NULL) { + if (register_gateway("uWSGI zergpool", zergpool_loop, NULL) == NULL) { uwsgi_log("unable to register the zergpool gateway\n"); exit(1); } diff --git a/uwsgi.h b/uwsgi.h index 0832bacd..59e16624 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -378,13 +378,15 @@ struct uwsgi_gateway { char *name; char *fullname; - void (*loop) (int); + void (*loop) (int, void *); pid_t pid; int num; int use_signals; int internal_subscription_pipe[2]; uint64_t respawns; + + void *data; }; struct uwsgi_gateway_socket { @@ -2213,7 +2215,7 @@ char *uwsgi_cheap_string(char *, int); int uwsgi_parse_array(char *, uint16_t, char **, uint16_t *, uint8_t *); -struct uwsgi_gateway *register_gateway(char *, void (*)(int)); +struct uwsgi_gateway *register_gateway(char *, void (*)(int, void *), void *); void gateway_respawn(int); void uwsgi_gateway_go_cheap(char *, int, int *);