Hand UDP sockets over on a reload too

A UDP service's socket is kept as a listener like a TCP one, the
protocol part of what must match: on a reload a new UDP service on the
same address takes it over. It is the way the old service's sessions
answer as well, and they go on answering through it; the new service
reads what comes, a datagram of an old session forwarded to it as
before, sessions being found by the service's and the client's
addresses whichever service made them. The socket stays open while the
old sessions do.

The takeover case keeps a udppm session going through three reloads
and wants every datagram back; with the build before this datagrams
were lost.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Vladimir Dubrovin 2026-10-14 23:30:52 +03:00
parent 724c92f3f6
commit df09565fe4
2 changed files with 74 additions and 10 deletions

View File

@ -875,11 +875,14 @@ static int srvprepare(struct srvstart *s, int argc, char **argv){
reload, a service of the new configuration listening on the same reload, a service of the new configuration listening on the same
address, port, network namespace and device with the same options takes address, port, network namespace and device with the same options takes
it over. The old service stops accepting once its configuration is no 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: longer in force; whoever lets go of the socket last closes it. A UDP
a UDP service's clients read from its socket. */ service's socket is the way its sessions answer too: the old sessions go
on answering through it, the new service reading what comes, the
sessions found by client address whichever service made them. */
struct listener { struct listener {
struct listener *next; struct listener *next;
SOCKET sock; SOCKET sock;
int isudp;
PROXYSOCKADDRTYPE sa; PROXYSOCKADDRTYPE sa;
char *netns, *device; char *netns, *device;
struct sockoptset opts; struct sockoptset opts;
@ -892,7 +895,7 @@ struct listener {
static struct listener *listeners = NULL; static struct listener *listeners = NULL;
_3proxy_mutex_t listener_mutex; _3proxy_mutex_t listener_mutex;
static int listenerkeep(struct srvparam *srv, SOCKET sock); static int listenerkeep(struct srvparam *srv, SOCKET sock, int isudp);
static void listenerstop(struct srvparam *srv); static void listenerstop(struct srvparam *srv);
static void listenerrelease(struct srvparam *srv); static void listenerrelease(struct srvparam *srv);
#endif #endif
@ -912,6 +915,10 @@ static int srvlisten(struct srvstart *s){
/* taken over from a service of the configuration before: listening /* taken over from a service of the configuration before: listening
already, only the backlog may be new */ already, only the backlog may be new */
sock = srv->srvsock = srv->listener->sock; sock = srv->srvsock = srv->listener->sock;
#ifndef NOUDPMAIN
if(s->isudp) defparam->clisock = sock;
else
#endif
if(srv->so._listen (srv->so.state, sock, srv->backlog?srv->backlog : 1+(srv->maxchild>>3))==-1) { if(srv->so._listen (srv->so.state, sock, srv->backlog?srv->backlog : 1+(srv->maxchild>>3))==-1) {
sprintf((char *)buf, "listen(): %s", strerror(errno)); sprintf((char *)buf, "listen(): %s", strerror(errno));
if(!srv->silent)dolog(defparam, buf); if(!srv->silent)dolog(defparam, buf);
@ -981,13 +988,17 @@ static int srvlisten(struct srvstart *s){
#ifndef STDMAIN #ifndef STDMAIN
/* kept for a service of a later configuration to take over; without /* kept for a service of a later configuration to take over; without
memory for it the service closes the socket itself */ memory for it the service closes the socket itself */
listenerkeep(srv, sock); listenerkeep(srv, sock, 0);
#endif #endif
} }
#endif #endif
#ifndef NOUDPMAIN #ifndef NOUDPMAIN
if(s->isudp) if(s->isudp){
defparam->clisock = sock; defparam->clisock = sock;
#ifndef STDMAIN
listenerkeep(srv, sock, 1);
#endif
}
#endif #endif
srv->fds.fd = sock; srv->fds.fd = sock;
srv->fds.events = POLLIN; srv->fds.events = POLLIN;
@ -1524,7 +1535,8 @@ static int samestr(const char *a, const char *b){
return (!a && !b) || (a && b && !strcmp(a, b)); return (!a && !b) || (a && b && !strcmp(a, b));
} }
static int samelistener(struct listener *l, struct srvparam *srv){ static int samelistener(struct listener *l, struct srvparam *srv, int isudp){
if(l->isudp != isudp) return 0;
#ifdef __linux__ #ifdef __linux__
if(!samestr(l->netns, srv->inetns)) return 0; if(!samestr(l->netns, srv->inetns)) return 0;
#endif #endif
@ -1544,10 +1556,10 @@ void mainclaim(void *service){
struct srvparam *srv = &s->srv; struct srvparam *srv = &s->srv;
struct listener *l; struct listener *l;
if(s->isudp || s->iscbc) return; if(!s->isudp && s->iscbc) return; /* connects back: no socket of its own */
_3proxy_mutex_lock(&listener_mutex); _3proxy_mutex_lock(&listener_mutex);
for(l = listeners; l; l = l->next){ for(l = listeners; l; l = l->next){
if(l->owner && l->sock != INVALID_SOCKET && l->owner->config != srv->config && samelistener(l, srv)){ if(l->owner && l->sock != INVALID_SOCKET && l->owner->config != srv->config && samelistener(l, srv, s->isudp)){
l->owner = srv; l->owner = srv;
l->users++; l->users++;
srv->listener = l; srv->listener = l;
@ -1559,12 +1571,13 @@ void mainclaim(void *service){
/* A socket a service opened and bound itself, kept for whoever may take it /* 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. */ over; 0, or 1 for want of memory, when the service closes it itself. */
static int listenerkeep(struct srvparam *srv, SOCKET sock){ static int listenerkeep(struct srvparam *srv, SOCKET sock, int isudp){
struct listener *l; struct listener *l;
if(!(l = malloc(sizeof(*l)))) return 1; if(!(l = malloc(sizeof(*l)))) return 1;
memset(l, 0, sizeof(*l)); memset(l, 0, sizeof(*l));
l->sock = sock; l->sock = sock;
l->isudp = isudp;
l->sa = srv->intsa; l->sa = srv->intsa;
l->opts = srv->lissockopts; l->opts = srv->lissockopts;
#ifdef __linux__ #ifdef __linux__
@ -1650,7 +1663,7 @@ int reattach(struct clientparam *param){
struct srvparam *cur = PSRV(param), *newest, *orig; struct srvparam *cur = PSRV(param), *newest, *orig;
int version = runconf->version, res; int version = runconf->version, res;
/* without a listener (UDP, connecting back) nothing replaces it */ /* without a listener (connecting back) nothing replaces it */
if(!cur->listener){ if(!cur->listener){
param->version = version; param->version = version;
return 0; return 0;

View File

@ -67,6 +67,57 @@ def run(t):
t.eq(0, done["failed"], "no request is refused or reset by a reload " t.eq(0, done["failed"], "no request is refused or reset by a reload "
f"(first errors: {done['errors']})") f"(first errors: {done['errors']})")
# --- UDP: the socket is handed over too, and a session of the old
# service goes on through it, its datagrams forwarded by the new one ---
echo = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
echo.bind(("127.0.0.1", 0))
echo_port = echo.getsockname()[1]
def echo_server():
while True:
try:
data, addr = echo.recvfrom(2048)
except OSError:
return
echo.sendto(data, addr)
threading.Thread(target=echo_server, daemon=True).start()
udp = t.free_port()
uconf = conf + f"\n udppm {udp} 127.0.0.1 {echo_port}\n"
t.write_config("takeover", uconf)
t.http(f"http://127.0.0.1:{adm}/R")
time.sleep(1.5)
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
client.settimeout(1)
sent = lost = 0
stopudp = time.time() + 5
reloads = 0
next_reload = time.time() + 0.5
while time.time() < stopudp:
if reloads < 3 and time.time() >= next_reload:
t.write_config("takeover", uconf)
t.http(f"http://127.0.0.1:{adm}/R")
reloads += 1
next_reload = time.time() + 1.2
msg = f"n{sent}".encode()
client.sendto(msg, ("127.0.0.1", udp))
sent += 1
try:
data, _ = client.recvfrom(2048)
if data != msg:
lost += 1
except OSError:
lost += 1
time.sleep(0.02)
client.close()
echo.close()
t.ok(f"a UDP session sends through three reloads ({sent} datagrams)") if sent > 100 else \
t.fail("a UDP session sends through three reloads", "more than 100", sent)
t.eq(0, lost, "and every datagram comes back, the socket handed over")
t.contains(server.output(), f"{udp} 0.0.0.0:0 0 0 0 Accepting connections, taken over",
"the UDP service takes the socket over")
# a changed port is a new socket: the old one closes, the new one opens # a changed port is a new socket: the old one closes, the new one opens
moved = t.free_port() moved = t.free_port()
t.write_config("takeover", conf.replace(f"proxy -p{proxy}", f"proxy -p{moved}")) t.write_config("takeover", conf.replace(f"proxy -p{proxy}", f"proxy -p{moved}"))