diff --git a/event.c b/event.c index fdcd9b76..35c7330e 100644 --- a/event.c +++ b/event.c @@ -149,6 +149,40 @@ int event_queue_add_fd_write(int eq, int fd) { return fd; } +void *event_queue_alloc(int nevents) { + + return uwsgi_malloc(sizeof(struct epoll_event) * nevents); +} + +int event_queue_interesting_fd(void *events, int id) { + struct epoll_event *ee = (struct epoll_event *) events; + return ee[id].data.fd; +} + +int event_queue_interesting_fd_has_error(void *events, int id) { + struct epoll_event *ee = (struct epoll_event *) events; + if (ee[id].events & EPOLLHUP || ee[id].events & EPOLLERR) { + return 1; + } + return 0; +} + +int event_queue_wait_multi(int eq, int timeout, void *events, int nevents) { + + int ret; + + if (timeout > 0) { + timeout = timeout*1000; + } + + ret = epoll_wait(eq, (struct epoll_event *) events, nevents, timeout); + if (ret < 0) { + uwsgi_error("epoll_wait()"); + } + + return ret; +} + int event_queue_wait(int eq, int timeout, int *interesting_fd) { int ret; diff --git a/plugins/fastrouter/fastrouter.c b/plugins/fastrouter/fastrouter.c index 4b5da9fa..265eadf4 100644 --- a/plugins/fastrouter/fastrouter.c +++ b/plugins/fastrouter/fastrouter.c @@ -12,7 +12,10 @@ #include "../../uwsgi.h" -#define LONG_ARGS_FASTROUTER 150001 +#define LONG_ARGS_FASTROUTER 150001 +#define LONG_ARGS_FASTROUTER_EVENTS 150002 +#define LONG_ARGS_FASTROUTER_USE_PATTERN 150003 +#define LONG_ARGS_FASTROUTER_USE_BASE 150004 #define FASTROUTER_STATUS_FREE 0 #define FASTROUTER_STATUS_CONNECTING 1 @@ -23,11 +26,21 @@ struct uwsgi_fastrouter { char *socket_name; int use_cache; + int nevents; + + char *pattern; + int pattern_len; + + char *base; + int base_len; } ufr; struct option fastrouter_options[] = { {"fastrouter", required_argument, 0, LONG_ARGS_FASTROUTER}, {"fastrouter-use-cache", no_argument, &ufr.use_cache, 1}, + {"fastrouter-use-pattern", required_argument, 0, LONG_ARGS_FASTROUTER_USE_PATTERN}, + {"fastrouter-use-base", required_argument, 0, LONG_ARGS_FASTROUTER_USE_BASE}, + {"fastrouter-events", required_argument, 0, LONG_ARGS_FASTROUTER_EVENTS}, {0, 0, 0, 0}, }; @@ -85,6 +98,12 @@ void fastrouter_loop() { ssize_t len; int i; + char *tmp_socket_name; + int tmp_socket_name_len; + + char *magic_table[0xff]; + + void *events; struct msghdr msg; union { struct cmsghdr cmsg; @@ -111,15 +130,24 @@ void fastrouter_loop() { fr_server = bind_to_tcp(ufr.socket_name, uwsgi.listen_queue, ufr.socket_name); fr_queue = event_queue_init(); + + events = event_queue_alloc(ufr.nevents); + event_queue_add_fd_read(fr_queue, fr_server); + if (ufr.pattern) { + init_magic_table(magic_table); + } + for (;;) { - nevents = event_queue_wait(fr_queue, -1, &interesting_fd); + nevents = event_queue_wait_multi(fr_queue, -1, events, ufr.nevents); - if (nevents > 0) { + for (i=0;ifd); + fr_table[fr_session->fd] = NULL; + if (fr_session->instance_fd != -1) { + close(fr_session->instance_fd); + fr_table[fr_session->instance_fd] = NULL; + } + free(fr_session); + continue; + } + switch(fr_session->status) { case FASTROUTER_STATUS_RECV_HDR: @@ -194,6 +233,18 @@ void fastrouter_loop() { if (ufr.use_cache) { fr_session->instance_address = uwsgi_cache_get(fr_session->hostname, fr_session->hostname_len, &fr_session->instance_address_len); } + else if (ufr.pattern) { + magic_table['s'] = uwsgi_concat2n(fr_session->hostname, fr_session->hostname_len, "", 0); + tmp_socket_name = magic_sub(ufr.pattern, ufr.pattern_len, &tmp_socket_name_len, magic_table); + free(magic_table['s']); + fr_session->instance_address_len = tmp_socket_name_len; + fr_session->instance_address = tmp_socket_name; + } + else if (ufr.base) { + tmp_socket_name = uwsgi_concat2nn(ufr.base, ufr.base_len, fr_session->hostname, fr_session->hostname_len, &tmp_socket_name_len); + fr_session->instance_address_len = tmp_socket_name_len; + fr_session->instance_address = tmp_socket_name; + } // no address found if (!fr_session->instance_address_len) { @@ -207,6 +258,9 @@ void fastrouter_loop() { 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 (tmp_socket_name) free(tmp_socket_name); + if (fr_session->instance_fd < 0) { close(fr_session->fd); fr_table[fr_session->fd] = NULL; @@ -386,6 +440,8 @@ int fastrouter_init() { exit(1); } + if (!ufr.nevents) ufr.nevents = 64; + if (register_gateway("fastrouter", fastrouter_loop) == NULL) { uwsgi_log("unable to register the fastrouter gateway\n"); exit(1); @@ -401,6 +457,19 @@ int fastrouter_opt(int i, char *optarg) { case LONG_ARGS_FASTROUTER: ufr.socket_name = optarg; return 1; + case LONG_ARGS_FASTROUTER_EVENTS: + ufr.nevents = atoi(optarg); + return 1; + case LONG_ARGS_FASTROUTER_USE_PATTERN: + ufr.pattern = optarg; + // optimization + ufr.pattern_len = strlen(ufr.pattern); + return 1; + case LONG_ARGS_FASTROUTER_USE_BASE: + ufr.base = optarg; + // optimization + ufr.base_len = strlen(ufr.base); + return 1; } return 0; } diff --git a/socket.c b/socket.c index 81be899c..4b1f13fb 100644 --- a/socket.c +++ b/socket.c @@ -454,6 +454,11 @@ int timed_connect(struct pollfd *fdpoll, const struct sockaddr *addr, int addr_s ret = connect(fdpoll->fd, addr, addr_size); + if (async) { + if (ret < 0 && errno != EINPROGRESS) { + return -1; + } + } /* re-set blocking socket */ arg &= (~O_NONBLOCK); diff --git a/utils.c b/utils.c index ca71111c..fb2d0c53 100644 --- a/utils.c +++ b/utils.c @@ -846,6 +846,23 @@ char *uwsgi_concat2n(char *one, int s1, char *two, int s2) { } +char *uwsgi_concat2nn(char *one, int s1, char *two, int s2, int *len) { + + char *buf; + *len = s1 + s2 + 1; + + + buf = uwsgi_malloc(*len); + buf[*len-1] = 0; + + memcpy( buf, one, s1); + memcpy( buf + s1, two, s2); + + return buf; + +} + + char *uwsgi_concat3n(char *one, int s1, char *two, int s2, char *three, int s3) { char *buf; diff --git a/uwsgi.h b/uwsgi.h index dec97d24..3d7c70dc 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -1314,6 +1314,7 @@ char *uwsgi_concat(int,...); char *uwsgi_concatn(int,...); char *uwsgi_concat2(char *, char *); char *uwsgi_concat2n(char *, int, char *, int); +char *uwsgi_concat2nn(char *, int, char *, int, int*); char *uwsgi_concat3(char *, char *, char *); char *uwsgi_concat3n(char *, int, char *, int, char *, int); char *uwsgi_concat4(char *, char *, char *, char *); @@ -1391,10 +1392,14 @@ inline void *uwsgi_malloc(size_t); int event_queue_init(void); +void *event_queue_alloc(int); int event_queue_add_fd_read(int, int); int event_queue_add_fd_write(int, int); int event_queue_del_fd(int, int); int event_queue_wait(int, int, int *); +int event_queue_wait_multi(int, int, void *, int); +int event_queue_interesting_fd(void *, int); +int event_queue_interesting_fd_has_error(void *, int); int event_queue_add_timer(int, int *, int); struct uwsgi_timer *event_queue_ack_timer(int);