An event loop: sockets and timers for one thread

src/loop.c, not used yet: one thread waiting for many sockets and
timers, calling back what waits for them, for clients to be served by
a thread each serves many of.

Sockets are waited for with epoll on Linux, kqueue on the BSDs and
macOS, poll() (WSAPoll() on Windows) elsewhere or with LOOP_POLL,
level-triggered; one with nothing to wait for is taken from the system,
or an error or a hang up would be reported all the same. One deleted
while the sockets ready are being called is not called.

Timers wait in a wheel of 1024 one-second slots. One due within a turn
waits at the head of the slot of its second, all there due in the same
second; one due later behind them, in the order of their seconds, or,
if due before the last there, at the head of the last slot of the
turn, looked at again then. Moving a timer later costs nothing while
it waits in the wheel: it is looked at when its old deadline comes and
moved then. When its slot comes a timer due within the second waits in
a heap, so it fires to the millisecond, as does one set within the
second. The clock is monotonic.

Another thread posts to the loop, waking it with an eventfd, a pipe, or
on Windows a UDP socket sending to itself.

tests/looptest.c tests it on its own, timers on a clock of its own
through several turns of the wheel: built by CMake, run by the loop
case.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Vladimir Dubrovin 2026-10-24 21:30:13 +03:00
parent fdebdb16dc
commit cc98c498fe
7 changed files with 1240 additions and 2 deletions

View File

@ -390,6 +390,7 @@ endif()
# Source files for 3proxy core # Source files for 3proxy core
set(3PROXY_CORE_SOURCES set(3PROXY_CORE_SOURCES
src/3proxy.c src/3proxy.c
src/loop.c
src/auth.c src/auth.c
src/acl.c src/acl.c
src/limiter.c src/limiter.c
@ -627,6 +628,14 @@ elseif(CMAKE_SYSTEM_NAME STREQUAL "Linux")
target_link_libraries(3proxy PRIVATE dl) target_link_libraries(3proxy PRIVATE dl)
endif() 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 # Build 3proxy_crypt utility
add_executable(3proxy_crypt add_executable(3proxy_crypt
src/3proxy_crypt.c src/3proxy_crypt.c

View File

@ -31,6 +31,9 @@ ftp$(OBJSUFFICS): ftp.c proxy.h structures.h
sockgetchar$(OBJSUFFICS): sockgetchar.c proxy.h structures.h sockgetchar$(OBJSUFFICS): sockgetchar.c proxy.h structures.h
$(CC) $(CFLAGS) sockgetchar.c $(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) proxy$(OBJSUFFICS): proxy.c proxy.h structures.h $(PROXYMAIN)
$(CC) $(CFLAGS) $(DEFINEOPTION)WITHMAIN $(DEFINEOPTION)NOPORTMAP $(DEFINEOPTION)ANONYMOUS $(DEFINEOPTION)NOUDPMAIN proxy.c $(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 pcre$(OBJSUFFICS): pcre.c structures.h
$(CC) $(COUT)pcre$(OBJSUFFICS) $(CFLAGS) $(DEFINEOPTION)WITH_PCRE pcre.c $(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) $(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) 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) $(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)

683
src/loop.c Normal file
View File

@ -0,0 +1,683 @@
/*
3APA3A simplest proxy server
(c) 2002-2026 by Vladimir Dubrovin <vlad@3proxy.org>
please read License Agreement
*/
#include "proxy.h"
#include "loop.h"
#include <limits.h>
#ifdef LOOP_EPOLL
#include <sys/epoll.h>
#include <sys/eventfd.h>
#endif
#ifdef LOOP_KQUEUE
#include <sys/event.h>
#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;
}

118
src/loop.h Normal file
View File

@ -0,0 +1,118 @@
/*
3APA3A simplest proxy server
(c) 2002-2026 by Vladimir Dubrovin <vlad@3proxy.org>
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

View File

@ -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 means to stand on its own should start with one - otherwise an earlier
`allow *` matches first and the rule under test is never reached. `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 ## What is not covered yet
42 of the 112 configuration commands appear in a test, and the count says 42 of the 112 configuration commands appear in a test, and the count says

30
tests/cases/loop.py Normal file
View File

@ -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)

390
tests/looptest.c Normal file
View File

@ -0,0 +1,390 @@
/*
3APA3A simplest proxy server
(c) 2002-2026 by Vladimir Dubrovin <vlad@3proxy.org>
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;
}