diff --git a/src/3proxy.c b/src/3proxy.c index ff8ce84..90e6e07 100644 --- a/src/3proxy.c +++ b/src/3proxy.c @@ -558,6 +558,7 @@ int WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, LPWSTR lpCmdLine, int _3proxy_mutex_init(&config_mutex); _3proxy_mutex_init(&confref_mutex); _3proxy_mutex_init(&listener_mutex); + reattachfunc = reattach; _3proxy_mutex_init(&runningservices_mutex); _3proxy_mutex_init(&bandlim_mutex); _3proxy_mutex_init(&connlim_mutex); @@ -587,8 +588,9 @@ int WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, LPWSTR lpCmdLine, int confbegin(); initcommands(); - res = readconfig(fp); + /* numbered before it is read, as a reload numbers the new one */ conf->version++; + res = readconfig(fp); resolvfunc = conf->resolvfunc; /* readconfig() names the failing line, which an exit status would cut diff --git a/src/common.c b/src/common.c index b2dffc1..6df4f19 100644 --- a/src/common.c +++ b/src/common.c @@ -233,6 +233,7 @@ struct extparam confboot = { whatever runs meanwhile without a service of its own. */ struct extparam *conf = &confboot; struct extparam *runconf = &confboot; +int (*reattachfunc)(struct clientparam *param) = NULL; _3proxy_mutex_t confref_mutex; diff --git a/src/limiter.c b/src/limiter.c index c5b38cd..d25538e 100644 --- a/src/limiter.c +++ b/src/limiter.c @@ -15,8 +15,11 @@ int startconnlims (struct clientparam *param){ int ret = 0; param->connlim = 1; + /* released where they were taken, whatever configuration applies by + then: that one stays while the connection does */ + param->connlimconf = PSRV(param)->config; _3proxy_mutex_lock(&connlim_mutex); - for(ce = param->srv->config->connlimiter; ce; ce = ce->next) { + for(ce = param->connlimconf->connlimiter; ce; ce = ce->next) { if(ACLmatches(ce->ace, param)){ if(ce->ace->action == NOCONNLIM)break; if(!ce->period){ @@ -44,7 +47,7 @@ int startconnlims (struct clientparam *param){ } if(ret) { struct connlim * cee; - for(cee = param->srv->config->connlimiter; cee != ce; cee = cee->next) { + for(cee = param->connlimconf->connlimiter; cee != ce; cee = cee->next) { if(ACLmatches(cee->ace, param) && !cee->period && cee->rating) { cee->rating--; } @@ -59,7 +62,7 @@ void stopconnlims (struct clientparam *param){ struct connlim * ce; _3proxy_mutex_lock(&connlim_mutex); - for(ce = param->srv->config->connlimiter; ce; ce = ce->next) { + for(ce = (param->connlimconf? param->connlimconf : param->srv->config)->connlimiter; ce; ce = ce->next) { if(ACLmatches(ce->ace, param)){ if(ce->ace->action == NOCONNLIM)break; if(!ce->period && ce->rating){ @@ -78,28 +81,28 @@ void initbandlims (struct clientparam *param){ param->bandlimfunc = NULL; param->bandlims[0] = NULL; param->bandlimsout[0] = NULL; - if(!param->srv->config->bandlimfunc || (!param->srv->config->bandlimiter && !param->srv->config->bandlimiterout)) return; - for(i=0, be = param->srv->config->bandlimiter; be && inext) { + if(!PSRV(param)->config->bandlimfunc || (!PSRV(param)->config->bandlimiter && !PSRV(param)->config->bandlimiterout)) return; + for(i=0, be = PSRV(param)->config->bandlimiter; be && inext) { if(ACLmatches(be->ace, param)){ if(be->ace->action == NOBANDLIM) { break; } param->bandlims[i++] = be; - param->bandlimfunc = param->srv->config->bandlimfunc; + param->bandlimfunc = PSRV(param)->config->bandlimfunc; } } if(ibandlims[i] = NULL; - for(i=0, be = param->srv->config->bandlimiterout; be && inext) { + for(i=0, be = PSRV(param)->config->bandlimiterout; be && inext) { if(ACLmatches(be->ace, param)){ if(be->ace->action == NOBANDLIM) { break; } param->bandlimsout[i++] = be; - param->bandlimfunc = param->srv->config->bandlimfunc; + param->bandlimfunc = PSRV(param)->config->bandlimfunc; } } if(ibandlimsout[i] = NULL; - param->bandlimver = param->srv->config->bandlimver; + param->bandlimver = PSRV(param)->config->bandlimver; } unsigned bandlimitfunc(struct clientparam *param, unsigned nbytesin, unsigned nbytesout){ @@ -125,9 +128,9 @@ unsigned bandlimitfunc(struct clientparam *param, unsigned nbytesin, unsigned nb if(!nbytesin && !nbytesout) return 0; _3proxy_mutex_lock(&bandlim_mutex); - if(param->bandlimver != param->srv->config->bandlimver){ + if(param->bandlimver != PSRV(param)->config->bandlimver){ initbandlims(param); - param->bandlimver = param->srv->config->bandlimver; + param->bandlimver = PSRV(param)->config->bandlimver; } for(i=0; nbytesin&& ibandlims[i]; i++){ if( !param->bandlims[i]->basetime || @@ -172,7 +175,7 @@ void trafcountfunc(struct clientparam *param){ int countout = 0; _3proxy_mutex_lock(&tc_mutex); - for(tc = param->srv->config->trafcounter; tc; tc = tc->next) { + for(tc = PSRV(param)->config->trafcounter; tc; tc = tc->next) { if(ACLmatches(tc->ace, param)){ if(tc->ace->action == NOCOUNTIN) { @@ -188,7 +191,7 @@ void trafcountfunc(struct clientparam *param){ tc->v->updated = curtime; } } - if(countout) for(tc = param->srv->config->trafcounter; tc; tc = tc->next) { + if(countout) for(tc = PSRV(param)->config->trafcounter; tc; tc = tc->next) { if(ACLmatches(tc->ace, param)){ if(tc->ace->action == NOCOUNTOUT || tc->ace->action == NOCOUNTALL) break; if(tc->ace->action != COUNTOUT && tc->ace->action != COUNTALL ) { diff --git a/src/proxy.h b/src/proxy.h index 3fcbca7..751d546 100644 --- a/src/proxy.h +++ b/src/proxy.h @@ -226,6 +226,9 @@ extern int timetoexit, needreload; extern time_t curtime; /* the clock, kept by the main thread */ extern struct extparam *conf, *runconf; +/* the service whose rules apply to a client: its own, or after a reload the + newest one on its listener */ +#define PSRV(param) ((param)->newsrv? (param)->newsrv : (param)->srv) extern _3proxy_mutex_t confref_mutex, listener_mutex; struct extparam * confget(void); void confhold(struct extparam *c); @@ -397,6 +400,9 @@ void freeauth(struct auth *); void freefilter(struct filter *filter); void freeconf(struct extparam *confp); void confbegin(void); +int reattach(struct clientparam *param); +/* reattach() where configurations are reloaded, NULL elsewhere */ +extern int (*reattachfunc)(struct clientparam *param); struct passwords * copypwl (struct passwords *pwl); void freepwl(struct passwords *pw); int copyfilter(struct filter *, struct srvparam *srv); diff --git a/src/proxymain.c b/src/proxymain.c index a5616af..f461c73 100644 --- a/src/proxymain.c +++ b/src/proxymain.c @@ -883,13 +883,17 @@ struct listener { PROXYSOCKADDRTYPE sa; char *netns, *device; struct sockoptset opts; - struct srvparam *owner; /* the service accepting on it */ - int users; /* the services holding it */ + struct srvparam *owner; /* the service accepting on it, if any */ + int users; /* the services holding it: the one accepting, + and those which did until their last client + is gone, whose clients look up the newest + service through it after a reload */ }; static struct listener *listeners = NULL; _3proxy_mutex_t listener_mutex; static int listenerkeep(struct srvparam *srv, SOCKET sock); +static void listenerstop(struct srvparam *srv); static void listenerrelease(struct srvparam *srv); #endif @@ -1543,7 +1547,7 @@ void mainclaim(void *service){ 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)){ + if(l->owner && l->sock != INVALID_SOCKET && l->owner->config != srv->config && samelistener(l, srv)){ l->owner = srv; l->users++; srv->listener = l; @@ -1579,8 +1583,29 @@ static int listenerkeep(struct srvparam *srv, SOCKET sock){ return 0; } -/* A service lets go of its listening socket: closed if nobody else holds - it. */ +/* A service stops accepting: if it was the one accepting and no service + took the socket over, it is closed now, so that connections do not queue + up on it. */ +static void listenerstop(struct srvparam *srv){ + struct listener *l = srv->listener; + SOCKET sock = INVALID_SOCKET; + + _3proxy_mutex_lock(&listener_mutex); + if(l->owner == srv){ + l->owner = NULL; + sock = l->sock; + l->sock = INVALID_SOCKET; + } + _3proxy_mutex_unlock(&listener_mutex); + if(sock == INVALID_SOCKET) return; + so._closesocket(so.state, sock); +#ifdef WITH_UN + if(*SAFAMILY(&l->sa) == AF_UNIX && *SAADDR(&l->sa))unlink((char *)SAADDR(&l->sa)); +#endif +} + +/* A service lets go of its listener once its last client is gone; the last + to let go frees it. */ static void listenerrelease(struct srvparam *srv){ struct listener *l = srv->listener, **lp; int last; @@ -1595,15 +1620,71 @@ static void listenerrelease(struct srvparam *srv){ } _3proxy_mutex_unlock(&listener_mutex); if(!last) return; - so._closesocket(so.state, l->sock); + if(l->sock != INVALID_SOCKET){ + so._closesocket(so.state, l->sock); #ifdef WITH_UN - if(*SAFAMILY(&l->sa) == AF_UNIX && *SAADDR(&l->sa))unlink((char *)SAADDR(&l->sa)); + 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); } +/* Gives back a client's slot in a service it was not a child of. */ +static void slotrelease(struct srvparam *srv){ + _3proxy_mutex_lock(&srv->counter_mutex); + _3proxy_sem_unlock(srv->childslots); + _3proxy_mutex_unlock(&srv->counter_mutex); +} + +/* After a reload, unless noforce: a client goes by the rules of the newest + service on its listener, the one now accepting there, re-authenticated + as it. It takes one of that service's maxconn slots, which counts it + there and keeps the service, and its configuration, until the client is + done; a slot it held in a service before is given back. 0 to go on, or + the code to end the connection with: 26 if its service was removed, or + is another one now, 27 if the newest has no slot free, or what the + authentication says. */ +int reattach(struct clientparam *param){ + struct srvparam *cur = PSRV(param), *newest, *orig; + int version = runconf->version, res; + + /* without a listener (UDP, connecting back) nothing replaces it */ + if(!cur->listener){ + param->version = version; + return 0; + } + _3proxy_mutex_lock(&listener_mutex); + newest = cur->listener->owner; + if(newest == cur){ + /* the old service has not let go of the socket yet */ + _3proxy_mutex_unlock(&listener_mutex); + return 0; + } + /* param->service: the old service is a zombie by now */ + if(!newest || newest->service != param->service){ + _3proxy_mutex_unlock(&listener_mutex); + return 26; + } + /* while it accepts the newest is not freed, nor after it has a slot */ + if(_3proxy_sem_timedlock(newest->childslots, 0)){ + _3proxy_mutex_unlock(&listener_mutex); + return 27; + } + _3proxy_mutex_unlock(&listener_mutex); + if(param->newsrv) slotrelease(param->newsrv); + param->newsrv = newest; + param->version = version; + /* authenticated as the newest service's client: its access lists, users, + limits and counters */ + orig = param->srv; + param->srv = newest; + res = (*newest->authfunc)(param); + param->srv = orig; + return res; +} + /* A prepared service which could not be started. */ void maindrop(void *service){ struct srvstart *s = (struct srvstart *)service; @@ -1623,7 +1704,7 @@ _3proxy_sem_t udpinit; void srvinit(struct srvparam * srv, struct clientparam *param){ memset(srv, 0, sizeof(struct srvparam)); - srv->version = conf->version + 1; + srv->version = conf->version; srv->paused = conf->paused; srv->logfunc = conf->havelog?conf->logfunc:lognone; srv->noforce = conf->noforce; @@ -1717,11 +1798,11 @@ 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 */ + /* a listening socket kept as a listener goes on if a service of a later + configuration took it over */ if(srv->listener){ srv->srvsock = INVALID_SOCKET; - listenerrelease(srv); + listenerstop(srv); } #endif if(srv->srvsock != INVALID_SOCKET) { @@ -1781,6 +1862,7 @@ void srvfree(struct srvparam * srv){ #endif /* the last service of a configuration no longer in force releases it */ #ifndef STDMAIN + if(srv->listener) listenerrelease(srv); if(srv->config){ struct extparam *c = srv->config; @@ -1856,6 +1938,11 @@ void releaseparam(struct clientparam * param) { (*param->filters[param->nfilters].filter->filter_clear)(param->filters[param->nfilters].data); } if(param->connlim) stopconnlims(param); + /* a slot held in the newest service on the listener after a reload */ + if(param->newsrv){ + slotrelease(param->newsrv); + param->newsrv = NULL; + } #endif /* the sockets were closed and the filters cleared with it */ param->sostate = NULL; diff --git a/src/sockmap.c b/src/sockmap.c index c5e297c..57082b8 100644 --- a/src/sockmap.c +++ b/src/sockmap.c @@ -268,10 +268,10 @@ log(logbuf); (param->srv->logdumpcli && (param->statscli64 > param->srv->logdumpcli))) dolog(param, NULL); - if(param->version < param->srv->config->version){ - if(!param->srv->noforce && (res = (*param->srv->authfunc)(param)) && res != 2) {RETURN(res);} - param->paused = param->srv->config->paused; - param->version = param->srv->config->version; + /* a reload: the client goes by the newest configuration's rules */ + if(param->version < runconf->version){ + if(param->srv->noforce || !reattachfunc) param->version = runconf->version; + else if((res = (*reattachfunc)(param)) && res != 2) {RETURN(res);} } if((param->maxtrafin64 && param->statssrv64 >= param->maxtrafin64) || (param->maxtrafout64 && param->statscli64 >= param->maxtrafout64)){ diff --git a/src/structures.h b/src/structures.h index 3b96115..8495829 100644 --- a/src/structures.h +++ b/src/structures.h @@ -708,6 +708,12 @@ struct clientparam { *prev; void * sostate; struct srvparam *srv; + /* After a reload: the newest service on the same listener, one of whose + maxconn slots the client holds; its configuration's rules apply + (PSRV()). */ + struct srvparam *newsrv; + /* the configuration whose connection limits the client counts in */ + struct extparam *connlimconf; REDIRECTFUNC redirectfunc; BANDLIMFUNC bandlimfunc; TRAFCOUNTFUNC trafcountfunc; diff --git a/tests/cases/reload.py b/tests/cases/reload.py index c7cc69f..3cf3048 100644 --- a/tests/cases/reload.py +++ b/tests/cases/reload.py @@ -66,34 +66,47 @@ def run(t): t.eq(200, as_user("alice", "one"), "after a failed reload alice still gets in") t.ne(200, as_user("bob", "two"), "and bob, from the refused file, does not") - # --- a good file: the new configuration takes over ------------------- - slow = {} - - def download(): + # --- a good file: the new configuration takes over, and the connections + # of the old services go by its rules from then on --------------------- + def download(result): # about two seconds: 2 MB in 64 KB blocks, 60 ms apart r = t.http(f"http://127.0.0.1:{origin}/data?size=2000000&block=65536&delay=60", proxy=proxy, proxy_auth=("alice", "one")) - slow["status"], slow["length"] = r.status, r.length + result["status"], result["length"] = r.status, r.length - th = threading.Thread(target=download) - th.start() - time.sleep(0.3) + def during_reload(users, port): + result = {} + th = threading.Thread(target=download, args=(result,)) + th.start() + time.sleep(0.3) + t.write_config("reload", config(origin, port, adm, users)) + reload_and_wait() + return th, result + # alice stays: her download goes on under the new service + th, kept = during_reload("alice:CL:one bob:CL:two", first) + t.eq(200, as_user("bob", "two"), "the new configuration lets bob in") + th.join(20) + t.eq(2000000, kept.get("length"), + "a download of a user the new configuration keeps finishes whole") + + # alice is removed: her download is cut, being re-authenticated + th, cut = during_reload("bob:CL:two", first) + th.join(20) + t.ok("a download of a user the new configuration removes is cut") \ + if cut.get("length", 0) < 2000000 else \ + t.fail("a download of a user the new configuration removes is cut", + "less than 2000000 bytes", cut.get("length")) + t.ne(200, as_user("alice", "one"), "and alice cannot come back") + + # the service moves to another port: the old one closes t.write_config("reload", config(origin, second, adm, "bob:CL:two")) reload_and_wait() - 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}" - t.eq(200, as_user("bob", "two", moved), "the new configuration lets bob in") - t.ne(200, as_user("alice", "one", moved), "and not alice, who is gone from it") - - # the old service stops accepting, its port closes + t.ok("the moved service listens on its new port") if t.wait_port(second) else \ + t.fail("the moved service listens on its new port", f"port {second} open", "closed") + t.eq(200, as_user("bob", "two", f"127.0.0.1:{second}"), "bob gets in there") deadline = time.time() + 5 while time.time() < deadline and t.wait_port(first, timeout=0.2): time.sleep(0.2) - t.ok("the old service stops listening") if not t.wait_port(first, timeout=0.2) else \ - t.fail("the old service stops listening", f"port {first} closed", "open") - - th.join(20) - t.eq(200, slow.get("status"), "a download through the old service finishes") - t.eq(2000000, slow.get("length"), "with the whole answer") + t.ok("the old port is closed") if not t.wait_port(first, timeout=0.2) else \ + t.fail("the old port is closed", f"port {first} closed", "open")