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.
Under Windows, if 3proxy is installed as a service you can use diff --git a/man/3proxy.8 b/man/3proxy.8 index 7b89b36..7f5f385 100644 --- a/man/3proxy.8 +++ b/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 diff --git a/src/3proxy.c b/src/3proxy.c index cd44c5b..ff8ce84 100644 --- a/src/3proxy.c +++ b/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); diff --git a/src/conf.c b/src/conf.c index 7693dbb..0d1b610 100644 --- a/src/conf.c +++ b/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; diff --git a/src/proxy.h b/src/proxy.h index 986b8e2..3fcbca7 100644 --- a/src/proxy.h +++ b/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); diff --git a/src/proxymain.c b/src/proxymain.c index bac67dc..a5616af 100644 --- a/src/proxymain.c +++ b/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 diff --git a/src/structures.h b/src/structures.h index 52b7de9..3b96115 100644 --- a/src/structures.h +++ b/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; diff --git a/tests/cases/reload.py b/tests/cases/reload.py index ac351f2..c7cc69f 100644 --- a/tests/cases/reload.py +++ b/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}" diff --git a/tests/cases/takeover.py b/tests/cases/takeover.py new file mode 100644 index 0000000..83a4e85 --- /dev/null +++ b/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")