Hand listening sockets over on a reload

A service's TCP listening socket is kept as a listener: when a reload
commits, before the old services are told to stop, each new service
takes over the listener of an older configuration's service on the same
address, port, network namespace and device with the same -ol options.
It skips socket() and bind(), calling only listen() again for its own
backlog, and logs "Accepting connections, taken over". The old service
stops accepting at its next poll and lets go of the socket; whoever
lets go last closes it, unlinking a Unix socket's path.

So a service staying where it was no longer leaves a gap: before, the
old one closed its socket and the new one opened another, and a
connection queued on the old socket was reset. UDP services, whose
clients read from the service's socket, and two services of the same
configuration on one address keep their own sockets as before.

A test case makes requests nonstop through three reloads and wants none
refused or reset; against the build before this, connections were
reset. The reload case no longer waits out the switch. SIGUSR1 in
3proxy(8) describes how a reload behaves now.

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

doc/html/man8/3proxy.8.html

@@ -154,7 +154,16 @@ connections</p>
 
 <p style="margin-left:9%;"><b>SIGUSR1</b></p>
 
-<p style="margin-left:18%;">reload configuration</p>
+<p style="margin-left:18%;">reload configuration. The file is read whole before anything changes: if
+it has an error, the configuration in force stays as it is, its services
+running. Otherwise the new services start and the old ones stop accepting
+within a second; connections already open finish under the old
+configuration. A new service listening on the same address and port, in
+the same network namespace and on the same device, with the same <b>-ol</b>
+options, takes over the old service's TCP listening socket, so no
+connection is refused or lost meanwhile; any other listening socket is
+closed by the old service and opened by the new one. Settings left out of
+the new file go back to their defaults.</p>
 
 <p style="margin-left:9%; margin-top: 1em">Under Windows,
 if <b>3proxy</b> is installed as a service you can use

man/3proxy.8

@@ -115,7 +115,16 @@ configuration
 start to accept new connections
 .TP
 .B SIGUSR1
-reload configuration
+reload configuration. The file is read whole before anything changes: if
+it has an error, the configuration in force stays as it is, its services
+running. Otherwise the new services start and the old ones stop accepting
+within a second; connections already open finish under the old
+configuration. A new service listening on the same address and port, in
+the same network namespace and on the same device, with the same -ol
+options, takes over the old service's TCP listening socket, so no
+connection is refused or lost meanwhile; any other listening socket is
+closed by the old service and opened by the new one. Settings left out of
+the new file go back to their defaults.
 .PP
 Under Windows, if
 .BR 3proxy

src/3proxy.c

@@ -557,6 +557,7 @@ int WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, LPWSTR lpCmdLine, int
 #endif
   _3proxy_mutex_init(&config_mutex);
   _3proxy_mutex_init(&confref_mutex);
+  _3proxy_mutex_init(&listener_mutex);
   _3proxy_mutex_init(&runningservices_mutex);
   _3proxy_mutex_init(&bandlim_mutex);
   _3proxy_mutex_init(&connlim_mutex);

src/conf.c

@@ -86,6 +86,7 @@ struct trafvalue * trafvalue(unsigned number){
 int mainprepare(int argc, char** argv, void **service);
 int mainrun(void *service);
 void maindrop(void *service);
+void mainclaim(void *service);
 
 struct proxydef childdef = {NULL, 0, 0, S_NOSERVICE, ""};
 
@@ -2638,7 +2639,9 @@ int reload (void){
 	runconf = new;
 	_3proxy_mutex_unlock(&confref_mutex);
 	resolvfunc = new->resolvfunc;
-	/* the old services stop accepting, the new ones start */
+	/* the new services take over the listening sockets of the old ones on
+	   the same addresses, which then stop accepting, and start */
+	for(d = deferred; d; d = d->next) mainclaim(d->service);
 	old->paused++;
 	while((d = deferred)){
 		deferred = d->next;

src/proxy.h

@@ -226,7 +226,7 @@ extern int timetoexit, needreload;
 extern time_t curtime;		/* the clock, kept by the main thread */
 
 extern struct extparam *conf, *runconf;
-extern _3proxy_mutex_t confref_mutex;
+extern _3proxy_mutex_t confref_mutex, listener_mutex;
 struct extparam * confget(void);
 void confhold(struct extparam *c);
 void confput(struct extparam *c);

src/proxymain.c

@@ -870,6 +870,29 @@ static int srvprepare(struct srvstart *s, int argc, char **argv){
 
 /* The listening socket. A UDP one also receives the datagrams of new
    clients, so it is their client socket too. 0, or the code to exit with. */
+#ifndef STDMAIN
+/* A listening socket: it outlives the service that opened it when, on a
+   reload, a service of the new configuration listening on the same
+   address, port, network namespace and device with the same options takes
+   it over. The old service stops accepting once its configuration is no
+   longer in force; whoever lets go of the socket last closes it. TCP only:
+   a UDP service's clients read from its socket. */
+struct listener {
+	struct listener *next;
+	SOCKET sock;
+	PROXYSOCKADDRTYPE sa;
+	char *netns, *device;
+	struct sockoptset opts;
+	struct srvparam *owner;	/* the service accepting on it */
+	int users;		/* the services holding it */
+};
+static struct listener *listeners = NULL;
+_3proxy_mutex_t listener_mutex;
+
+static int listenerkeep(struct srvparam *srv, SOCKET sock);
+static void listenerrelease(struct srvparam *srv);
+#endif
+
 static int srvlisten(struct srvstart *s){
  struct srvparam *srv = &s->srv;
  struct clientparam *defparam = &s->defparam;
@@ -880,6 +903,26 @@ static int srvlisten(struct srvstart *s){
  unsigned long ul = 1;
 #endif
 
+#ifndef STDMAIN
+ if(srv->listener){
+	/* taken over from a service of the configuration before: listening
+	   already, only the backlog may be new */
+	sock = srv->srvsock = srv->listener->sock;
+	if(srv->so._listen (srv->so.state, sock, srv->backlog?srv->backlog : 1+(srv->maxchild>>3))==-1) {
+		sprintf((char *)buf, "listen(): %s", strerror(errno));
+		if(!srv->silent)dolog(defparam, buf);
+		return -4;
+	}
+	if(!srv->silent){
+		sprintf((char *)buf, "Accepting connections, taken over [%"PRIu64"/%"PRIu64"]", (uint64_t)getpid(), (uint64_t)pthread_self());
+		dolog(defparam, buf);
+	}
+	srv->fds.fd = sock;
+	srv->fds.events = POLLIN;
+	return 0;
+ }
+#endif
+
 #ifndef NOTCPMAIN
  if(!s->isudp){
 	sock=nssocket(srv, &srv->so, srv->so.state, NS_IN, SASOCK(&srv->intsa), SOCK_STREAM,
@@ -931,6 +974,11 @@ static int srvlisten(struct srvstart *s){
 		if(!srv->silent)dolog(defparam, buf);
 		return -4;
 	}
+#ifndef STDMAIN
+	/* kept for a service of a later configuration to take over; without
+	   memory for it the service closes the socket itself */
+	listenerkeep(srv, sock);
+#endif
  }
 #endif
 #ifndef NOUDPMAIN
@@ -1468,6 +1516,94 @@ int mainrun(void *service){
  return res;
 }
 
+static int samestr(const char *a, const char *b){
+ return (!a && !b) || (a && b && !strcmp(a, b));
+}
+
+static int samelistener(struct listener *l, struct srvparam *srv){
+#ifdef __linux__
+ if(!samestr(l->netns, srv->inetns)) return 0;
+#endif
+#if defined SO_BINDTODEVICE || defined IP_BOUND_IF
+ if(!samestr(l->device, srv->ibindtodevice)) return 0;
+#endif
+ return SASIZE(&l->sa) == SASIZE(&srv->intsa) &&
+	!memcmp(&l->sa, &srv->intsa, SASIZE(&srv->intsa)) &&
+	!memcmp(&l->opts, &srv->lissockopts, sizeof(l->opts));
+}
+
+/* On a reload, before the old services are told to stop: a prepared
+   service takes over the listening socket of an older configuration's
+   service on the same address, if there is one. */
+void mainclaim(void *service){
+ struct srvstart *s = (struct srvstart *)service;
+ struct srvparam *srv = &s->srv;
+ struct listener *l;
+
+ if(s->isudp || s->iscbc) return;
+ _3proxy_mutex_lock(&listener_mutex);
+ for(l = listeners; l; l = l->next){
+	if(l->owner && l->owner->config != srv->config && samelistener(l, srv)){
+		l->owner = srv;
+		l->users++;
+		srv->listener = l;
+		break;
+	}
+ }
+ _3proxy_mutex_unlock(&listener_mutex);
+}
+
+/* A socket a service opened and bound itself, kept for whoever may take it
+   over; 0, or 1 for want of memory, when the service closes it itself. */
+static int listenerkeep(struct srvparam *srv, SOCKET sock){
+ struct listener *l;
+
+ if(!(l = malloc(sizeof(*l)))) return 1;
+ memset(l, 0, sizeof(*l));
+ l->sock = sock;
+ l->sa = srv->intsa;
+ l->opts = srv->lissockopts;
+#ifdef __linux__
+ if(srv->inetns && !(l->netns = strdup(srv->inetns))) { free(l); return 1; }
+#endif
+#if defined SO_BINDTODEVICE || defined IP_BOUND_IF
+ if(srv->ibindtodevice && !(l->device = strdup(srv->ibindtodevice))) { free(l->netns); free(l); return 1; }
+#endif
+ l->owner = srv;
+ l->users = 1;
+ _3proxy_mutex_lock(&listener_mutex);
+ l->next = listeners;
+ listeners = l;
+ _3proxy_mutex_unlock(&listener_mutex);
+ srv->listener = l;
+ return 0;
+}
+
+/* A service lets go of its listening socket: closed if nobody else holds
+   it. */
+static void listenerrelease(struct srvparam *srv){
+ struct listener *l = srv->listener, **lp;
+ int last;
+
+ srv->listener = NULL;
+ _3proxy_mutex_lock(&listener_mutex);
+ if(l->owner == srv) l->owner = NULL;
+ last = !--l->users;
+ if(last) for(lp = &listeners; *lp; lp = &(*lp)->next) if(*lp == l) {
+	*lp = l->next;
+	break;
+ }
+ _3proxy_mutex_unlock(&listener_mutex);
+ if(!last) return;
+ so._closesocket(so.state, l->sock);
+#ifdef WITH_UN
+ if(*SAFAMILY(&l->sa) == AF_UNIX && *SAADDR(&l->sa))unlink((char *)SAADDR(&l->sa));
+#endif
+ if(l->netns) free(l->netns);
+ if(l->device) free(l->device);
+ free(l);
+}
+
 /* A prepared service which could not be started. */
 void maindrop(void *service){
  struct srvstart *s = (struct srvstart *)service;
@@ -1580,6 +1716,14 @@ void srvinit2(struct srvparam * srv, struct clientparam *param){
 void srvfree(struct srvparam * srv){
  int i;
 
+#ifndef STDMAIN
+ /* a listening socket kept as a listener is closed by its last holder,
+    which may be a service of a later configuration */
+ if(srv->listener){
+	srv->srvsock = INVALID_SOCKET;
+	listenerrelease(srv);
+ }
+#endif
  if(srv->srvsock != INVALID_SOCKET) {
     so._closesocket(srv->so.state, srv->srvsock);
 #ifdef WITH_UN

src/structures.h

@@ -240,6 +240,7 @@ struct node;
 struct symbol;
 struct pluginlink;
 struct srvparam;
+struct listener;
 
 typedef void (*LOGFUNC)(struct clientparam * param, const unsigned char *);
 typedef int (*AUTHFUNC)(struct clientparam * param);
@@ -619,6 +620,8 @@ struct srvparam {
 	struct sockfuncs so;
 	struct srvparam *next;
 	struct srvparam *prev;
+	struct listener *listener;	/* its listening socket, which a newer
+					   service may take over on a reload */
 	struct clientparam *child;
 	PROXYSERVICE service;
 	LOGFUNC logfunc;

tests/cases/reload.py

@@ -81,10 +81,6 @@ def run(t):
 
     t.write_config("reload", config(origin, second, adm, "bob:CL:two"))
     reload_and_wait()
-    # The old services stop accepting within a second; until then both the
-    # old and the new origin listen, and a connection the old one has queued
-    # is reset when it closes (a listener is not handed over yet).
-    time.sleep(1.5)
     t.ok("the new services start") if t.wait_port(second) else \
         t.fail("the new services start", f"port {second} open", "closed")
     moved = f"127.0.0.1:{second}"

tests/cases/takeover.py

@@ -0,0 +1,83 @@
+"""Listening sockets across a reload: a service of the new configuration on
+the same address and port takes over the old service's socket, so clients
+connecting all the while are never refused nor reset."""
+
+import socket
+import threading
+import time
+
+
+def run(t):
+    origin = t.free_port()
+    proxy = t.free_port()
+    adm = t.free_port()
+    conf = f"""
+        log
+        auth iponly
+        allow *
+        http data * /data
+        httpsrv -p{origin}
+
+        flush
+        auth iponly
+        allow *
+        proxy -p{proxy}
+        admin -p{adm}
+    """
+    server = t.start("takeover", conf, ports=[origin, proxy, adm])
+
+    done = {"ok": 0, "failed": 0, "errors": []}
+    stop = threading.Event()
+
+    def client():
+        url = f"http://127.0.0.1:{origin}/data?size=2000"
+        while not stop.is_set():
+            r = t.http(url, proxy=f"127.0.0.1:{proxy}")
+            if r.status == 200 and r.length == 2000:
+                done["ok"] += 1
+            else:
+                done["failed"] += 1
+                if len(done["errors"]) < 3:
+                    done["errors"].append(r.error or r.status)
+            # paced, so as not to run out of local ports
+            time.sleep(0.02)
+
+    clients = [threading.Thread(target=client) for _ in range(4)]
+    for c in clients:
+        c.start()
+
+    for i in range(3):
+        time.sleep(0.5)
+        before = server.output().count("taken over")
+        t.write_config("takeover", conf)
+        t.http(f"http://127.0.0.1:{adm}/R")
+        deadline = time.time() + 10
+        while time.time() < deadline and server.output().count("taken over") < before + 3:
+            time.sleep(0.05)
+    # the old services stop within a second of each reload
+    time.sleep(1.5)
+    stop.set()
+    for c in clients:
+        c.join(20)
+
+    taken = server.output().count("Accepting connections, taken over")
+    t.eq(9, taken, "each reload hands the three listening sockets over")
+    t.ok(f"requests go on through the reloads ({done['ok']})") if done["ok"] > 100 else \
+        t.fail("requests go on through the reloads", "more than 100", done["ok"])
+    t.eq(0, done["failed"], "no request is refused or reset by a reload "
+         f"(first errors: {done['errors']})")
+
+    # a changed port is a new socket: the old one closes, the new one opens
+    moved = t.free_port()
+    t.write_config("takeover", conf.replace(f"proxy -p{proxy}", f"proxy -p{moved}"))
+    t.http(f"http://127.0.0.1:{adm}/R")
+    t.ok("a service moved to another port listens there") if t.wait_port(moved) else \
+        t.fail("a service moved to another port listens there", "open", "closed")
+    deadline = time.time() + 5
+    while time.time() < deadline and t.wait_port(proxy, timeout=0.2):
+        time.sleep(0.2)
+    with socket.socket() as s:
+        s.settimeout(1)
+        refused = s.connect_ex(("127.0.0.1", proxy)) != 0
+    t.ok("and its old port is closed") if refused else \
+        t.fail("and its old port is closed", "refused", "accepted")