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 <noreply@anthropic.com>

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;
 

src/dnspr.c

@@ -179,7 +179,7 @@ static int dnsforward(struct udploop *u, struct udpsession *ss){
 		if(bindwithrange(param, param->remsock, &param->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 *)&param->sinsr, SASIZE(&param->sinsr))

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

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;

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

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],