From: Pierre Habouzit Date: Sun, 2 Dec 2007 12:01:20 +0000 (+0100) Subject: Refactor epoll module. X-Git-Url: http://git.madism.org/?p=apps%2Fpfixtools.git;a=commitdiff_plain;h=8a872b1a9ec12dd44292603137c41f1b29b9f756 Refactor epoll module. Signed-off-by: Pierre Habouzit --- diff --git a/buffer.c b/buffer.c index 4e41b3c..847f202 100644 --- a/buffer.c +++ b/buffer.c @@ -37,20 +37,40 @@ #include #include +#include "mem.h" #include "buffer.h" #define BUFSIZ_INCREMENT 256 void buffer_resize(buffer_t *buf, ssize_t newsize) { - if (newsize >= buf->size) { - /* rounds newsize to the 1024 multiple just after newsize+1 */ - newsize = (newsize + BUFSIZ_INCREMENT) & ~(BUFSIZ_INCREMENT - 1); - p_realloc(&buf->data, newsize); + p_allocgrow(&buf->data, newsize + 1, &buf->size); +} + +ssize_t buffer_addvf(buffer_t *buf, const char *fmt, va_list ap) +{ + ssize_t pos; + int len, available; + va_list ap2; + + va_copy(ap2, ap); + + pos = buf->len; + available = buf->size - pos; + + len = vsnprintf(buf->data + pos, available, fmt, ap); + if (len >= available) { + buffer_resize(buf, buf->len + len); + len = vsnprintf(buf->data + pos, available, fmt, ap2); } + buf->len = pos + len; + buf->data[buf->len] = '\0'; + va_end(ap2); + return len; } -void buffer_consume(buffer_t *buf, ssize_t len) { +void buffer_consume(buffer_t *buf, ssize_t len) +{ if (len <= 0) return; diff --git a/buffer.h b/buffer.h index 816788a..24bc590 100644 --- a/buffer.h +++ b/buffer.h @@ -36,6 +36,7 @@ #ifndef PFIXTOOLS_BUFFER_H #define PFIXTOOLS_BUFFER_H +#include #include "mem.h" #include "str.h" @@ -102,6 +103,20 @@ static inline void buffer_addch(buffer_t *buf, int c) { buffer_extendch(buf, 1, c); } +__attribute__((format(printf,2,0))) +ssize_t buffer_addvf(buffer_t *, const char *fmt, va_list); + +static inline __attribute__((format(printf,2,3))) +ssize_t buffer_addf(buffer_t *buf, const char *fmt, ...) +{ + ssize_t res; + va_list args; + va_start(args, fmt); + res = buffer_addvf(buf, fmt, args); + va_end(args); + return res; +} + void buffer_consume(buffer_t *buf, ssize_t len); ssize_t buffer_read(buffer_t *buf, int fd, ssize_t count); diff --git a/epoll.c b/epoll.c index 3d16789..e3df03e 100644 --- a/epoll.c +++ b/epoll.c @@ -35,7 +35,7 @@ #include "epoll.h" -int epollfd = -1; +static int epollfd = -1; static int epoll_initialize(void) { @@ -50,3 +50,26 @@ static void epoll_shutdown(void) module_init(epoll_initialize); module_exit(epoll_shutdown); + +void epoll_register(int fd, uint32_t events, void *ptr) +{ + struct epoll_event evt = { .events = events, .data.ptr = ptr }; + if (epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &evt)) { + UNIXERR("epoll_ctl"); + abort(); + } +} + +void epoll_modify(int fd, uint32_t events, void *ptr) +{ + struct epoll_event evt = { .events = events, .data.ptr = ptr }; + if (epoll_ctl(epollfd, EPOLL_CTL_MOD, fd, &evt)) { + UNIXERR("epoll_ctl"); + abort(); + } +} + +int epoll_select(struct epoll_event *events, int maxevents, int timeout) +{ + return epoll_wait(epollfd, events, maxevents, timeout); +} diff --git a/epoll.h b/epoll.h index bd52773..57179d2 100644 --- a/epoll.h +++ b/epoll.h @@ -39,6 +39,8 @@ #include #include "common.h" -extern int epollfd; +void epoll_register(int fd, uint32_t events, void *ptr); +void epoll_modify(int fd, uint32_t events, void *ptr); +int epoll_select(struct epoll_event *events, int maxevents, int timeout); #endif diff --git a/main-postlicyd.c b/main-postlicyd.c index 39adf81..7a5630e 100644 --- a/main-postlicyd.c +++ b/main-postlicyd.c @@ -88,14 +88,16 @@ typedef struct query_t { const char *encryption_cipher; const char *encryption_keysize; const char *etrn_domain; + + const char *eoq; } query_t; typedef struct plicyd_t { unsigned listener : 1; - unsigned watchwr : 1; int fd; buffer_t ibuf; buffer_t obuf; + query_t q; } plicyd_t; @@ -106,7 +108,6 @@ static plicyd_t *plicyd_new(void) return plicyd; } -#if 0 static void plicyd_delete(plicyd_t **plicyd) { if (*plicyd) { @@ -117,7 +118,6 @@ static void plicyd_delete(plicyd_t **plicyd) p_delete(plicyd); } } -#endif static int postfix_parsejob(query_t *query, char *p) { @@ -213,46 +213,50 @@ static int postfix_parsejob(query_t *query, char *p) #undef PARSE_CHECK } -static void *policy_run(int fd, void *data) +__attribute__((format(printf,2,0))) +static void policy_answer(plicyd_t *pcy, const char *fmt, ...) { - buffer_t buf; - - buffer_init(&buf); - for (;;) { - ssize_t search_offs = MAX(0, buf.len - 1); - int nb = buffer_read(&buf, fd, -1); - const char *eoq; - query_t q; - - if (nb < 0) { - if (errno == EAGAIN || errno == EINTR) - continue; - UNIXERR("read"); - break; - } - if (nb == 0) { - if (buf.len) - syslog(LOG_ERR, "unexpected end of data"); - break; - } - - eoq = strstr(buf.data + search_offs, "\n\n"); - if (!eoq) - continue; + va_list args; + va_start(args, fmt); + buffer_addvf(&pcy->obuf, fmt, args); + va_end(args); + buffer_addstr(&pcy->obuf, "\n\n"); + buffer_consume(&pcy->ibuf, pcy->q.eoq - pcy->ibuf.data); + epoll_modify(pcy->fd, EPOLLIN | EPOLLOUT, pcy); +} - if (postfix_parsejob(&q, buf.data) < 0) - break; +static void policy_process(plicyd_t *pcy) +{ + policy_answer(pcy, "DUNNO"); +} - buffer_consume(&buf, eoq + strlen("\n\n") - buf.data); - if (xwrite(fd, "DUNNO\n\n", strlen("DUNNO\n\n"))) { - UNIXERR("write"); - break; - } +static int policy_run(plicyd_t *pcy) +{ + ssize_t search_offs = MAX(0, pcy->ibuf.len - 1); + int nb = buffer_read(&pcy->ibuf, pcy->fd, -1); + const char *eoq; + + if (nb < 0) { + if (errno == EAGAIN || errno == EINTR) + return 0; + UNIXERR("read"); + return -1; + } + if (nb == 0) { + if (pcy->ibuf.len) + syslog(LOG_ERR, "unexpected end of data"); + return -1; } - buffer_wipe(&buf); - close(fd); - return NULL; + if (!(eoq = strstr(pcy->ibuf.data + search_offs, "\n\n"))) + return 0; + + if (postfix_parsejob(&pcy->q, pcy->ibuf.data) < 0) + return -1; + pcy->q.eoq = eoq + strlen("\n\n"); + epoll_modify(pcy->fd, 0, pcy); + policy_process(pcy); + return 0; } int start_listener(int port) @@ -261,7 +265,6 @@ int start_listener(int port) .sin_family = AF_INET, .sin_addr = { htonl(INADDR_LOOPBACK) }, }; - struct epoll_event evt = { .events = EPOLLIN }; plicyd_t *tmp; int sock; @@ -271,15 +274,29 @@ int start_listener(int port) return -1; } - evt.data.ptr = tmp = plicyd_new(); + tmp = plicyd_new(); tmp->fd = sock; tmp->listener = true; - if (epoll_ctl(epollfd, EPOLL_CTL_ADD, sock, &evt) < 0) { - UNIXERR("epoll_ctl"); - return -1; - } + epoll_register(sock, EPOLLIN, tmp); return 0; } + +void start_client(plicyd_t *d) +{ + plicyd_t *tmp; + int sock; + + sock = accept_nonblock(d->fd); + if (sock < 0) { + UNIXERR("accept"); + return; + } + + tmp = plicyd_new(); + tmp->fd = sock; + epoll_register(sock, EPOLLIN, tmp); +} + /* administrivia {{{ */ static int main_initialize(void) @@ -367,7 +384,7 @@ int main(int argc, char *argv[]) struct epoll_event evts[1024]; int n; - n = epoll_wait(epollfd, evts, countof(evts), -1); + n = epoll_select(evts, countof(evts), -1); if (n < 0) { if (errno != EAGAIN && errno != EINTR) { UNIXERR("epoll_wait"); @@ -380,18 +397,27 @@ int main(int argc, char *argv[]) plicyd_t *d = evts[n].data.ptr; if (d->listener) { - int fd = accept(d->fd, NULL, 0); - if (fd < 0) { - if (errno != EINTR && errno != EAGAIN) { - UNIXERR("accept"); - return EXIT_FAILURE; - } + start_client(d); + continue; + } + + if (evts[n].events & EPOLLIN) { + if (policy_run(d) < 0) { + plicyd_delete(&d); continue; } - thread_launch(policy_run, fd, NULL); + } + + if ((evts[n].events & EPOLLOUT) && d->obuf.len) { + if (buffer_write(&d->obuf, d->fd) < 0) { + plicyd_delete(&d); + continue; + } + if (!d->obuf.len) { + epoll_modify(d->fd, EPOLLIN, d); + } } } - threads_join(); } syslog(LOG_INFO, "Stopping..."); diff --git a/main-srsd.c b/main-srsd.c index 25bac08..ce4113b 100644 --- a/main-srsd.c +++ b/main-srsd.c @@ -55,7 +55,6 @@ typedef struct srsd_t { unsigned listener : 1; unsigned decoder : 1; - unsigned watchwr : 1; int fd; buffer_t ibuf; buffer_t obuf; @@ -101,6 +100,11 @@ void urldecode(char *s, char *end) int process_srs(srs_t *srs, const char *domain, srsd_t *srsd) { + int res = buffer_read(&srsd->ibuf, srsd->fd, -1); + + if ((res < 0 && errno != EINTR && errno != EAGAIN) || res == 0) + return -1; + while (srsd->ibuf.len > 4) { char buf[BUFSIZ], *p, *q, *nl; int err; @@ -166,7 +170,6 @@ int start_listener(int port, bool decoder) .sin_family = AF_INET, .sin_addr = { htonl(INADDR_LOOPBACK) }, }; - struct epoll_event evt = { .events = EPOLLIN }; srsd_t *tmp; int sock; @@ -176,20 +179,16 @@ int start_listener(int port, bool decoder) return -1; } - evt.data.ptr = tmp = srsd_new(); + tmp = srsd_new(); tmp->fd = sock; tmp->decoder = decoder; tmp->listener = true; - if (epoll_ctl(epollfd, EPOLL_CTL_ADD, sock, &evt) < 0) { - UNIXERR("epoll_ctl"); - return -1; - } + epoll_register(sock, EPOLLIN, tmp); return 0; } void start_client(srsd_t *srsd) { - struct epoll_event evt = { .events = EPOLLIN }; srsd_t *tmp; int sock; @@ -199,14 +198,10 @@ void start_client(srsd_t *srsd) return; } - evt.data.ptr = tmp = srsd_new(); + tmp = srsd_new(); tmp->decoder = srsd->decoder; tmp->fd = sock; - if (epoll_ctl(epollfd, EPOLL_CTL_ADD, sock, &evt) < 0) { - UNIXERR("epoll_ctl"); - srsd_delete(&tmp); - close(sock); - } + epoll_register(sock, EPOLLIN, tmp); } /* }}} */ @@ -260,7 +255,7 @@ int main_loop(srs_t *srs, const char *domain, int port_enc, int port_dec) struct epoll_event evts[1024]; int n; - n = epoll_wait(epollfd, evts, countof(evts), -1); + n = epoll_select(evts, countof(evts), -1); if (n < 0) { if (errno != EAGAIN && errno != EINTR) { UNIXERR("epoll_wait"); @@ -278,40 +273,23 @@ int main_loop(srs_t *srs, const char *domain, int port_enc, int port_dec) } if (evts[n].events & EPOLLIN) { - int res = buffer_read(&srsd->ibuf, srsd->fd, -1); - - if ((res < 0 && errno != EINTR && errno != EAGAIN) - || res == 0) - { - srsd_delete(&srsd); - continue; - } - if (process_srs(srs, domain, srsd) < 0) { srsd_delete(&srsd); continue; } + if (srsd->obuf.len) { + epoll_register(srsd->fd, EPOLLIN | EPOLLOUT, srsd); + } } if ((evts[n].events & EPOLLOUT) && srsd->obuf.len) { - int res = buffer_write(&srsd->obuf, srsd->fd); - if (res < 0) { + if (buffer_write(&srsd->obuf, srsd->fd) < 0) { srsd_delete(&srsd); continue; } - } - - if (srsd->watchwr == !srsd->obuf.len) { - struct epoll_event evt = { - .events = EPOLLIN | (srsd->obuf.len ? EPOLLOUT : 0), - .data.ptr = srsd, - }; - if (epoll_ctl(epollfd, EPOLL_CTL_MOD, srsd->fd, &evt) < 0) { - UNIXERR("epoll_ctl"); - srsd_delete(&srsd); - continue; + if (!srsd->obuf.len) { + epoll_modify(srsd->fd, EPOLLIN, srsd); } - srsd->watchwr = srsd->obuf.len != 0; } } }