Event loops: connections accepted on a loop

A service whose clients are served on loops took them in a thread of
its own, woken by the system for each and handing it to a loop: some
5000 switches of threads in 5 seconds under the mix-http benchmark.
Its tcploop() now runs on a fiber of the first loop (loopcall(),
tcploopcall()), its thread waiting for it to return, so connections
are accepted where they are polled for: some 40 switches.

On a loop, a connection taken, the next one is tried at once, until
accept() finds none (EAGAIN, now NEXT_RETRY without a log line), 64 at
most before the others of the loop are let go first (fiberlater(),
which resumes the fiber in the loop's next turn: a poll of a socket
known to be ready does not yield). Waiting for a slot of maxchild does
not block the loop: the slot is tried again every 100 ms.

The listening socket is polled through a dup() of its own: a loop knows
a socket by its number, and on a reload the socket handed over is
polled by the service taking it over too, whose wait would replace this
one's. Not on Windows, where a socket has no dup(), nor with -r.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/mainloop.c

@@ -1262,12 +1262,97 @@ static void loopspawn(struct loopjob *job, struct clientparam *param){
 	looppost(&lp->l, &job->m);
 }
 
+/* what fiberlater() posts: the fiber resumed */
+struct laterjob {
+	struct loopmsg m;
+	struct looper *lp;
+	struct fiber *f;
+};
+
+static void latermsg(struct loop *l, struct loopmsg *m){
+	struct laterjob *job = (struct laterjob *)m;
+
+	fiberrun(job->lp, job->f);
+}
+
+/* The fiber lets what is ready for the others of its loop go first: it is
+   resumed in the loop's next turn. */
+static void fiberlater(void){
+	struct laterjob job;
+
+	if(!curlooper || !fibercurrent()) return;
+	job.m.func = latermsg;
+	job.lp = curlooper;
+	job.f = fibercurrent();
+	looppost(&job.lp->l, &job.m);
+	fiberyield();
+}
+
+/* for tcploopcall(): not on Windows */
+#ifndef _WIN32
+/* what loopcall() posts: func(arg) on a fiber of the first loop */
+struct loopcalljob {
+	struct loopmsg m;
+	void (*func)(void *arg);
+	void *arg;
+	size_t stacksize;
+	int res;
+	_3proxy_sem_t done;
+};
+
+static void loopcallfiber(void *arg){
+	struct loopcalljob *job = (struct loopcalljob *)arg;
+
+	job->func(job->arg);
+	job->res = 0;
+	_3proxy_sem_unlock(job->done);
+}
+
+static void loopcallmsg(struct loop *l, struct loopmsg *m){
+	struct loopcalljob *job = (struct loopcalljob *)m;
+	struct looper *lp = curlooper;
+	struct fiberslot *fs;
+
+	if(!(fs = fibertake(lp, job->stacksize))){
+		_3proxy_sem_unlock(job->done);
+		return;
+	}
+	fiberstart(&fs->f, loopcallfiber, job);
+	fiberrun(lp, &fs->f);
+}
+
+/* func(arg) on a fiber of the first loop, this thread waiting until it
+   returns: 0 once it has, -1 if it could not be run there. */
+static int loopcall(void (*func)(void *arg), void *arg, int stacksize){
+	struct loopcalljob job;
+
+	memset(&job, 0, sizeof(job));
+	if(loopsstart() || _3proxy_sem_init(job.done, 0, 1)) return -1;
+	job.m.func = loopcallmsg;
+	job.func = func;
+	job.arg = arg;
+	job.stacksize = fiberstack(stacksize);
+	job.res = -1;
+	looppost(&loopers[0].l, &job.m);
+	_3proxy_sem_lock(job.done);
+	_3proxy_sem_destroy(job.done);
+	return job.res;
+}
+#endif
+
 #else
 
 struct loopjob;
+#ifndef _WIN32
+static int loopcall(void (*func)(void *arg), void *arg, int stacksize){
+	return -1;
+}
+#endif
 static int onlooper(void){
 	return 0;
 }
+static void fiberlater(void){
+}
 static int loopwanted(struct srvparam *srv){
 	return 0;
 }

src/maintcp.c

@@ -104,7 +104,9 @@ static int tcpnext(struct srvstart *s, SOCKET *client, int *nlog){
  else {
 	new_sock = srv->so._accept(srv->so.state, srv->srvsock, (struct sockaddr*)&defparam->sincr, &size);
 	if(new_sock == INVALID_SOCKET){
+		/* on a loop, accepted until there are none */
 #ifdef _WIN32
+		if(WSAGetLastError() == WSAEWOULDBLOCK) return NEXT_RETRY;
 		switch(WSAGetLastError()){
 		case WSAEMFILE:
 		case WSAENOBUFS:
@@ -119,6 +121,7 @@ static int tcpnext(struct srvstart *s, SOCKET *client, int *nlog){
 		}
 
 #else
+		if(errno == EAGAIN || errno == EWOULDBLOCK) return NEXT_RETRY;
 		switch (errno){
 #ifdef EMFILE
 		case EMFILE:
@@ -187,26 +190,38 @@ static int tcploop(struct srvstart *s, int *haveslot){
  unsigned char buf[256];
  /* -r: a TCP service connects out for each client instead of listening */
  int connectback = s->iscbc;
+ /* on a fiber of a loop: the loop is not to block, waiting for a slot; a
+    client accepted, the next one is tried before polling */
+ int onloop = onlooper(), more = 0, inrow = 0;
+ /* polled there through a descriptor of its own: a loop knows a socket by
+    its number, which the listening one, handed over on a reload, would
+    share with the service taking it over */
+ struct pollfd lfd = srv->fds;
 
+#ifndef _WIN32
+ if(onloop && (lfd.fd = dup(srv->srvsock)) < 0) return -1;
+#endif
  for (;;) {
 	for(;;){
 		/* A connection is only taken with a slot of maxchild in hand, which
 		   goes to its child; a slot not handed on is kept for the next one.
 		   The wait is timed to notice a reload. */
 		while(!(*haveslot) && srv->config->paused == srv->paused){
-			if(!_3proxy_sem_timedlock(srv->childslots, 1000)){
+			if(!_3proxy_sem_timedlock(srv->childslots, onloop? 0 : 1000)){
 				(*haveslot) = 1;
 				nfull = 0;
 				break;
 			}
-			if(!srv->silent && !(nfull++ % 5)) {
+			if(onloop) sockpoll(&srv->so, srv->so.state, NULL, 0, 100);
+			if(!srv->silent && !(nfull++ % (onloop? 50 : 5))) {
 				sprintf((char *)buf, "Warning: too many connected clients (%d/%d)", srv->childcount, srv->maxchild);
 				dolog(defparam, buf);
 			}
 		}
 		if (connectback) break;
 		if (srv->config->paused != srv->paused) break;
-		error = sockpoll(&srv->so, srv->so.state, &srv->fds, 1, 1000);
+		if (more) break;
+		error = sockpoll(&srv->so, srv->so.state, onloop? &lfd : &srv->fds, 1, 1000);
 		if (error >= 1) break;
 		if (error == 0) continue;
 		if (errno != EAGAIN &&	errno != EINTR) {
@@ -219,9 +234,56 @@ static int tcploop(struct srvstart *s, int *haveslot){
 	error = 0;
 	next = tcpnext(s, &new_sock, &nlog);
 	if(next == NEXT_STOP) break;
+	/* not too many in a row: the clients given to this loop wait for it */
+	more = onloop && next == NEXT_CLIENT;
+	if(more && ++inrow >= 64){
+		fiberlater();
+		inrow = 0;
+	}
+	if(!more) inrow = 0;
 	if(next == NEXT_RETRY) continue;
 	if(!srvspawn(s, new_sock)) (*haveslot) = 0;
  }
- return 0;
+#ifndef _WIN32
+ if(onloop){
+	if(fiberclosefunc) fiberclosefunc(lfd.fd);
+	close(lfd.fd);
+ }
+#endif
+ return 0;
+}
+
+#ifndef _WIN32
+struct tcploopcall {
+	struct srvstart *s;
+	int *haveslot;
+	int res;
+};
+
+static void tcploopfiber(void *arg){
+	struct tcploopcall *c = (struct tcploopcall *)arg;
+
+	c->res = tcploop(c->s, c->haveslot);
+}
+#endif
+
+/* tcploop() on a fiber of the first loop, for a service whose clients are
+   served on loops: accepting there, not in a thread of its own woken for
+   each client. 0 once it has run, *res what it returned. Not on Windows,
+   where a listening socket has no dup() to be polled through. */
+static int tcploopcall(struct srvstart *s, int *haveslot, int *res){
+#ifndef _WIN32
+	struct tcploopcall c;
+
+	if(s->iscbc || !loopwanted(&s->srv)) return -1;
+	c.s = s;
+	c.haveslot = haveslot;
+	c.res = 0;
+	if(loopcall(tcploopfiber, &c, s->srv.stacksize)) return -1;
+	*res = c.res;
+	return 0;
+#else
+	return -1;
+#endif
 }
 #endif

src/proxymain.c

@@ -164,7 +164,7 @@ static int srvrun(struct srvstart *s){
  else
 #endif
 #ifndef NOTCPMAIN
- res = tcploop(s, &haveslot);
+ if(tcploopcall(s, &haveslot, &res)) res = tcploop(s, &haveslot);
 #endif
  _3proxy_mutex_lock(&tpool.mutex);
  tpool.services--;