From 57818ad497d4bbbc761918ac1e86e17e101a9d16 Mon Sep 17 00:00:00 2001 From: Vladimir Dubrovin <3proxy@3proxy.ru> Date: Thu, 22 Oct 2026 23:53:06 +0300 Subject: [PATCH] UDP loop: deadlines in a heap, on a monotonic clock The loop went through every session three times a turn: for the nearest deadline, to register its sockets with epoll, and for the deadlines come and the sessions ended. A datagram cost as much as the sessions there were, idle ones too. - The deadlines are in a binary heap, the nearest at its top: the wait is until the top, and the sessions timed out are taken from the top down to the first not due. A datagram only moves a session's deadline later, which leaves the heap as it is: the session is looked at again when its old deadline comes, and put back where its deadline is then, once in its timeout however many datagrams it had. A deadline set earlier by a handler goes up the heap. - A session ended is put on a list and let go at the end of the turn, and its sockets are registered with epoll after the loop or a handler had it, not every session every turn. - Deadlines are milliseconds of monotime(), CLOCK_MONOTONIC (the performance counter on Windows), which the time of day being set does not move: before, a clock set back kept sessions for as long, and one set forward ended them all. The loop reads it once a turn, in u->now. With epoll, a datagram costs the same with 20000 idle sessions as with none (38-40 us of the proxy's CPU a round trip, ARM), where it took 59 before. poll() elsewhere still builds its set every turn. Tests: a session heard from every half second lives through two seconds of idle timeout, and ends two seconds after the last datagram. Co-Authored-By: Claude Opus 5.5 --- src/common.c | 24 +++++ src/dnspr.c | 2 +- src/mainudp.c | 238 +++++++++++++++++++++++++++++++++-------- src/proxy.h | 15 ++- src/proxymain.c | 12 ++- tests/cases/udploop.py | 23 ++++ 6 files changed, 266 insertions(+), 48 deletions(-) diff --git a/src/common.c b/src/common.c index 6dfaa9f..59228e7 100644 --- a/src/common.c +++ b/src/common.c @@ -866,6 +866,30 @@ int scanaddr(const unsigned char *s, uint32_t * ip, uint32_t * mask) { /* The default resolver: name servers, configured or the system's, or NULL for the system routines. resolvmode is the configuration's, for names resolved outside a service; each service has its own. */ +/* Milliseconds on a clock which only goes forward, at the same pace as the + time of day but not moved with it: a deadline does not come at once, or + never, when the clock of the system is set. */ +uint64_t monotime(void){ +#ifdef _WIN32 + LARGE_INTEGER c, f; + + QueryPerformanceFrequency(&f); + QueryPerformanceCounter(&c); + return (uint64_t)(c.QuadPart / f.QuadPart) * 1000 + + (uint64_t)(c.QuadPart % f.QuadPart) * 1000 / (uint64_t)f.QuadPart; +#elif defined(CLOCK_MONOTONIC) + struct timespec ts; + + clock_gettime(CLOCK_MONOTONIC, &ts); + return (uint64_t)ts.tv_sec * 1000 + (uint64_t)ts.tv_nsec / 1000000; +#else + struct timeval tv; + + gettimeofday(&tv, NULL); + return (uint64_t)tv.tv_sec * 1000 + (uint64_t)tv.tv_usec / 1000; +#endif +} + RESOLVFUNC resolvfunc = NULL; int resolvmode = RESOLVE_DEFAULT; diff --git a/src/dnspr.c b/src/dnspr.c index 8171f10..418655b 100644 --- a/src/dnspr.c +++ b/src/dnspr.c @@ -179,7 +179,7 @@ static int dnsforward(struct udploop *u, struct udpsession *ss){ if(bindwithrange(param, param->remsock, ¶m->sinsl, param->extport)) continue; param->sinsr = ns[q->server].addr; ss->idle = 0; - ss->deadline = time(NULL) + param->srv->timeouts[DNS_TO]; + ss->deadline = u->now + (uint64_t)param->srv->timeouts[DNS_TO] * 1000; if(q->tcp){ /* connected, then written to, as the loop sees it writable */ if(param->srv->so._connect(param->sostate, param->remsock, (struct sockaddr *)¶m->sinsr, SASIZE(¶m->sinsr)) diff --git a/src/mainudp.c b/src/mainudp.c index 5a91202..d7b68f1 100644 --- a/src/mainudp.c +++ b/src/mainudp.c @@ -38,6 +38,11 @@ struct udpstate { #endif int nalloc; int nfull; + /* the sessions with a deadline, the nearest first (a binary heap) */ + struct udpsession **heap; + int nheap, hsize; + /* the sessions ended, to be let go once the loop is through with them */ + struct udpsession *dead; }; #ifdef UDPEPOLL @@ -150,6 +155,7 @@ static struct udpsession *udpnew(struct udpstate *st){ ss->param = param; ss->svc = st->svc; ss->regsock = ss->regctrl = INVALID_SOCKET; + ss->hidx = -1; #ifdef MODULEMAINFUNC if(makefilters(srv, param) > CONTINUE){ freeparam(param); @@ -200,8 +206,154 @@ static void udpremove(struct udpstate *st, struct udpsession *ss){ if(ss->keyed) hashdelete(&st->keys, &ss->param->sincr); } -static void udprefresh(struct udpsession *ss, time_t now){ - if(ss->idle) ss->deadline = now + ss->idle; +/* Its client or its destination was heard from: the deadline moves on, + where the heap does not see it yet (udpafter(), udpexpire()). */ +static void udprefresh(struct udpsession *ss, uint64_t now){ + if(ss->idle) ss->deadline = now + (uint64_t)ss->idle * 1000; +} + +/* The heap of the deadlines: at its top the nearest, so the loop learns + how long it may wait, and which have come, without going through every + session. */ +static void udpheapset(struct udpstate *st, int i, struct udpsession *ss){ + st->heap[i] = ss; + ss->hidx = i; +} + +static void udpheapup(struct udpstate *st, int i){ + struct udpsession *ss = st->heap[i]; + + while(i > 0){ + int p = (i - 1) / 2; + + if(st->heap[p]->hkey <= ss->hkey) break; + udpheapset(st, i, st->heap[p]); + i = p; + } + udpheapset(st, i, ss); +} + +static void udpheapdown(struct udpstate *st, int i){ + struct udpsession *ss = st->heap[i]; + + for(;;){ + int c = 2 * i + 1; + + if(c >= st->nheap) break; + if(c + 1 < st->nheap && st->heap[c+1]->hkey < st->heap[c]->hkey) c++; + if(ss->hkey <= st->heap[c]->hkey) break; + udpheapset(st, i, st->heap[c]); + i = c; + } + udpheapset(st, i, ss); +} + +/* 0, or 21 without memory */ +static int udpheapadd(struct udpstate *st, struct udpsession *ss){ + if(st->nheap == st->hsize){ + int want = st->hsize? 2 * st->hsize : 64; + struct udpsession **h = realloc(st->heap, want * sizeof(*h)); + + if(!h) return 21; + st->heap = h; + st->hsize = want; + } + udpheapset(st, st->nheap++, ss); + udpheapup(st, ss->hidx); + return 0; +} + +static void udpheapdel(struct udpstate *st, struct udpsession *ss){ + int i = ss->hidx; + struct udpsession *last; + + if(i < 0) return; + ss->hidx = -1; + last = st->heap[--st->nheap]; + if(last == ss) return; + udpheapset(st, i, last); + if(i > 0 && st->heap[(i - 1) / 2]->hkey > last->hkey) udpheapup(st, i); + else udpheapdown(st, i); +} + +/* After the loop or a handler had the session: if it ends (end, or done), + on the list of those to let go; otherwise its sockets watched as they + are now and its deadline in the heap. A deadline moved later stays where + it was in the heap, only one moved earlier goes up. */ +static void udpafter(struct udpstate *st, struct udpsession *ss, int end){ + if(end || ss->done){ + /* done 2: on the list already */ + if(ss->done != 2){ + ss->done = 2; + ss->dnext = st->dead; + st->dead = ss; + } + return; + } +#ifdef UDPEPOLL + udpsync(st, ss); +#endif + /* none: if in the heap, it leaves once its old one comes */ + if(!ss->deadline) return; + if(ss->hidx < 0){ + ss->hkey = ss->deadline; + if(udpheapadd(st, ss)){ + /* it would never time out */ + ss->param->res = 21; + udpafter(st, ss, 1); + } + } + else if(ss->deadline < ss->hkey){ + ss->hkey = ss->deadline; + udpheapup(st, ss->hidx); + } +} + +/* The sessions whose deadline has come, from the top of the heap down to + the first not yet due. One refreshed since it went in is put back where + its deadline is now: each session costs a step of the heap once in its + timeout, however many datagrams it has. */ +static void udpexpire(struct udpstate *st){ + struct udpsession *ss; + uint64_t now = st->u.now; + + while(st->nheap && (ss = st->heap[0])->hkey <= now){ + if(ss->done || !ss->deadline){ + udpheapdel(st, ss); + continue; + } + if(ss->deadline > now){ + ss->hkey = ss->deadline; + udpheapdown(st, 0); + continue; + } + if(!st->svc->timeout || (*st->svc->timeout)(&st->u, ss)){ + if(!st->svc->timeout) ss->param->res = st->svc->timeoutres; + udpheapdel(st, ss); + udpafter(st, ss, 1); + continue; + } + /* it goes on: at its new deadline, or if it set none later, at the + next turn of the loop, not again in this one */ + if(ss->deadline && ss->deadline <= now) ss->deadline = now + 1; + udpafter(st, ss, 0); + } +} + +/* The sessions ended in this turn of the loop let go. */ +static void udpreap(struct udpstate *st){ + struct udpsession *ss; + + while((ss = st->dead)){ + st->dead = ss->dnext; + ss->dnext = NULL; + udpheapdel(st, ss); + udpremove(st, ss); +#ifdef UDPEPOLL + udpunwatch(st, ss); +#endif + udpend(ss); + } } #ifndef STDMAIN @@ -217,8 +369,11 @@ static void udphandover(struct udpstate *st){ #ifdef UDPEPOLL udpunwatch(st, ss); #endif + /* the deadline, on the same clock, goes along; the heap does not */ + ss->hidx = -1; last = ss; } + st->nheap = 0; if(!last) return; if(!l){ udpendall(st->head, 26); @@ -254,9 +409,14 @@ static void udpadopt(struct udpstate *st){ ss->next = ss->prev = NULL; /* watched by the old service's loop, not by this one yet */ ss->regsock = ss->regctrl = INVALID_SOCKET; + ss->hidx = -1; param = ss->param; old = param->srv; - if(old == srv) { udpadd(st, ss); continue; } + if(old == srv) { + udpadd(st, ss); + udpafter(st, ss, 0); + continue; + } if(param->service != srv->service){ param->res = 26; udpend(ss); @@ -303,6 +463,7 @@ static void udpadopt(struct udpstate *st){ continue; } udpadd(st, ss); + udpafter(st, ss, 0); } } #endif @@ -326,7 +487,7 @@ static int udpolder(struct srvparam *srv){ /* The datagrams waiting on the service's socket: to a session, or a new one. Some at a time, so that the sessions get their turn. */ -static void udpclients(struct udpstate *st, time_t now){ +static void udpclients(struct udpstate *st){ struct srvparam *srv = st->u.srv; struct clientparam *defparam = st->u.defparam; struct udpsession *ss; @@ -350,23 +511,26 @@ static void udpclients(struct udpstate *st, time_t now){ !memcmp(SAADDR(&ss->param->sincr), SAADDR(&defparam->sincr), SAADDRLEN(&defparam->sincr)) && *SAPORT(&ss->param->sincr) == *SAPORT(&defparam->sincr)){ if(ss->done) continue; - udprefresh(ss, now); - if(st->svc->client && (*st->svc->client)(&st->u, ss, srv->udpbuf, len)) ss->done = 1; + udprefresh(ss, st->u.now); + udpafter(st, ss, st->svc->client && (*st->svc->client)(&st->u, ss, srv->udpbuf, len)); continue; } if(!(ss = udpnew(st))) continue; ss->param->sincr = defparam->sincr; udpadd(st, ss); - if((*st->svc->first)(&st->u, ss, srv->udpbuf, len)) ss->done = 1; - else if(!ss->deadline || ss->idle) udprefresh(ss, now); + if((*st->svc->first)(&st->u, ss, srv->udpbuf, len)) { + udpafter(st, ss, 1); + continue; + } + if(!ss->deadline || ss->idle) udprefresh(ss, st->u.now); + udpafter(st, ss, 0); } } static int udploop(struct srvstart *s){ struct udpstate st; struct srvparam *srv = &s->srv; - struct udpsession *ss, *next; - time_t now; + struct udpsession *ss; int i, n, timeo, res = 0, wait = 0, srvready; #ifndef UDPEPOLL int nfds; @@ -412,29 +576,31 @@ static int udploop(struct srvstart *s){ #endif st.nalloc = want; } - now = time(NULL); + st.u.now = monotime(); timeo = 1000; #ifndef STDMAIN /* waiting for an older service's sessions: soon looked for again */ if((wait = udpolder(srv))) timeo = 20; #endif + /* until the nearest deadline at most */ + if(st.nheap){ + uint64_t first = st.heap[0]->hkey; + + if(first <= st.u.now) timeo = 0; + else if(first - st.u.now < (uint64_t)timeo) timeo = (int)(first - st.u.now); + } srvready = 0; #ifdef UDPEPOLL if(st.srvevents != (wait? 0 : POLLIN)){ st.srvevents = wait? 0 : POLLIN; udpwatch(&st, EPOLL_CTL_MOD, srv->srvsock, st.srvevents, NULL); } - for(ss = st.head; ss; ss = ss->next){ - if(ss->deadline && (ss->deadline - now) * 1000 < timeo) - timeo = ss->deadline > now? (int)(ss->deadline - now) * 1000 : 0; - udpsync(&st, ss); - } n = epoll_wait(st.ep, st.evs, st.nalloc, timeo); if(n < 0 && errno != EINTR){ res = 91; break; } - now = time(NULL); + st.u.now = monotime(); for(i = 0; i < n; i++){ uintptr_t tag = (uintptr_t)st.evs[i].data.ptr; @@ -445,10 +611,10 @@ static int udploop(struct srvstart *s){ ss = (struct udpsession *)(tag & ~(uintptr_t)1); if(ss->done) continue; /* a parent's control connection: its end ends the session */ - if(tag & 1) ss->done = 1; + if(tag & 1) udpafter(&st, ss, 1); else { - udprefresh(ss, now); - if((*st.svc->server)(&st.u, ss)) ss->done = 1; + udprefresh(ss, st.u.now); + udpafter(&st, ss, (*st.svc->server)(&st.u, ss)); } } #else @@ -458,8 +624,6 @@ static int udploop(struct srvstart *s){ st.fds[nfds].revents = 0; st.of[nfds++] = NULL; for(ss = st.head; ss; ss = ss->next){ - if(ss->deadline && (ss->deadline - now) * 1000 < timeo) - timeo = ss->deadline > now? (int)(ss->deadline - now) * 1000 : 0; if(ss->param->remsock != INVALID_SOCKET){ st.fds[nfds].fd = ss->param->remsock; st.fds[nfds].events = ss->events? ss->events : POLLIN; @@ -479,41 +643,27 @@ static int udploop(struct srvstart *s){ res = 91; break; } - now = time(NULL); + st.u.now = monotime(); for(i = 1; n > 0 && i < nfds; i++){ ss = st.of[i]; if(!st.fds[i].revents || ss->done) continue; if(st.fds[i].fd == ss->param->remsock){ - udprefresh(ss, now); - if((*st.svc->server)(&st.u, ss)) ss->done = 1; + udprefresh(ss, st.u.now); + udpafter(&st, ss, (*st.svc->server)(&st.u, ss)); } - else ss->done = 1; + else udpafter(&st, ss, 1); } srvready = n > 0 && st.fds[0].revents; #endif - if(srvready && !wait) udpclients(&st, now); - for(ss = st.head; ss; ss = next){ - next = ss->next; - if(!ss->done && ss->deadline && ss->deadline <= now){ - if(st.svc->timeout) ss->done = (*st.svc->timeout)(&st.u, ss); - else { - ss->done = 1; - ss->param->res = st.svc->timeoutres; - } - } - if(ss->done){ - udpremove(&st, ss); -#ifdef UDPEPOLL - udpunwatch(&st, ss); -#endif - udpend(ss); - } - } + if(srvready && !wait) udpclients(&st); + udpexpire(&st); + udpreap(&st); } #ifndef STDMAIN if(srv->config->paused != srv->paused) udphandover(&st); #endif udpendall(st.head, 26); + free(st.heap); if(st.svc->keyed) destroyhashtable(&st.keys); #ifdef UDPEPOLL close(st.ep); diff --git a/src/proxy.h b/src/proxy.h index 1359833..0da1a67 100644 --- a/src/proxy.h +++ b/src/proxy.h @@ -249,8 +249,17 @@ struct udpsession { struct udpsession *next, *prev; struct clientparam *param; const struct udpservice *svc; - time_t deadline; /* when it ends unless refreshed */ + /* when it ends unless refreshed, in ms of monotime() (u->now), 0: + never; a handler may set it, later or earlier */ + uint64_t deadline; int idle; /* seconds of silence it lasts, 0: deadline fixed */ + /* the loop's timers: a heap ordered by hkey, where it is at hidx (-1: + not in it). hkey is the deadline as it was when it last went in or + up: a later deadline, as each datagram sets, leaves it where it is, + to be looked at again only once its old one comes. */ + uint64_t hkey; + int hidx; + struct udpsession *dnext; /* the loop's list of those ended */ int keyed; /* found by its client's address */ int done; /* to be ended, param->res saying how */ short events; /* polled for on remsock, 0: POLLIN */ @@ -285,8 +294,12 @@ struct udploop { struct clientparam *defparam; /* scratch of UDPBUFSIZE for a handler */ unsigned char *buf; + /* monotime() as the loop last read it, for the deadlines */ + uint64_t now; }; /* an answer to the session's client, from the service's socket */ +/* ms on a clock only going forward, whatever is done to the time of day */ +uint64_t monotime(void); int udpanswer(struct udploop *u, struct clientparam *param, unsigned char *buf, int len); extern const struct udpservice udppmservice, dnsprservice; diff --git a/src/proxymain.c b/src/proxymain.c index 000b849..a019f53 100644 --- a/src/proxymain.c +++ b/src/proxymain.c @@ -85,11 +85,19 @@ static void udpend(struct udpsession *ss); static void udpendall(struct udpsession *list, int res); static void udpadd(struct udpstate *st, struct udpsession *ss); static void udpremove(struct udpstate *st, struct udpsession *ss); -static void udprefresh(struct udpsession *ss, time_t now); +static void udprefresh(struct udpsession *ss, uint64_t now); +static void udpheapset(struct udpstate *st, int i, struct udpsession *ss); +static void udpheapup(struct udpstate *st, int i); +static void udpheapdown(struct udpstate *st, int i); +static int udpheapadd(struct udpstate *st, struct udpsession *ss); +static void udpheapdel(struct udpstate *st, struct udpsession *ss); +static void udpafter(struct udpstate *st, struct udpsession *ss, int end); +static void udpexpire(struct udpstate *st); +static void udpreap(struct udpstate *st); static void udphandover(struct udpstate *st); static void udpadopt(struct udpstate *st); static int udpolder(struct srvparam *srv); -static void udpclients(struct udpstate *st, time_t now); +static void udpclients(struct udpstate *st); static int udploop(struct srvstart *s); #ifndef NOTCPMAIN static int tcploop(struct srvstart *s, int *haveslot); diff --git a/tests/cases/udploop.py b/tests/cases/udploop.py index ebd15b1..2e50cd2 100644 --- a/tests/cases/udploop.py +++ b/tests/cases/udploop.py @@ -172,6 +172,29 @@ def run(t): t.fail("two seconds of silence end it, logged with 92", "a 00092 line", "none") s.close() + # --- a session heard from lives on past its first deadline, and ends + # its idle time after the last datagram, neither sooner nor much later + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + answered = 0 + for i in range(10): + if exchange(s, idle, f"k{i}".encode()) == f"echo:k{i}".encode(): + answered += 1 + time.sleep(0.5) + cport = s.getsockname()[1] + last = time.time() + t.eq(10, answered, "a datagram every half second for five seconds is answered") + t.ok("and the session lives on through them") if not idle_ended() else \ + t.fail("and the session lives on through them", "no 00092 line", "ended") + while time.time() < last + 6 and not idle_ended(): + time.sleep(0.1) + took = time.time() - last + if idle_ended() and 1.5 <= took <= 3.5: + t.ok("then two seconds of silence end it") + else: + t.fail("then two seconds of silence end it", "ended 1.5-3.5 s after the last", + f"{'ended' if idle_ended() else 'not ended'} after {took:.1f} s") + s.close() + # --- maxconn: a client without a free slot is not answered ------------- held = [socket.socket(socket.AF_INET, socket.SOCK_DGRAM) for _ in range(2)] t.eq([b"echo:a", b"echo:a"], [exchange(h, limited, b"a") for h in held],