From 34dbfc597e050367abb7e93e54bab5a6606232fa Mon Sep 17 00:00:00 2001 From: "roberto@sirius" Date: Sun, 21 Mar 2010 21:01:27 +0100 Subject: [PATCH] proxy refactoring --- async.c | 32 ++++++ logging.c | 4 +- proxy.c | 262 ++++++++++++++++------------------------------- uwsgi.h | 11 ++ uwsgi_pymodule.c | 1 - 5 files changed, 131 insertions(+), 179 deletions(-) diff --git a/async.c b/async.c index 19264bea..97e19178 100644 --- a/async.c +++ b/async.c @@ -65,6 +65,21 @@ int async_add(int queuefd, int fd, int etype) { return 0; } +int async_mod(int queuefd, int fd, int etype) { + struct epoll_event ee; + + memset(&ee, 0, sizeof(struct epoll_event)); + ee.events = etype; + ee.data.fd = fd; + + if (epoll_ctl(queuefd, EPOLL_CTL_MOD, fd, &ee)) { + perror("epoll_ctl()"); + return -1; + } + + return 0; +} + int async_del(int queuefd, int fd, int etype) { struct epoll_event ee; @@ -143,6 +158,23 @@ int async_add(int queuefd, int fd, int etype) { return 0; } +int async_mod(int queuefd, int fd, int etype) { + struct kevent kev; + + EV_SET(&kev, fd, ASYNC_OUT, EV_DISABLE, 0, 0, NULL); + if (kevent(queuefd, &kev, 1, NULL, 0, NULL) < 0) { + perror("kevent()"); + return -1; + } + + EV_SET(&kev, fd, etype, EV_ADD, 0, 0, NULL); + if (kevent(queuefd, &kev, 1, NULL, 0, NULL) < 0) { + perror("kevent()"); + return -1; + } + return 0; +} + int async_del(int queuefd, int fd, int etype) { struct kevent kev; diff --git a/logging.c b/logging.c index 91a538ad..c8f8729b 100644 --- a/logging.c +++ b/logging.c @@ -55,7 +55,7 @@ void log_request(struct wsgi_request *wsgi_req) { #endif } - fprintf(stderr, "%s[pid: %d|app: %d|req: %d/%llu] %.*s (%.*s) {%d vars in %d bytes} [%.*s] %.*s %.*s => generated %lu bytes in %ld msecs%s(%.*s %d) %d headers in %d bytes (%d async switches)\n", + fprintf(stderr, "%s[pid: %d|app: %d|req: %d/%llu] %.*s (%.*s) {%d vars in %d bytes} [%.*s] %.*s %.*s => generated %llu bytes in %ld msecs%s(%.*s %d) %d headers in %d bytes (%d async switches)\n", first_part, uwsgi.mypid, wsgi_req->app_id, @@ -68,7 +68,7 @@ void log_request(struct wsgi_request *wsgi_req) { 24, time_request, wsgi_req->method_len, wsgi_req->method, wsgi_req->uri_len, wsgi_req->uri, - wsgi_req->response_size, + (uint64_t) wsgi_req->response_size, (long int) (microseconds - microseconds2) / 1000, via, wsgi_req->protocol_len, wsgi_req->protocol, diff --git a/proxy.c b/proxy.c index 10db387e..e9811ccc 100644 --- a/proxy.c +++ b/proxy.c @@ -14,41 +14,6 @@ #include "uwsgi.h" -#ifdef __linux__ -#include -#define UWSGI_PROXY_USE_EPOLL 1 -#define EV_FD eevents[i].data.fd -#define EV_EV eevents[i].events -#define EV_IN EPOLLIN -#define EV_OUT EPOLLOUT -#define NEV_FD ee.data.fd -#define NEV_EV ee.events -#define NEV_ADD epoll_ctl(epfd, EPOLL_CTL_ADD, NEV_FD, &ee) -#define NEV_MOD epoll_ctl(epfd, EPOLL_CTL_MOD, NEV_FD, &ee) -#define EV_NAME "epoll_ctl()" -#define EV_IS_IN EV_EV & EV_IN -#define EV_IS_OUT EV_EV & EV_OUT -#elif defined(__sun__) - -#else -#include -#define UWSGI_PROXY_USE_KQUEUE 1 -#define EV_FD krevents[i].ident -#define EV_EV krevents[i].filter -#define EV_IN EVFILT_READ -#define EV_OUT EVFILT_WRITE -#define NEV_FD kev.ident -#define NEV_EV kev.filter -#define NEV_ADD kevent(kq, &kev, 1, NULL, 0, NULL) < 0 -#define NEV_MOD kevent(kq, &kev, 1, NULL, 0, NULL) < 0 -#define EV_NAME "kevent()" -#define EV_IS_IN EV_EV == EV_IN -#define EV_IS_OUT EV_EV == EV_OUT -#endif - -#include - - #define UWSGI_PROXY_CONNECTING 1 #define UWSGI_PROXY_WAITING 2 @@ -139,14 +104,14 @@ static int uwsgi_proxy_find_next_node(int current_node) { void uwsgi_proxy(int proxyfd) { + int efd ; + #ifdef __linux__ - int epfd; - struct epoll_event ee; struct epoll_event *eevents; + struct epoll_event ev; #else - int kq; - struct kevent *krevents; - struct kevent kev; + struct kevent *eevents; + struct kevent ev; #endif int max_events = 64; @@ -183,53 +148,25 @@ void uwsgi_proxy(int proxyfd) { } memset(upcs, 0, sizeof(struct uwsgi_proxy_connection) * max_connections); -#ifdef __linux__ - //init epoll - epfd = epoll_create(256); - if (epfd < 0) { - perror("epoll_create()"); + efd = async_queue_init(proxyfd); + if (efd < 0) { exit(1); } - // allocate memory for events +#ifdef __linux__ eevents = malloc(sizeof(struct epoll_event) * max_events); + memset(&ev, 0, sizeof(struct epoll_event)); +#else + eevents = malloc(sizeof(struct kevent) * max_events); + memset(&ev, 0, sizeof(struct kevent)); +#endif + if (!eevents) { perror("malloc()"); exit(1); } - // now add the proxyfd to the epoll list - - ee.events = EPOLLIN; - ee.data.fd = proxyfd; - - if (epoll_ctl(epfd, EPOLL_CTL_ADD, proxyfd, &ee)) { - perror("epoll_ctl()"); - exit(1); - } -#else - kq = kqueue(); - if (kq < 0) { - perror("kqueue()"); - exit(1); - } - - // allocate memory for events - krevents = malloc(sizeof(struct kevent) * max_events); - if (!krevents) { - perror("malloc()"); - exit(1); - } - - EV_SET(&kev, proxyfd, EVFILT_READ, EV_ADD, 0, 0, NULL); - if (kevent(kq, &kev, 1, NULL, 0, NULL) < 0) { - perror("kevent()"); - exit(1); - } - -#endif - signal(SIGINT, (void *) &end_proxy); signal(SIGTERM, (void *) &reload_proxy); signal(SIGHUP, (void *) &reload_proxy); @@ -237,107 +174,92 @@ void uwsgi_proxy(int proxyfd) { for (;;) { -#ifdef __linux__ - nevents = epoll_wait(epfd, eevents, max_events, -1); + nevents = async_wait(efd, eevents, max_events, -1, 0); if (nevents < 0) { perror("epoll_wait()"); continue; } -#else - nevents = kevent(kq, NULL, 0, krevents, max_events, NULL); - if (nevents < 0) { - perror("kevent()"); - continue; - } -#endif - for (i = 0; i < nevents; i++) { - if (EV_FD == proxyfd) { + if (eevents[i].ASYNC_FD == proxyfd) { - if (EV_IS_IN) { + if (eevents[i].ASYNC_IS_IN) { // new connection, accept it - NEV_FD = accept(proxyfd, (struct sockaddr *) &upc_addr, &upc_len); - if (NEV_FD < 0) { + ev.ASYNC_FD = accept(proxyfd, (struct sockaddr *) &upc_addr, &upc_len); + if (ev.ASYNC_FD < 0) { perror("accept()"); continue; } - upcs[NEV_FD].node = -1; + upcs[ev.ASYNC_FD].node = -1; // now connect to the first worker available - upcs[NEV_FD].dest_fd = socket(AF_INET, SOCK_STREAM, 0); - if (upcs[NEV_FD].dest_fd < 0) { + upcs[ev.ASYNC_FD].dest_fd = socket(AF_INET, SOCK_STREAM, 0); + if (upcs[ev.ASYNC_FD].dest_fd < 0) { perror("socket()"); - uwsgi_proxy_close(upcs, NEV_FD); + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } - upcs[upcs[NEV_FD].dest_fd].node = -1; + upcs[upcs[ev.ASYNC_FD].dest_fd].node = -1; // set nonblocking - if (ioctl(upcs[NEV_FD].dest_fd, FIONBIO, &nonblocking)) { + if (ioctl(upcs[ev.ASYNC_FD].dest_fd, FIONBIO, &nonblocking)) { perror("ioctl()"); - uwsgi_proxy_close(upcs, NEV_FD); + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } - upcs[NEV_FD].status = 0; - upcs[NEV_FD].retry = 0; + upcs[ev.ASYNC_FD].status = 0; + upcs[ev.ASYNC_FD].retry = 0; next_node = uwsgi_proxy_find_next_node(next_node); if (next_node == -1) { fprintf(stderr, "unable to find an available worker in the cluster !\n"); - uwsgi_proxy_close(upcs, NEV_FD); + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } - upcs[upcs[NEV_FD].dest_fd].node = next_node; - rc = connect(upcs[NEV_FD].dest_fd, (struct sockaddr *) &uwsgi.shared->nodes[next_node].ucn_addr, sizeof(struct sockaddr_in)); + upcs[upcs[ev.ASYNC_FD].dest_fd].node = next_node; + rc = connect(upcs[ev.ASYNC_FD].dest_fd, (struct sockaddr *) &uwsgi.shared->nodes[next_node].ucn_addr, sizeof(struct sockaddr_in)); uwsgi.shared->nodes[next_node].connections++; if (!rc) { // connected to worker, put it in the epoll_list - NEV_EV = EV_IN; - if (NEV_ADD) { - perror(EV_NAME); - uwsgi_proxy_close(upcs, NEV_FD); + if (async_add(efd, ev.ASYNC_FD, ASYNC_IN)) { + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } - upcs[upcs[NEV_FD].dest_fd].dest_fd = NEV_FD; - upcs[upcs[NEV_FD].dest_fd].status = 0; - upcs[upcs[NEV_FD].dest_fd].retry = 0; + upcs[upcs[ev.ASYNC_FD].dest_fd].dest_fd = ev.ASYNC_FD; + upcs[upcs[ev.ASYNC_FD].dest_fd].status = 0; + upcs[upcs[ev.ASYNC_FD].dest_fd].retry = 0; - NEV_FD = upcs[NEV_FD].dest_fd; + ev.ASYNC_FD = upcs[ev.ASYNC_FD].dest_fd; - NEV_EV = EV_IN; - if (NEV_ADD) { - perror(EV_NAME); - uwsgi_proxy_close(upcs, NEV_FD); + if (async_add(efd, ev.ASYNC_FD, ASYNC_IN)) { + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } // re-set blocking - if (ioctl(upcs[upcs[NEV_FD].dest_fd].dest_fd, FIONBIO, &blocking)) { + if (ioctl(upcs[upcs[ev.ASYNC_FD].dest_fd].dest_fd, FIONBIO, &blocking)) { perror("ioctl()"); - uwsgi_proxy_close(upcs, NEV_FD); + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } } else if (errno == EINPROGRESS) { // the socket is waiting, set status to CONNECTING - upcs[NEV_FD].status = UWSGI_PROXY_WAITING; - upcs[upcs[NEV_FD].dest_fd].dest_fd = NEV_FD; - upcs[upcs[NEV_FD].dest_fd].status = UWSGI_PROXY_CONNECTING; - upcs[upcs[NEV_FD].dest_fd].retry = 0; + upcs[ev.ASYNC_FD].status = UWSGI_PROXY_WAITING; + upcs[upcs[ev.ASYNC_FD].dest_fd].dest_fd = ev.ASYNC_FD; + upcs[upcs[ev.ASYNC_FD].dest_fd].status = UWSGI_PROXY_CONNECTING; + upcs[upcs[ev.ASYNC_FD].dest_fd].retry = 0; - NEV_FD = upcs[NEV_FD].dest_fd; - NEV_EV = EV_OUT; - if (NEV_ADD) { - perror(EV_NAME); - uwsgi_proxy_close(upcs, NEV_FD); + ev.ASYNC_FD = upcs[ev.ASYNC_FD].dest_fd; + if (async_add(efd, ev.ASYNC_FD, ASYNC_OUT)) { + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } } @@ -345,7 +267,7 @@ void uwsgi_proxy(int proxyfd) { // connection failed, retry with the next node ? perror("connect()"); // close only when all node are tried - uwsgi_proxy_close(upcs, NEV_FD); + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } @@ -358,110 +280,98 @@ void uwsgi_proxy(int proxyfd) { } else { // this is for clients/workers - if (EV_IS_IN) { + if (eevents[i].ASYNC_IS_IN) { // is this a connected client/worker ? //fprintf(stderr,"ready %d\n", upcs[eevents[i].data.fd].status); - if (!upcs[EV_FD].status) { - if (upcs[EV_FD].dest_fd >= 0) { + if (!upcs[eevents[i].ASYNC_FD].status) { + if (upcs[eevents[i].ASYNC_FD].dest_fd >= 0) { - rlen = read(EV_FD, buffer, 4096); + rlen = read(eevents[i].ASYNC_FD, buffer, 4096); if (rlen < 0) { perror("read()"); - uwsgi_proxy_close(upcs, EV_FD); + uwsgi_proxy_close(upcs, eevents[i].ASYNC_FD); continue; } else if (rlen == 0) { - uwsgi_proxy_close(upcs, EV_FD); + uwsgi_proxy_close(upcs, eevents[i].ASYNC_FD); continue; } else { - wlen = write(upcs[EV_FD].dest_fd, buffer, rlen); + wlen = write(upcs[eevents[i].ASYNC_FD].dest_fd, buffer, rlen); if (wlen != rlen) { perror("write()"); - uwsgi_proxy_close(upcs, EV_FD); + uwsgi_proxy_close(upcs, eevents[i].ASYNC_FD); continue; } } } else { - uwsgi_proxy_close(upcs, EV_FD); + uwsgi_proxy_close(upcs, eevents[i].ASYNC_FD); continue; } } - else if (upcs[EV_FD].status == UWSGI_PROXY_WAITING) { + else if (upcs[eevents[i].ASYNC_FD].status == UWSGI_PROXY_WAITING) { // disconnected node continue; } else { - fprintf(stderr, "UNKNOWN STATUS %d\n", upcs[EV_FD].status); + fprintf(stderr, "UNKNOWN STATUS %d\n", upcs[eevents[i].ASYNC_FD].status); continue; } } - else if (EV_IS_OUT) { - if (upcs[EV_FD].status == UWSGI_PROXY_CONNECTING) { + else if (eevents[i].ASYNC_IS_OUT) { + if (upcs[eevents[i].ASYNC_FD].status == UWSGI_PROXY_CONNECTING) { #ifdef UWSGI_PROXY_USE_KQUEUE - if (getsockopt(EV_FD, SOL_SOCKET, SO_ERROR, (void *) (&soopt), &solen) < 0) { + if (getsockopt(eevents[i].ASYNC_FD, SOL_SOCKET, SO_ERROR, (void *) (&soopt), &solen) < 0) { perror("getsockopt()"); - uwsgi_proxy_close(upcs, NEV_FD); + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } /* is something bad ? */ if (soopt) { fprintf(stderr, "connect() %s\n", strerror(soopt)); // increase errors on node - fprintf(stderr, "*** marking cluster node %d/%s as failed ***\n", upcs[EV_FD].node, uwsgi.shared->nodes[upcs[EV_FD].node].name); - uwsgi.shared->nodes[upcs[EV_FD].node].errors++; - uwsgi.shared->nodes[upcs[EV_FD].node].status = UWSGI_NODE_FAILED; - uwsgi_proxy_close(upcs, NEV_FD); + fprintf(stderr, "*** marking cluster node %d/%s as failed ***\n", upcs[eevents[i].ASYNC_FD].node, uwsgi.shared->nodes[upcs[eevents[i].ASYNC_FD].node].name); + uwsgi.shared->nodes[upcs[eevents[i].ASYNC_FD].node].errors++; + uwsgi.shared->nodes[upcs[eevents[i].ASYNC_FD].node].status = UWSGI_NODE_FAILED; + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } // increase errors on node #endif - NEV_FD = upcs[EV_FD].dest_fd; - NEV_EV = EV_IN; - upcs[NEV_FD].status = 0; - if (NEV_ADD) { - perror(EV_NAME); - uwsgi_proxy_close(upcs, NEV_FD); + ev.ASYNC_FD = upcs[eevents[i].ASYNC_FD].dest_fd; + upcs[ev.ASYNC_FD].status = 0; + if (async_add(efd, ev.ASYNC_FD, ASYNC_IN)) { + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } - NEV_FD = upcs[NEV_FD].dest_fd; - upcs[NEV_FD].status = 0; + ev.ASYNC_FD = upcs[ev.ASYNC_FD].dest_fd; + upcs[ev.ASYNC_FD].status = 0; -#ifdef UWSGI_PROXY_USE_KQUEUE - EV_SET(&kev, NEV_FD, EVFILT_WRITE, EV_ADD | EV_DISABLE, 0, 0, NULL); - if (NEV_MOD) { - perror(EV_NAME); - uwsgi_proxy_close(upcs, NEV_FD); - continue; - } - EV_SET(&kev, NEV_FD, EVFILT_READ, EV_ADD, 0, 0, NULL); -#endif - if (NEV_MOD) { - perror(EV_NAME); - uwsgi_proxy_close(upcs, NEV_FD); + if (async_mod(efd, ev.ASYNC_FD, ASYNC_IN)) { + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } // re-set blocking - if (ioctl(NEV_FD, FIONBIO, &blocking)) { + if (ioctl(ev.ASYNC_FD, FIONBIO, &blocking)) { perror("ioctl()"); - uwsgi_proxy_close(upcs, NEV_FD); + uwsgi_proxy_close(upcs, ev.ASYNC_FD); continue; } } else { - fprintf(stderr, "strange event for %d\n", (int) EV_FD); + fprintf(stderr, "strange event for %d\n", (int) eevents[i].ASYNC_FD); } } else { - if (upcs[EV_FD].status == UWSGI_PROXY_CONNECTING) { - if (getsockopt(EV_FD, SOL_SOCKET, SO_ERROR, (void *) (&soopt), &solen) < 0) { + if (upcs[eevents[i].ASYNC_FD].status == UWSGI_PROXY_CONNECTING) { + if (getsockopt(eevents[i].ASYNC_FD, SOL_SOCKET, SO_ERROR, (void *) (&soopt), &solen) < 0) { perror("getsockopt()"); } /* is something bad ? */ @@ -470,14 +380,14 @@ void uwsgi_proxy(int proxyfd) { } // increase errors on node - fprintf(stderr, "*** marking cluster node %d/%s as failed ***\n", upcs[EV_FD].node, uwsgi.shared->nodes[upcs[EV_FD].node].name); - uwsgi.shared->nodes[upcs[EV_FD].node].errors++; - uwsgi.shared->nodes[upcs[EV_FD].node].status = UWSGI_NODE_FAILED; + fprintf(stderr, "*** marking cluster node %d/%s as failed ***\n", upcs[eevents[i].ASYNC_FD].node, uwsgi.shared->nodes[upcs[eevents[i].ASYNC_FD].node].name); + uwsgi.shared->nodes[upcs[eevents[i].ASYNC_FD].node].errors++; + uwsgi.shared->nodes[upcs[eevents[i].ASYNC_FD].node].status = UWSGI_NODE_FAILED; } else { - fprintf(stderr, "STRANGE EVENT !!! %d %d %d\n", (int) EV_FD, EV_EV, upcs[EV_FD].status); + fprintf(stderr, "STRANGE EVENT !!! %d %d %d\n", (int) eevents[i].ASYNC_FD, (int) eevents[i].ASYNC_EV, upcs[eevents[i].ASYNC_FD].status); } - uwsgi_proxy_close(upcs, EV_FD); + uwsgi_proxy_close(upcs, eevents[i].ASYNC_FD); continue; } } diff --git a/uwsgi.h b/uwsgi.h index 5687eb88..6af67e59 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -60,6 +60,8 @@ #define MAX_PYARGV 10 +#include + #ifdef __linux__ #include #include @@ -651,6 +653,7 @@ struct wsgi_request *find_first_available_wsgi_req(struct uwsgi_server *); struct wsgi_request *find_wsgi_req_by_fd(struct uwsgi_server *, int, int); int async_add(int, int , int) ; +int async_mod(int, int , int) ; int async_wait(int, void *, int, int, int); int async_del(int, int , int) ; int async_queue_init(int); @@ -662,10 +665,18 @@ void async_expire_timeouts(struct uwsgi_server *); #ifdef __linux__ #define ASYNC_FD data.fd #define ASYNC_EV events +#define ASYNC_IN EPOLLIN +#define ASYNC_OUT EPOLLOUT +#define ASYNC_IS_IN ASYNC_EV & ASYNC_IN +#define ASYNC_IS_OUT ASYNC_EV & ASYNC_OUT #elif defined(__sun__) #else #define ASYNC_FD ident #define ASYNC_EV filter +#define ASYNC_IN EVFILT_READ +#define ASYNC_OUT EVFILT_WRITE +#define ASYNC_IS_IN ASYNC_EV == ASYNC_IN +#define ASYNC_IS_OUT ASYNC_EV == ASYNC_OUT #endif #endif diff --git a/uwsgi_pymodule.c b/uwsgi_pymodule.c index f5e7e1d8..d5da4ced 100644 --- a/uwsgi_pymodule.c +++ b/uwsgi_pymodule.c @@ -31,7 +31,6 @@ PyObject *py_uwsgi_async_sleep(PyObject * self, PyObject * args) { sec_timeout = (time_t) timeout ; - fprintf(stderr,"timeout sleep: %d\n", (int) sec_timeout); if (sec_timeout > 0) { async_set_timeout(uwsgi.wsgi_req, sec_timeout); }