diff --git a/CMakeLists.txt b/CMakeLists.txt index 861d19a..11bd950 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -390,6 +390,7 @@ endif() # Source files for 3proxy core set(3PROXY_CORE_SOURCES src/3proxy.c + src/loop.c src/auth.c src/acl.c src/limiter.c @@ -627,6 +628,14 @@ elseif(CMAKE_SYSTEM_NAME STREQUAL "Linux") target_link_libraries(3proxy PRIVATE dl) endif() +# The event loop on its own, for tests/cases/loop.py: not installed +add_executable(looptest tests/looptest.c src/loop.c) +target_include_directories(looptest PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src) +target_link_libraries(looptest PRIVATE Threads::Threads) +if(WIN32) + target_link_libraries(looptest PRIVATE ${WINDOWS_LIBS}) +endif() + # Build 3proxy_crypt utility add_executable(3proxy_crypt src/3proxy_crypt.c diff --git a/src/Makefile.inc b/src/Makefile.inc index ace05c8..7f6d8d4 100644 --- a/src/Makefile.inc +++ b/src/Makefile.inc @@ -31,6 +31,9 @@ ftp$(OBJSUFFICS): ftp.c proxy.h structures.h sockgetchar$(OBJSUFFICS): sockgetchar.c proxy.h structures.h $(CC) $(CFLAGS) sockgetchar.c +loop$(OBJSUFFICS): loop.c loop.h proxy.h structures.h + $(CC) $(CFLAGS) loop.c + proxy$(OBJSUFFICS): proxy.c proxy.h structures.h $(PROXYMAIN) $(CC) $(CFLAGS) $(DEFINEOPTION)WITHMAIN $(DEFINEOPTION)NOPORTMAP $(DEFINEOPTION)ANONYMOUS $(DEFINEOPTION)NOUDPMAIN proxy.c @@ -200,6 +203,6 @@ ssl$(OBJSUFFICS): ssl.c structures.h proxy.h ssl.h pcre$(OBJSUFFICS): pcre.c structures.h $(CC) $(COUT)pcre$(OBJSUFFICS) $(CFLAGS) $(DEFINEOPTION)WITH_PCRE pcre.c -$(BUILDDIR)3proxy$(EXESUFFICS): 3proxy$(OBJSUFFICS) mainfunc$(OBJSUFFICS) srvproxy$(OBJSUFFICS) srvpop3p$(OBJSUFFICS) srvimapp$(OBJSUFFICS) srvsmtpp$(OBJSUFFICS) srvftppr$(OBJSUFFICS) srvsocks$(OBJSUFFICS) srvtcppm$(OBJSUFFICS) srvtlspr$(OBJSUFFICS) srvauto$(OBJSUFFICS) srvudppm$(OBJSUFFICS) sockmap$(OBJSUFFICS) udpsockmap$(OBJSUFFICS) sockgetchar$(OBJSUFFICS) common$(OBJSUFFICS) sem$(OBJSUFFICS) auth$(OBJSUFFICS) acl$(OBJSUFFICS) limiter$(OBJSUFFICS) redirect$(OBJSUFFICS) authradius$(OBJSUFFICS) hash$(OBJSUFFICS) hashtables$(OBJSUFFICS) resolve$(OBJSUFFICS) sql$(OBJSUFFICS) conf$(OBJSUFFICS) log$(OBJSUFFICS) datatypes$(OBJSUFFICS) md4$(OBJSUFFICS) md5$(OBJSUFFICS) blake2$(OBJSUFFICS) 3proxy_crypt$(OBJSUFFICS) base64$(OBJSUFFICS) ftp$(OBJSUFFICS) stringtable$(OBJSUFFICS) $(HTTPSRV_OBJS) srvdnspr$(OBJSUFFICS) plugins$(OBJSUFFICS) mdhash$(OBJSUFFICS) $(SSL_OBJS) $(PCRE_OBJS) $(TRANSPARENT_OBJS) $(COMPATLIBS) $(VERSIONDEP) - $(LN) $(LNOUT)$(BUILDDIR)3proxy$(EXESUFFICS) $(LDFLAGS) $(VERFILE) 3proxy$(OBJSUFFICS) mainfunc$(OBJSUFFICS) auth$(OBJSUFFICS) acl$(OBJSUFFICS) limiter$(OBJSUFFICS) redirect$(OBJSUFFICS) authradius$(OBJSUFFICS) hash$(OBJSUFFICS) hashtables$(OBJSUFFICS) resolve$(OBJSUFFICS) sql$(OBJSUFFICS) conf$(OBJSUFFICS) datatypes$(OBJSUFFICS) srvauto$(OBJSUFFICS) srvproxy$(OBJSUFFICS) srvpop3p$(OBJSUFFICS) srvimapp$(OBJSUFFICS) srvsmtpp$(OBJSUFFICS) srvftppr$(OBJSUFFICS) srvsocks$(OBJSUFFICS) srvtcppm$(OBJSUFFICS) srvtlspr$(OBJSUFFICS) srvudppm$(OBJSUFFICS) sockmap$(OBJSUFFICS) udpsockmap$(OBJSUFFICS) sockgetchar$(OBJSUFFICS) common$(OBJSUFFICS) sem$(OBJSUFFICS) log$(OBJSUFFICS) 3proxy_crypt$(OBJSUFFICS) md4$(OBJSUFFICS) md5$(OBJSUFFICS) blake2$(OBJSUFFICS) base64$(OBJSUFFICS) ftp$(OBJSUFFICS) stringtable$(OBJSUFFICS) $(HTTPSRV_OBJS) srvdnspr$(OBJSUFFICS) plugins$(OBJSUFFICS) mdhash$(OBJSUFFICS) $(SSL_OBJS) $(PCRE_OBJS) $(TRANSPARENT_OBJS) $(COMPATLIBS) $(LIBS) $(PCRE_LIBS) +$(BUILDDIR)3proxy$(EXESUFFICS): 3proxy$(OBJSUFFICS) loop$(OBJSUFFICS) mainfunc$(OBJSUFFICS) srvproxy$(OBJSUFFICS) srvpop3p$(OBJSUFFICS) srvimapp$(OBJSUFFICS) srvsmtpp$(OBJSUFFICS) srvftppr$(OBJSUFFICS) srvsocks$(OBJSUFFICS) srvtcppm$(OBJSUFFICS) srvtlspr$(OBJSUFFICS) srvauto$(OBJSUFFICS) srvudppm$(OBJSUFFICS) sockmap$(OBJSUFFICS) udpsockmap$(OBJSUFFICS) sockgetchar$(OBJSUFFICS) common$(OBJSUFFICS) sem$(OBJSUFFICS) auth$(OBJSUFFICS) acl$(OBJSUFFICS) limiter$(OBJSUFFICS) redirect$(OBJSUFFICS) authradius$(OBJSUFFICS) hash$(OBJSUFFICS) hashtables$(OBJSUFFICS) resolve$(OBJSUFFICS) sql$(OBJSUFFICS) conf$(OBJSUFFICS) log$(OBJSUFFICS) datatypes$(OBJSUFFICS) md4$(OBJSUFFICS) md5$(OBJSUFFICS) blake2$(OBJSUFFICS) 3proxy_crypt$(OBJSUFFICS) base64$(OBJSUFFICS) ftp$(OBJSUFFICS) stringtable$(OBJSUFFICS) $(HTTPSRV_OBJS) srvdnspr$(OBJSUFFICS) plugins$(OBJSUFFICS) mdhash$(OBJSUFFICS) $(SSL_OBJS) $(PCRE_OBJS) $(TRANSPARENT_OBJS) $(COMPATLIBS) $(VERSIONDEP) + $(LN) $(LNOUT)$(BUILDDIR)3proxy$(EXESUFFICS) $(LDFLAGS) $(VERFILE) 3proxy$(OBJSUFFICS) loop$(OBJSUFFICS) mainfunc$(OBJSUFFICS) auth$(OBJSUFFICS) acl$(OBJSUFFICS) limiter$(OBJSUFFICS) redirect$(OBJSUFFICS) authradius$(OBJSUFFICS) hash$(OBJSUFFICS) hashtables$(OBJSUFFICS) resolve$(OBJSUFFICS) sql$(OBJSUFFICS) conf$(OBJSUFFICS) datatypes$(OBJSUFFICS) srvauto$(OBJSUFFICS) srvproxy$(OBJSUFFICS) srvpop3p$(OBJSUFFICS) srvimapp$(OBJSUFFICS) srvsmtpp$(OBJSUFFICS) srvftppr$(OBJSUFFICS) srvsocks$(OBJSUFFICS) srvtcppm$(OBJSUFFICS) srvtlspr$(OBJSUFFICS) srvudppm$(OBJSUFFICS) sockmap$(OBJSUFFICS) udpsockmap$(OBJSUFFICS) sockgetchar$(OBJSUFFICS) common$(OBJSUFFICS) sem$(OBJSUFFICS) log$(OBJSUFFICS) 3proxy_crypt$(OBJSUFFICS) md4$(OBJSUFFICS) md5$(OBJSUFFICS) blake2$(OBJSUFFICS) base64$(OBJSUFFICS) ftp$(OBJSUFFICS) stringtable$(OBJSUFFICS) $(HTTPSRV_OBJS) srvdnspr$(OBJSUFFICS) plugins$(OBJSUFFICS) mdhash$(OBJSUFFICS) $(SSL_OBJS) $(PCRE_OBJS) $(TRANSPARENT_OBJS) $(COMPATLIBS) $(LIBS) $(PCRE_LIBS) diff --git a/src/loop.c b/src/loop.c new file mode 100644 index 0000000..f84cc8c --- /dev/null +++ b/src/loop.c @@ -0,0 +1,683 @@ +/* + 3APA3A simplest proxy server + (c) 2002-2026 by Vladimir Dubrovin + + please read License Agreement +*/ + +#include "proxy.h" +#include "loop.h" +#include + +#ifdef LOOP_EPOLL +#include +#include +#endif +#ifdef LOOP_KQUEUE +#include +#endif + +/* Milliseconds of a clock which only goes forward. */ +uint64_t looptime(void){ +#ifdef _WIN32 + static LARGE_INTEGER freq; + LARGE_INTEGER c; + + if(!freq.QuadPart) QueryPerformanceFrequency(&freq); + QueryPerformanceCounter(&c); + return (uint64_t)(c.QuadPart / freq.QuadPart) * 1000 + + (uint64_t)(c.QuadPart % freq.QuadPart) * 1000 / (uint64_t)freq.QuadPart; +#else + struct timespec ts; + + clock_gettime(CLOCK_MONOTONIC, &ts); + return (uint64_t)ts.tv_sec * 1000 + (uint64_t)ts.tv_nsec / 1000000; +#endif +} + + +/* ---- what wakes the loop ---------------------------------------------- */ + +#ifndef LOOP_EPOLL +static void setnonblock(SOCKET s){ +#ifdef _WIN32 + unsigned long ul = 1; + ioctlsocket(s, FIONBIO, &ul); +#else + fcntl(s, F_SETFL, O_NONBLOCK | fcntl(s, F_GETFL)); +#endif +} +#endif + +static int wakeinit(struct loop *l){ +#ifdef LOOP_EPOLL + int fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); + + if(fd < 0) return 1; + l->wakerd = l->wakewr = fd; +#elif defined _WIN32 + /* no pipe to poll: a UDP socket sending to itself */ + struct sockaddr_in sa; + int len = sizeof(sa); + SOCKET s = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + + if(s == INVALID_SOCKET) return 1; + memset(&sa, 0, sizeof(sa)); + sa.sin_family = AF_INET; + sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + if(bind(s, (struct sockaddr *)&sa, sizeof(sa)) || getsockname(s, (struct sockaddr *)&sa, &len) || + connect(s, (struct sockaddr *)&sa, sizeof(sa))){ + closesocket(s); + return 1; + } + setnonblock(s); + l->wakerd = l->wakewr = s; +#else + int p[2]; + + if(pipe(p)) return 1; + setnonblock(p[0]); + setnonblock(p[1]); + fcntl(p[0], F_SETFD, FD_CLOEXEC); + fcntl(p[1], F_SETFD, FD_CLOEXEC); + l->wakerd = p[0]; + l->wakewr = p[1]; +#endif + return 0; +} + +static void wakefree(struct loop *l){ +#ifdef _WIN32 + closesocket(l->wakerd); +#else + close(l->wakerd); + if(l->wakewr != l->wakerd) close(l->wakewr); +#endif +} + +static void wake(struct loop *l){ +#ifdef LOOP_EPOLL + uint64_t one = 1; + if(write(l->wakewr, &one, sizeof(one)) != sizeof(one)){} +#elif defined _WIN32 + send(l->wakewr, "w", 1, 0); +#else + if(write(l->wakewr, "w", 1) != 1){} +#endif +} + +static void wakedrain(struct loop *l){ + char buf[64]; + +#ifdef _WIN32 + while(recv(l->wakerd, buf, sizeof(buf), 0) > 0); +#else + while(read(l->wakerd, buf, sizeof(buf)) > 0); +#endif +} + + +/* ---- the system's part: arming a socket for what it waits for ---------- */ + +#ifdef LOOP_KQUEUE +static int kqarm(struct loop *l, struct loopio *io, short events){ + struct kevent ch[2]; + int n = 0; + + if((events & POLLIN) && !(io->armed & POLLIN)) + EV_SET(ch + n++, io->fd, EVFILT_READ, EV_ADD, 0, 0, io); + else if(!(events & POLLIN) && (io->armed & POLLIN)) + EV_SET(ch + n++, io->fd, EVFILT_READ, EV_DELETE, 0, 0, io); + if((events & POLLOUT) && !(io->armed & POLLOUT)) + EV_SET(ch + n++, io->fd, EVFILT_WRITE, EV_ADD, 0, 0, io); + else if(!(events & POLLOUT) && (io->armed & POLLOUT)) + EV_SET(ch + n++, io->fd, EVFILT_WRITE, EV_DELETE, 0, 0, io); + if(n && kevent(l->fd, ch, n, NULL, 0, NULL) && (events & ~io->armed)) return 1; + io->armed = events; + return 0; +} +#endif + +#ifdef LOOP_EPOLL +static int eparm(struct loop *l, struct loopio *io, short events){ + struct epoll_event ev; + int op; + + memset(&ev, 0, sizeof(ev)); + ev.data.ptr = io; + if(events & POLLIN) ev.events |= EPOLLIN; + if(events & POLLOUT) ev.events |= EPOLLOUT; + /* not registered while nothing is waited for: an error or a hang up + would be reported all the same */ + if(!events) op = EPOLL_CTL_DEL; + else op = io->armed? EPOLL_CTL_MOD : EPOLL_CTL_ADD; + if(!events && !io->armed) return 0; + if(epoll_ctl(l->fd, op, io->fd, &ev) && events) return 1; + io->armed = events; + return 0; +} +#endif + +#ifdef LOOP_POLL +static int pollgrow(struct loop *l){ + unsigned n = l->maxpfds? l->maxpfds * 2 : 64; + struct pollfd *p; + struct loopio **q; + + if(!(p = realloc(l->pfds, n * sizeof(*p)))) return 1; + l->pfds = p; + if(!(q = realloc(l->pios, n * sizeof(*q)))) return 1; + l->pios = q; + l->maxpfds = n; + return 0; +} + +/* in the array while something is waited for: a hang up would be + reported all the same */ +static int pollarm(struct loop *l, struct loopio *io, short events){ + if(events && io->idx < 0){ + if(l->npfds == l->maxpfds && pollgrow(l)) return 1; + io->idx = (int)l->npfds++; + l->pfds[io->idx].fd = io->fd; + l->pios[io->idx] = io; + } + else if(!events && io->idx >= 0){ + unsigned last = --l->npfds; + + if((unsigned)io->idx != last){ + l->pfds[io->idx] = l->pfds[last]; + l->pios[io->idx] = l->pios[last]; + l->pios[io->idx]->idx = io->idx; + } + io->idx = -1; + } + if(io->idx >= 0) { + l->pfds[io->idx].events = events; + l->pfds[io->idx].revents = 0; + } + io->armed = events; + return 0; +} +#endif + +static int arm(struct loop *l, struct loopio *io, short events){ +#ifdef LOOP_EPOLL + return eparm(l, io, events); +#elif defined LOOP_KQUEUE + return kqarm(l, io, events); +#else + return pollarm(l, io, events); +#endif +} + + +/* ---- timers --------------------------------------------------------------- */ + +static void linkinit(struct looplink *s){ + s->prev = s->next = s; +} + +static void linkhead(struct looplink *s, struct looplink *e){ + e->next = s->next; + e->prev = s; + s->next->prev = e; + s->next = e; +} + +static void linktail(struct looplink *s, struct looplink *e){ + e->prev = s->prev; + e->next = s; + s->prev->next = e; + s->prev = e; +} + +static void unlink_(struct looplink *e){ + e->prev->next = e->next; + e->next->prev = e->prev; + e->prev = e->next = e; +} + +#define HEAPLESS(a, b) ((a)->deadline < (b)->deadline) + +static void heapput(struct loop *l, unsigned i, struct looptimer *t){ + l->heap[i] = t; + t->heapidx = i + 1; +} + +static void heapup(struct loop *l, unsigned i){ + struct looptimer *t = l->heap[i]; + + while(i){ + unsigned p = (i - 1) / 2; + + if(!HEAPLESS(t, l->heap[p])) break; + heapput(l, i, l->heap[p]); + i = p; + } + heapput(l, i, t); +} + +static void heapdown(struct loop *l, unsigned i){ + struct looptimer *t = l->heap[i]; + + for(;;){ + unsigned c = 2 * i + 1; + + if(c >= l->nheap) break; + if(c + 1 < l->nheap && HEAPLESS(l->heap[c + 1], l->heap[c])) c++; + if(!HEAPLESS(l->heap[c], t)) break; + heapput(l, i, l->heap[c]); + i = c; + } + heapput(l, i, t); +} + +static int heapadd(struct loop *l, struct looptimer *t){ + if(l->nheap == l->maxheap){ + unsigned n = l->maxheap? l->maxheap * 2 : 64; + struct looptimer **h = realloc(l->heap, n * sizeof(*h)); + + if(!h) return 1; + l->heap = h; + l->maxheap = n; + } + heapput(l, l->nheap++, t); + heapup(l, l->nheap - 1); + t->where = LT_HEAP; + return 0; +} + +static void heapremove(struct loop *l, struct looptimer *t){ + unsigned i = t->heapidx - 1; + + t->heapidx = 0; + if(i != --l->nheap){ + heapput(l, i, l->heap[l->nheap]); + heapdown(l, i); + heapup(l, l->heap[i]->heapidx - 1); + } +} + +static void timerremove(struct loop *l, struct looptimer *t){ + switch(t->where){ + case LT_WHEEL: + l->nwheel--; + /* fall through */ + case LT_DUE: + unlink_(&t->link); + break; + case LT_HEAP: + heapremove(l, t); + break; + } + t->where = LT_NONE; +} + +/* A timer due within the second the wheel is past waits in the heap, to + the millisecond. One due later waits in the slot of its second: at the + head, all of them due in the same second, if that is within a turn; + else behind them, kept in the order of their seconds, if it is not due + before the last there, else at the head of the last slot of the turn, + to be looked at again then. */ +static void timerinsert(struct loop *l, struct looptimer *t){ + uint64_t s = t->deadline / 1000; + struct looplink *slot; + + if(s < l->second){ + if(heapadd(l, t)){ + /* no memory for the heap: due now rather than lost */ + linktail(&l->due, &t->link); + t->where = LT_DUE; + } + return; + } + slot = l->slots + (s & (LOOPSLOTS - 1)); + if(s - l->second < LOOPSLOTS) linkhead(slot, &t->link); + else if(slot->prev == slot || ((struct looptimer *)slot->prev)->slotted <= s) linktail(slot, &t->link); + else { + s = l->second + LOOPSLOTS - 1; + slot = l->slots + (s & (LOOPSLOTS - 1)); + linkhead(slot, &t->link); + } + t->slotted = s; + t->where = LT_WHEEL; + l->nwheel++; +} + +/* Sets t to fire at deadline (ms of looptime()), wherever it was. */ +void looptimerset(struct loop *l, struct looptimer *t, uint64_t deadline){ + timerremove(l, t); + t->deadline = deadline; + timerinsert(l, t); +} + +/* Moves t later, which costs nothing while it waits in the wheel: it is + looked at when the old deadline comes and moved then. */ +void looptimerlater(struct loop *l, struct looptimer *t, uint64_t deadline){ + if(t->where == LT_WHEEL && deadline >= t->deadline) t->deadline = deadline; + else looptimerset(l, t, deadline); +} + +void looptimerdel(struct loop *l, struct looptimer *t){ + timerremove(l, t); +} + +/* The wheel turned to now: what its slots hold for the seconds passed is + due, or within the second in the heap, or moved later where it was; + what the heap holds up to now is due. */ +static void timersturn(struct loop *l){ + uint64_t sec = l->now / 1000; + + while(l->nwheel && l->second <= sec){ + uint64_t x = l->second++; + struct looplink *slot = l->slots + (x & (LOOPSLOTS - 1)), *e, *next; + + for(e = slot->next; e != slot; e = next){ + struct looptimer *t = (struct looptimer *)e; + + next = e->next; + if(t->slotted > x) break; + unlink_(e); + l->nwheel--; + if(t->deadline <= l->now){ + linktail(&l->due, e); + t->where = LT_DUE; + } + else timerinsert(l, t); + } + } + if(!l->nwheel && l->second <= sec) l->second = sec + 1; + while(l->nheap && l->heap[0]->deadline <= l->now){ + struct looptimer *t = l->heap[0]; + + heapremove(l, t); + linktail(&l->due, &t->link); + t->where = LT_DUE; + } +} + +/* What is due fires: what a timer firing sets due again fires on the + next turn. */ +static void timersfire(struct loop *l){ + struct looplink now; + + if(l->due.next == &l->due) return; + now.next = l->due.next; + now.prev = l->due.prev; + now.next->prev = now.prev->next = &now; + linkinit(&l->due); + while(now.next != &now){ + struct looptimer *t = (struct looptimer *)now.next; + + unlink_(&t->link); + t->where = LT_NONE; + t->func(l, t); + } +} + + +/* ---- the loop ----------------------------------------------------------- */ + +int loopinit(struct loop *l){ + unsigned i; + + memset(l, 0, sizeof(*l)); + for(i = 0; i < LOOPSLOTS; i++) linkinit(l->slots + i); + linkinit(&l->due); + l->postedtail = &l->posted; + l->clock = looptime; + l->now = l->clock(); + l->second = l->now / 1000 + 1; + if(wakeinit(l)) return 1; +#ifdef LOOP_EPOLL + { + struct epoll_event ev; + + memset(&ev, 0, sizeof(ev)); + ev.events = EPOLLIN; + ev.data.ptr = NULL; + if((l->fd = epoll_create1(EPOLL_CLOEXEC)) < 0 || + epoll_ctl(l->fd, EPOLL_CTL_ADD, l->wakerd, &ev)){ + if(l->fd >= 0) close(l->fd); + wakefree(l); + return 1; + } + } +#elif defined LOOP_KQUEUE + { + struct kevent ch; + + if((l->fd = kqueue()) < 0) { + wakefree(l); + return 1; + } + fcntl(l->fd, F_SETFD, FD_CLOEXEC); + EV_SET(&ch, l->wakerd, EVFILT_READ, EV_ADD, 0, 0, NULL); + if(kevent(l->fd, &ch, 1, NULL, 0, NULL)){ + close(l->fd); + wakefree(l); + return 1; + } + } +#else + if(pollgrow(l)){ + wakefree(l); + return 1; + } + l->pfds[0].fd = l->wakerd; + l->pfds[0].events = POLLIN; + l->pios[0] = NULL; + l->npfds = 1; +#endif + _3proxy_mutex_init(&l->mutex); + return 0; +} + +/* What was added or set is the caller's to take back first. */ +void loopfree(struct loop *l){ +#if defined LOOP_EPOLL || defined LOOP_KQUEUE + close(l->fd); +#else + free(l->pfds); + free(l->pios); +#endif + wakefree(l); + free(l->heap); + _3proxy_mutex_destroy(&l->mutex); +} + +/* io->fd, io->events and io->func set: waited for from now. */ +int loopadd(struct loop *l, struct loopio *io){ + short events = io->events; + + io->armed = 0; + io->idx = -1; + io->events = 0; + return loopmod(l, io, events); +} + +int loopmod(struct loop *l, struct loopio *io, short events){ + events &= (POLLIN | POLLOUT); + if(arm(l, io, events)) return 1; + io->events = events; + return 0; +} + +/* Before the socket is closed: no longer waited for, nor called if ready + in this turn. */ +void loopdel(struct loop *l, struct loopio *io){ + int i; + + arm(l, io, 0); + io->events = 0; + for(i = 0; i < l->nready; i++) if(l->ready[i] == io) l->ready[i] = NULL; +} + +/* m->func is to be called by the loop: from any thread. */ +void looppost(struct loop *l, struct loopmsg *m){ + int first; + + m->next = NULL; + _3proxy_mutex_lock(&l->mutex); + *l->postedtail = m; + l->postedtail = &m->next; + first = !l->woken; + l->woken = 1; + _3proxy_mutex_unlock(&l->mutex); + if(first) wake(l); +} + +/* looprun() returns once this turn is over: from any thread. */ +void loopstop(struct loop *l){ + int first; + + l->stop = 1; + _3proxy_mutex_lock(&l->mutex); + first = !l->woken; + l->woken = 1; + _3proxy_mutex_unlock(&l->mutex); + if(first) wake(l); +} + +static void postedrun(struct loop *l){ + struct loopmsg *m; + + _3proxy_mutex_lock(&l->mutex); + m = l->posted; + l->posted = NULL; + l->postedtail = &l->posted; + l->woken = 0; + _3proxy_mutex_unlock(&l->mutex); + while(m){ + struct loopmsg *next = m->next; + + m->func(l, m); + m = next; + } +} + +static void readyadd(struct loop *l, struct loopio *io, short revents){ + int i; + + /* kqueue reports reading and writing apart */ + for(i = 0; i < l->nready; i++) if(l->ready[i] == io) { + l->readyev[i] |= revents; + return; + } + l->ready[l->nready] = io; + l->readyev[l->nready++] = revents; +} + +/* How long the system may be waited for: until the next second of the + wheel, or what is first in the heap, or maxwait (-1 for no limit). */ +static int waittime(struct loop *l, int maxwait){ + uint64_t until = 0; + int have = 0; + + if(l->due.next != &l->due) return 0; + if(l->nwheel){ + until = l->second * 1000; + have = 1; + } + if(l->nheap && (!have || l->heap[0]->deadline < until)){ + until = l->heap[0]->deadline; + have = 1; + } + if(!have) return maxwait; + if(until <= l->now) return 0; + if(until - l->now > INT_MAX) return maxwait; + if(maxwait >= 0 && (uint64_t)maxwait < until - l->now) return maxwait; + return (int)(until - l->now); +} + +/* One turn: waits up to maxwait ms (-1: for as long as nothing comes), + then calls what is posted, what is ready, what is due. Returns the + number of sockets found ready, -1 if waiting failed. */ +int looponce(struct loop *l, int maxwait){ + int n, i, timeo, woken = 0; + + timeo = waittime(l, maxwait); + l->nready = 0; +#ifdef LOOP_EPOLL + { + struct epoll_event ev[LOOPREADY]; + + n = epoll_wait(l->fd, ev, LOOPREADY, timeo); + for(i = 0; i < n; i++){ + struct loopio *io = ev[i].data.ptr; + short revents = 0; + + if(!io){ + woken = 1; + continue; + } + if(ev[i].events & EPOLLIN) revents |= POLLIN; + if(ev[i].events & EPOLLOUT) revents |= POLLOUT; + if(ev[i].events & EPOLLERR) revents |= POLLERR; + if(ev[i].events & EPOLLHUP) revents |= POLLHUP; + readyadd(l, io, revents); + } + } +#elif defined LOOP_KQUEUE + { + struct kevent ev[LOOPREADY]; + struct timespec ts, *tsp = NULL; + + if(timeo >= 0){ + ts.tv_sec = timeo / 1000; + ts.tv_nsec = (long)(timeo % 1000) * 1000000; + tsp = &ts; + } + n = kevent(l->fd, NULL, 0, ev, LOOPREADY, tsp); + for(i = 0; i < n; i++){ + struct loopio *io = ev[i].udata; + short revents = 0; + + if(!io){ + woken = 1; + continue; + } + if(ev[i].flags & EV_ERROR) revents |= POLLERR; + else if(ev[i].filter == EVFILT_READ) revents |= POLLIN; + else if(ev[i].filter == EVFILT_WRITE) revents |= POLLOUT; + if(ev[i].flags & EV_EOF) revents |= POLLHUP; + readyadd(l, io, revents); + } + } +#else + { + unsigned j; + +#if defined WITH_POLL + n = poll(l->pfds, (nfds_t)l->npfds, timeo); +#elif defined WITH_WSAPOLL + n = WSAPoll(l->pfds, l->npfds, timeo); +#else + n = mypoll(l->pfds, l->npfds, timeo); +#endif + if(n > 0 && l->pfds[0].revents) woken = 1; + for(j = 1; n > 0 && j < l->npfds && l->nready < LOOPREADY; j++) + if(l->pfds[j].revents) readyadd(l, l->pios[j], l->pfds[j].revents); + } +#endif + if(n < 0 && errno != EINTR) return -1; + l->now = l->clock(); + if(woken){ + wakedrain(l); + postedrun(l); + } + for(i = 0; i < l->nready; i++){ + struct loopio *io = l->ready[i]; + + if(io) io->func(l, io, l->readyev[i]); + } + n = l->nready; + l->nready = 0; + timersturn(l); + timersfire(l); + return n; +} + +void looprun(struct loop *l){ + while(!l->stop) if(looponce(l, -1) < 0) break; +} diff --git a/src/loop.h b/src/loop.h new file mode 100644 index 0000000..02dc95d --- /dev/null +++ b/src/loop.h @@ -0,0 +1,118 @@ +/* + 3APA3A simplest proxy server + (c) 2002-2026 by Vladimir Dubrovin + + please read License Agreement +*/ + +/* An event loop: one thread waiting for many sockets and timers at once, + calling back what waits for them. Sockets are waited for with epoll on + Linux, kqueue on the BSDs and macOS, poll() (WSAPoll() on Windows) + elsewhere or with LOOP_POLL. A loop is run by one thread: only + looppost() and loopstop() may be called by another. */ + +#ifndef _LOOP_H +#define _LOOP_H + +#if !defined LOOP_POLL && defined __linux__ +#define LOOP_EPOLL +#elif !defined LOOP_POLL && (defined __APPLE__ || defined __FreeBSD__ || defined __NetBSD__ || defined __OpenBSD__ || defined __DragonFly__) +#define LOOP_KQUEUE +#elif !defined LOOP_POLL +#define LOOP_POLL +#endif + +/* Seconds the wheel of timers turns in, a power of 2: a timer due within + as many seconds waits in the slot of its second, a later one in the same + slot behind them, and is looked at once a turn. */ +#define LOOPSLOTS 1024 +/* sockets ready taken from the system at once */ +#define LOOPREADY 64 + +struct loop; + +struct looplink { + struct looplink *prev, *next; +}; + +/* A timer: func is called once looptime() reaches deadline. Its own until + set, and once it fired or was deleted. */ +struct looptimer { + struct looplink link; /* in a slot of the wheel, or due */ + uint64_t deadline; /* ms of looptime() */ + uint64_t slotted; /* the second of the slot it waits in */ + unsigned heapidx; /* its place in the heap, from 1; 0 if not there */ + int where; /* LT_* */ + void (*func)(struct loop *l, struct looptimer *t); + void *data; +}; + +#define LT_NONE 0 +#define LT_WHEEL 1 +#define LT_HEAP 2 +#define LT_DUE 3 + +/* A socket: func is called with revents (POLLIN, POLLOUT, POLLERR, + POLLHUP) while it is ready for events, level-triggered. With events 0 + it stays added but is not waited for. */ +struct loopio { + SOCKET fd; + short events; /* POLLIN, POLLOUT */ + short armed; /* what the system waits for */ + int idx; /* poll(): its place in the array, -1 if none */ + void (*func)(struct loop *l, struct loopio *io, short revents); + void *data; +}; + +/* What another thread gives the loop: func is called by the loop's + thread. */ +struct loopmsg { + struct loopmsg *next; + void (*func)(struct loop *l, struct loopmsg *m); + void *data; +}; + +struct loop { +#if defined LOOP_EPOLL || defined LOOP_KQUEUE + int fd; +#else + struct pollfd *pfds; + struct loopio **pios; + unsigned npfds, maxpfds; +#endif + /* what wakes the loop from another thread: an eventfd, the read end + of a pipe, or on Windows a UDP socket sending to itself */ + SOCKET wakerd, wakewr; + _3proxy_mutex_t mutex; + struct loopmsg *posted, **postedtail; + int woken; + volatile int stop; + struct looplink slots[LOOPSLOTS]; + struct looplink due; + struct looptimer **heap; + unsigned nheap, maxheap; + unsigned nwheel; + uint64_t now; /* looptime() of this turn */ + uint64_t second; /* the next second of the wheel to turn to */ + uint64_t (*clock)(void); /* looptime(), but for tests */ + /* the sockets ready, being called: one deleted meanwhile is NULL */ + struct loopio *ready[LOOPREADY]; + short readyev[LOOPREADY]; + int nready; +}; + +uint64_t looptime(void); +int loopinit(struct loop *l); +void loopfree(struct loop *l); +int loopadd(struct loop *l, struct loopio *io); +int loopmod(struct loop *l, struct loopio *io, short events); +void loopdel(struct loop *l, struct loopio *io); +void looptimerset(struct loop *l, struct looptimer *t, uint64_t deadline); +void looptimerlater(struct loop *l, struct looptimer *t, uint64_t deadline); +void looptimerdel(struct loop *l, struct looptimer *t); +void looppost(struct loop *l, struct loopmsg *m); +void loopstop(struct loop *l); +int looponce(struct loop *l, int maxwait); +void looprun(struct loop *l); + +#endif diff --git a/tests/README.md b/tests/README.md index dcfceca..f146eb1 100644 --- a/tests/README.md +++ b/tests/README.md @@ -68,6 +68,11 @@ Note that access rules accumulate until `flush`, so a service section that means to stand on its own should start with one - otherwise an earlier `allow *` matches first and the rule under test is never reached. +The event loop (`src/loop.c`) is tested on its own by `tests/looptest.c`, +which CMake builds next to the 3proxy binary and the `loop` case runs: timers +on a clock of its own, through several turns of the wheel, sockets, and +other threads. Where it was not built the case is skipped. + ## What is not covered yet 42 of the 112 configuration commands appear in a test, and the count says diff --git a/tests/cases/loop.py b/tests/cases/loop.py new file mode 100644 index 0000000..328830b --- /dev/null +++ b/tests/cases/loop.py @@ -0,0 +1,30 @@ +"""The event loop on its own: tests/looptest.c, built by CMake next to the +3proxy binary, run with a few seeds for its timers. Skipped where it was not +built (the Makefiles do not build it).""" + +import os +import subprocess + + +def run(t): + name = "looptest.exe" if os.name == "nt" else "looptest" + path = os.path.join(os.path.dirname(os.path.abspath(t.binary)), name) + if not os.path.exists(path): + t.skip("looptest is built by CMake only") + return + for seed in ("1", "2", "3"): + try: + done = subprocess.run([path, seed], stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, timeout=120) + except subprocess.TimeoutExpired: + t.fail(f"looptest {seed} finishes") + continue + out = done.stdout.decode(errors="replace") + for line in out.splitlines(): + if line.startswith("ok "): + t.ok(f"seed {seed}: {line[3:]}") + elif line.startswith("FAIL "): + label, _, what = line[5:].partition(": ") + t.fail(f"seed {seed}: {label}", actual=what) + if done.returncode and "FAIL " not in out: + t.fail(f"looptest {seed} exits cleanly", 0, done.returncode) diff --git a/tests/looptest.c b/tests/looptest.c new file mode 100644 index 0000000..c1b03dc --- /dev/null +++ b/tests/looptest.c @@ -0,0 +1,390 @@ +/* + 3APA3A simplest proxy server + (c) 2002-2026 by Vladimir Dubrovin + + please read License Agreement +*/ + +/* The event loop on its own: tests/cases/loop.py runs it. Each check is a + line, "ok LABEL" or "FAIL LABEL: what", and the exit code is the number + of failures. */ + +#include "../src/proxy.h" +#include "../src/loop.h" + +static int failures; + +static void check(int cond, const char *label, const char *what){ + if(cond) printf("ok %s\n", label); + else { + printf("FAIL %s: %s\n", label, what); + failures++; + } + fflush(stdout); +} + +static unsigned long long rnd = 88172645463325252ULL; + +static unsigned long long rand64(void){ + rnd ^= rnd << 13; + rnd ^= rnd >> 7; + rnd ^= rnd << 17; + return rnd; +} + +static uint64_t rand_upto(uint64_t n){ + return n? rand64() % n : 0; +} + + +/* ---- timers on a clock of the test's ------------------------------------ */ + +#define NTIMERS 20000 + +static uint64_t fakenow; + +static uint64_t fakeclock(void){ + return fakenow; +} + +struct ttimer { + struct looptimer t; + uint64_t expect; /* the deadline it has now */ + int fired, deleted, pending, rearmed; +}; + +static struct ttimer timers[NTIMERS]; +static uint64_t prevnow, latest; +static int early, late, twice, wrongdead; + +static void tfire(struct loop *l, struct looptimer *lt){ + struct ttimer *t = (struct ttimer *)lt; + + if(t->deleted) wrongdead++; + if(!t->pending) twice++; + if(l->now < t->expect) early++; + /* fired on the first turn its time had come */ + if(prevnow >= t->expect) late++; + t->pending = 0; + t->fired++; + /* some set themselves again from their own call */ + if(!t->rearmed && rand_upto(20) == 0){ + t->rearmed = 1; + t->expect = l->now + 1 + rand_upto(3000000); + if(t->expect > latest) latest = t->expect; + t->pending = 1; + looptimerset(l, lt, t->expect); + } +} + +static void timertest(void){ + struct loop l; + uint64_t start; + int i, pending, lost = 0, deletedfired = 0, steps = 0; + char what[256]; + + if(loopinit(&l)){ + check(0, "timers: a loop is made", "loopinit() failed"); + return; + } + fakenow = start = 1000000ULL * 1000 + 123; + l.clock = fakeclock; + l.now = fakenow; + l.second = fakenow / 1000 + 1; + prevnow = fakenow; + + for(i = 0; i < NTIMERS; i++){ + struct ttimer *t = timers + i; + uint64_t r = rand_upto(10); + + /* within the second, within a turn of the wheel, past it */ + if(r < 4) t->expect = start + 1 + rand_upto(2000); + else if(r < 8) t->expect = start + 1 + rand_upto(1200000); + else t->expect = start + 1024000 + rand_upto(3000000); + t->t.func = tfire; + t->pending = 1; + looptimerset(&l, &t->t, t->expect); + if(t->expect > latest) latest = t->expect; + } + + /* until the last deadline ever set is past */ + while(fakenow <= latest + 2000){ + /* moved later, set earlier, deleted: some pending timer */ + for(i = 0; i < 20; i++){ + struct ttimer *t = timers + rand_upto(NTIMERS); + uint64_t r = rand_upto(10); + + if(!t->pending) continue; + if(r < 6){ + t->expect += rand_upto(700000); + if(t->expect > latest) latest = t->expect; + looptimerlater(&l, &t->t, t->expect); + } + else if(r < 8 && t->expect > fakenow + 1){ + t->expect = fakenow + 1 + rand_upto(t->expect - fakenow - 1); + looptimerset(&l, &t->t, t->expect); + } + else if(r < 9){ + looptimerdel(&l, &t->t); + t->pending = 0; + t->deleted = 1; + } + } + prevnow = fakenow; + /* steps of a few ms to a second, and now and then a long stall */ + if(rand_upto(500) == 0) fakenow += 1 + rand_upto(300000); + else fakenow += 1 + rand_upto(rand_upto(4) ? 1500 : 20); + looponce(&l, 0); + steps++; + } + pending = 0; + for(i = 0; i < NTIMERS; i++){ + if(timers[i].pending) pending++; + if(!timers[i].deleted && !timers[i].fired) lost++; + if(timers[i].deleted && timers[i].fired && !timers[i].rearmed) deletedfired++; + } + sprintf(what, "%d left pending, %d never fired", pending, lost); + check(!pending && !lost, "timers: every timer fires", what); + sprintf(what, "%d early", early); + check(!early, "timers: none fires before its deadline", what); + sprintf(what, "%d fired on a later turn than the first past their deadline", late); + check(!late, "timers: each fires on the first turn its deadline is past", what); + sprintf(what, "%d twice, %d deleted", twice, wrongdead + deletedfired); + check(!twice && !wrongdead && !deletedfired, "timers: once, and never once deleted", what); + sprintf(what, "%u in the wheel, %u in the heap", l.nwheel, l.nheap); + check(!l.nwheel && !l.nheap, "timers: nothing left behind", what); + printf("# %d steps of the clock\n", steps); + loopfree(&l); +} + + +/* ---- sockets -------------------------------------------------------------- */ + +static int pairsock(SOCKET *a, SOCKET *b){ + struct sockaddr_in sa; + SASIZETYPE len = sizeof(sa); + SOCKET ls = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + + *a = *b = INVALID_SOCKET; + if(ls == INVALID_SOCKET) return 1; + memset(&sa, 0, sizeof(sa)); + sa.sin_family = AF_INET; + sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + if(bind(ls, (struct sockaddr *)&sa, sizeof(sa)) || listen(ls, 1) || + getsockname(ls, (struct sockaddr *)&sa, &len) || + (*a = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) == INVALID_SOCKET || + connect(*a, (struct sockaddr *)&sa, sizeof(sa)) || + (*b = accept(ls, NULL, NULL)) == INVALID_SOCKET){ +#ifdef _WIN32 + closesocket(ls); +#else + close(ls); +#endif + return 1; + } +#ifdef _WIN32 + closesocket(ls); +#else + close(ls); +#endif + return 0; +} + +static void closesock(SOCKET s){ +#ifdef _WIN32 + closesocket(s); +#else + close(s); +#endif +} + +struct tio { + struct loopio io; + int calls; + short revents; + struct tio *other; /* deleted by this one's call */ +}; + +static void iocall(struct loop *l, struct loopio *io, short revents){ + struct tio *t = (struct tio *)io; + + t->calls++; + t->revents |= revents; + if(t->other) loopdel(l, &t->other->io); +} + +static void iotest(void){ + struct loop l; + struct tio a, c; + SOCKET s1, s2, s3, s4; + char buf[16]; + int i; + + memset(&a, 0, sizeof(a)); + memset(&c, 0, sizeof(c)); + if(loopinit(&l) || pairsock(&s1, &s2) || pairsock(&s3, &s4)){ + check(0, "sockets: a loop and sockets are made", "failed"); + return; + } + a.io.fd = s1; + a.io.events = POLLIN; + a.io.func = iocall; + check(!loopadd(&l, &a.io), "sockets: a socket is added", "loopadd() failed"); + + looponce(&l, 100); + check(a.calls == 0, "sockets: nothing to read, no call", "called"); + + send(s2, "x", 1, 0); + for(i = 0; i < 10 && !a.calls; i++) looponce(&l, 200); + check(a.calls >= 1 && (a.revents & POLLIN), "sockets: what comes is readable", "no POLLIN"); + recv(s1, buf, sizeof(buf), 0); + + a.calls = 0; + a.revents = 0; + loopmod(&l, &a.io, POLLOUT); + for(i = 0; i < 10 && !a.calls; i++) looponce(&l, 200); + check(a.calls >= 1 && (a.revents & POLLOUT) && !(a.revents & POLLIN), + "sockets: waited for writing, writable", "no POLLOUT"); + + loopmod(&l, &a.io, 0); + a.calls = 0; + send(s2, "y", 1, 0); + looponce(&l, 200); + check(a.calls == 0, "sockets: waited for nothing, not called", "called"); + + /* both ready: whichever is called first deletes the other */ + loopmod(&l, &a.io, POLLIN); + c.io.fd = s3; + c.io.events = POLLIN; + c.io.func = iocall; + loopadd(&l, &c.io); + a.other = &c; + c.other = &a; + send(s4, "z", 1, 0); + a.calls = c.calls = 0; + for(i = 0; i < 10 && a.calls + c.calls == 0; i++) looponce(&l, 200); + check(a.calls + c.calls == 1, "sockets: one deleted in this turn is not called", "both called"); + + /* the one deleted can be added again; the far side closing it is + readable */ + a.other = c.other = NULL; + if(!a.calls) loopadd(&l, &a.io); + if(!c.calls) loopadd(&l, &c.io); + loopdel(&l, &c.io); + recv(s1, buf, sizeof(buf), 0); + a.calls = 0; + a.revents = 0; + closesock(s2); + for(i = 0; i < 10 && !a.calls; i++) looponce(&l, 200); + check(a.calls >= 1 && (a.revents & (POLLIN | POLLHUP)) && recv(s1, buf, sizeof(buf), 0) == 0, + "sockets: the far side closing is readable", "not reported"); + + loopdel(&l, &a.io); + closesock(s1); + closesock(s3); + closesock(s4); + loopfree(&l); +} + + +/* ---- real time, and other threads --------------------------------------- */ + +static int timerfired; + +static void realfire(struct loop *l, struct looptimer *t){ + timerfired = 1; +} + +static int posted; + +static void postcall(struct loop *l, struct loopmsg *m){ + posted = 1; +} + +static struct loop tl; +static struct loopmsg msg; +static int dostop; + +#ifdef _WIN32 +static unsigned __stdcall poster(void *arg){ + Sleep(100); +#else +static void *poster(void *arg){ + usleep(100000); +#endif + if(dostop) loopstop(&tl); + else { + msg.func = postcall; + looppost(&tl, &msg); + } + return 0; +} + +static int startposter(void){ +#ifdef _WIN32 + HANDLE h = (HANDLE)_beginthreadex(NULL, 0, poster, NULL, 0, NULL); + + if(!h) return 1; + CloseHandle(h); + return 0; +#else + pthread_t t; + + if(pthread_create(&t, NULL, poster, NULL)) return 1; + pthread_detach(t); + return 0; +#endif +} + +static void realtest(void){ + struct looptimer t; + uint64_t start, took; + char what[64]; + + if(loopinit(&tl)){ + check(0, "real time: a loop is made", "loopinit() failed"); + return; + } + memset(&t, 0, sizeof(t)); + t.func = realfire; + start = looptime(); + looptimerset(&tl, &t, start + 150); + while(!timerfired) looponce(&tl, -1); + took = looptime() - start; + sprintf(what, "fired after %u ms", (unsigned)took); + check(took >= 150 && took < 1100, "real time: a timer fires when it is due", what); + + start = looptime(); + if(startposter()) check(0, "real time: a thread is started", "failed"); + else { + looponce(&tl, 5000); + took = looptime() - start; + sprintf(what, "after %u ms", (unsigned)took); + check(posted && took < 2000, "real time: a message from another thread wakes the loop", what); + } + + dostop = 1; + start = looptime(); + if(startposter()) check(0, "real time: a thread is started", "failed"); + else { + looprun(&tl); + took = looptime() - start; + sprintf(what, "after %u ms", (unsigned)took); + check(took < 2000, "real time: loopstop() from another thread ends looprun()", what); + } + loopfree(&tl); +} + +/* looptest [SEED]: another seed, other timers */ +int main(int argc, char **argv){ +#ifdef _WIN32 + WSADATA wd; + + WSAStartup(MAKEWORD(2, 2), &wd); +#endif + if(argc > 1 && strtoull(argv[1], NULL, 10)) rnd = strtoull(argv[1], NULL, 10); + timertest(); + iotest(); + realtest(); + return failures; +}