mirror of
https://github.com/3proxy/3proxy.git
synced 2026-10-28 22:05:48 +08:00
UDP loop: epoll on Linux
The UDP loop gave poll() every socket of its sessions each time round. On Linux it now keeps them in an epoll set: the service's socket and each session's socket and parent's control connection are registered once and changed as the session changes, a socket replaced or other events waited for, and an event leads straight to its session. A handler replacing a session's socket marks it to be registered again, as a closed socket leaves the set and its number may come back for the new one; dnspr does so moving on to the next name server. Sessions handed over on a reload leave the set of the old service's loop and are registered by the one adopting them; the service's socket is left unwatched while an older service still serves it. Elsewhere, or built with NOEPOLL, poll() is used as before. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
f5e646ba86
commit
d9b9d1b9b7
@ -156,6 +156,8 @@ static int dnsforward(struct udploop *u, struct udpsession *ss){
|
|||||||
param->srv->so._closesocket(param->sostate, param->remsock);
|
param->srv->so._closesocket(param->sostate, param->remsock);
|
||||||
param->remsock = INVALID_SOCKET;
|
param->remsock = INVALID_SOCKET;
|
||||||
}
|
}
|
||||||
|
/* another socket for the loop to watch */
|
||||||
|
ss->regsock = INVALID_SOCKET;
|
||||||
q->tcp = c->nservers[q->server].usetcp;
|
q->tcp = c->nservers[q->server].usetcp;
|
||||||
q->got = 0;
|
q->got = 0;
|
||||||
{
|
{
|
||||||
|
|||||||
136
src/mainudp.c
136
src/mainudp.c
@ -14,18 +14,76 @@
|
|||||||
clients come on the service's socket, those of their destinations on the
|
clients come on the service's socket, those of their destinations on the
|
||||||
sockets of their sessions, and every answer to a client leaves from the
|
sockets of their sessions, and every answer to a client leaves from the
|
||||||
service's socket. No thread for a client. */
|
service's socket. No thread for a client. */
|
||||||
|
/* On Linux the sockets are watched with epoll, registered once and changed
|
||||||
|
as the sessions change; elsewhere poll() is given them all each time.
|
||||||
|
NOEPOLL keeps poll() on Linux too. */
|
||||||
|
#if defined(__linux__) && !defined(NOEPOLL)
|
||||||
|
#define UDPEPOLL
|
||||||
|
#include <sys/epoll.h>
|
||||||
|
#endif
|
||||||
|
|
||||||
struct udpstate {
|
struct udpstate {
|
||||||
struct udploop u;
|
struct udploop u;
|
||||||
const struct udpservice *svc;
|
const struct udpservice *svc;
|
||||||
struct udpsession *head;
|
struct udpsession *head;
|
||||||
int count;
|
int count;
|
||||||
struct hashtable keys; /* a session by its client's address */
|
struct hashtable keys; /* a session by its client's address */
|
||||||
|
#ifdef UDPEPOLL
|
||||||
|
int ep;
|
||||||
|
struct epoll_event *evs;
|
||||||
|
short srvevents; /* the service's socket as registered */
|
||||||
|
#else
|
||||||
struct pollfd *fds;
|
struct pollfd *fds;
|
||||||
struct udpsession **of; /* the session of each of fds, NULL: the service */
|
struct udpsession **of; /* the session of each of fds, NULL: the service */
|
||||||
|
#endif
|
||||||
int nalloc;
|
int nalloc;
|
||||||
int nfull;
|
int nfull;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
#ifdef UDPEPOLL
|
||||||
|
/* A socket in the epoll set; its data the session, its lowest bit set for
|
||||||
|
a parent's control connection, or NULL for the service's socket. */
|
||||||
|
static int udpwatch(struct udpstate *st, int op, SOCKET s, short events, void *tag){
|
||||||
|
struct epoll_event e;
|
||||||
|
|
||||||
|
memset(&e, 0, sizeof(e));
|
||||||
|
e.events = ((events & POLLIN)? EPOLLIN : 0) | ((events & POLLOUT)? EPOLLOUT : 0);
|
||||||
|
e.data.ptr = tag;
|
||||||
|
return epoll_ctl(st->ep, op, s, &e);
|
||||||
|
}
|
||||||
|
|
||||||
|
/* The session's sockets registered as they are now. */
|
||||||
|
static void udpsync(struct udpstate *st, struct udpsession *ss){
|
||||||
|
SOCKET rem = ss->param->remsock, ctrl = ss->param->ctrlsocksrv;
|
||||||
|
short want = ss->events? ss->events : POLLIN;
|
||||||
|
|
||||||
|
if(ctrl == rem) ctrl = INVALID_SOCKET;
|
||||||
|
if(ss->regsock != rem){
|
||||||
|
if(ss->regsock != INVALID_SOCKET) udpwatch(st, EPOLL_CTL_DEL, ss->regsock, 0, NULL);
|
||||||
|
if(rem != INVALID_SOCKET) udpwatch(st, EPOLL_CTL_ADD, rem, want, ss);
|
||||||
|
ss->regsock = rem;
|
||||||
|
ss->regevents = want;
|
||||||
|
}
|
||||||
|
else if(rem != INVALID_SOCKET && ss->regevents != want){
|
||||||
|
udpwatch(st, EPOLL_CTL_MOD, rem, want, ss);
|
||||||
|
ss->regevents = want;
|
||||||
|
}
|
||||||
|
if(ss->regctrl != ctrl){
|
||||||
|
if(ss->regctrl != INVALID_SOCKET) udpwatch(st, EPOLL_CTL_DEL, ss->regctrl, 0, NULL);
|
||||||
|
if(ctrl != INVALID_SOCKET) udpwatch(st, EPOLL_CTL_ADD, ctrl, POLLIN, (char *)ss + 1);
|
||||||
|
ss->regctrl = ctrl;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/* Out of the set, before its sockets are closed or it goes to another
|
||||||
|
loop. */
|
||||||
|
static void udpunwatch(struct udpstate *st, struct udpsession *ss){
|
||||||
|
if(ss->regsock != INVALID_SOCKET) udpwatch(st, EPOLL_CTL_DEL, ss->regsock, 0, NULL);
|
||||||
|
if(ss->regctrl != INVALID_SOCKET) udpwatch(st, EPOLL_CTL_DEL, ss->regctrl, 0, NULL);
|
||||||
|
ss->regsock = ss->regctrl = INVALID_SOCKET;
|
||||||
|
}
|
||||||
|
#endif
|
||||||
|
|
||||||
static void udpkey2hash(const struct hashtable *ht, void *index, uint8_t *hash){
|
static void udpkey2hash(const struct hashtable *ht, void *index, uint8_t *hash){
|
||||||
PROXYSOCKADDRTYPE *sa = (PROXYSOCKADDRTYPE *)index;
|
PROXYSOCKADDRTYPE *sa = (PROXYSOCKADDRTYPE *)index;
|
||||||
uint32_t m1, m2;
|
uint32_t m1, m2;
|
||||||
@ -91,6 +149,7 @@ static struct udpsession *udpnew(struct udpstate *st){
|
|||||||
udpjoin(srv, param, 1);
|
udpjoin(srv, param, 1);
|
||||||
ss->param = param;
|
ss->param = param;
|
||||||
ss->svc = st->svc;
|
ss->svc = st->svc;
|
||||||
|
ss->regsock = ss->regctrl = INVALID_SOCKET;
|
||||||
#ifdef MODULEMAINFUNC
|
#ifdef MODULEMAINFUNC
|
||||||
if(makefilters(srv, param) > CONTINUE){
|
if(makefilters(srv, param) > CONTINUE){
|
||||||
freeparam(param);
|
freeparam(param);
|
||||||
@ -155,6 +214,9 @@ static void udphandover(struct udpstate *st){
|
|||||||
for(ss = st->head; ss; ss = ss->next){
|
for(ss = st->head; ss; ss = ss->next){
|
||||||
if(ss->keyed) hashdelete(&st->keys, &ss->param->sincr);
|
if(ss->keyed) hashdelete(&st->keys, &ss->param->sincr);
|
||||||
ss->keyed = 0;
|
ss->keyed = 0;
|
||||||
|
#ifdef UDPEPOLL
|
||||||
|
udpunwatch(st, ss);
|
||||||
|
#endif
|
||||||
last = ss;
|
last = ss;
|
||||||
}
|
}
|
||||||
if(!last) return;
|
if(!last) return;
|
||||||
@ -190,6 +252,8 @@ static void udpadopt(struct udpstate *st){
|
|||||||
while((ss = list)){
|
while((ss = list)){
|
||||||
list = ss->next;
|
list = ss->next;
|
||||||
ss->next = ss->prev = NULL;
|
ss->next = ss->prev = NULL;
|
||||||
|
/* watched by the old service's loop, not by this one yet */
|
||||||
|
ss->regsock = ss->regctrl = INVALID_SOCKET;
|
||||||
param = ss->param;
|
param = ss->param;
|
||||||
old = param->srv;
|
old = param->srv;
|
||||||
if(old == srv) { udpadd(st, ss); continue; }
|
if(old == srv) { udpadd(st, ss); continue; }
|
||||||
@ -303,7 +367,10 @@ static int udploop(struct srvstart *s){
|
|||||||
struct srvparam *srv = &s->srv;
|
struct srvparam *srv = &s->srv;
|
||||||
struct udpsession *ss, *next;
|
struct udpsession *ss, *next;
|
||||||
time_t now;
|
time_t now;
|
||||||
int i, n, nfds, timeo, res = 0, wait = 0;
|
int i, n, timeo, res = 0, wait = 0, srvready;
|
||||||
|
#ifndef UDPEPOLL
|
||||||
|
int nfds;
|
||||||
|
#endif
|
||||||
|
|
||||||
memset(&st, 0, sizeof(st));
|
memset(&st, 0, sizeof(st));
|
||||||
st.u.srv = srv;
|
st.u.srv = srv;
|
||||||
@ -315,18 +382,35 @@ static int udploop(struct srvstart *s){
|
|||||||
st.keys.recsize = sizeof(struct udpsession *);
|
st.keys.recsize = sizeof(struct udpsession *);
|
||||||
st.keys.hash_size = 8;
|
st.keys.hash_size = 8;
|
||||||
if(st.svc->keyed && inithashtable(&st.keys, 64, 256, 1048576)) return 21;
|
if(st.svc->keyed && inithashtable(&st.keys, 64, 256, 1048576)) return 21;
|
||||||
|
#ifdef UDPEPOLL
|
||||||
|
if((st.ep = epoll_create1(EPOLL_CLOEXEC)) < 0 || udpwatch(&st, EPOLL_CTL_ADD, srv->srvsock, POLLIN, NULL)){
|
||||||
|
if(st.ep >= 0) close(st.ep);
|
||||||
|
if(st.svc->keyed) destroyhashtable(&st.keys);
|
||||||
|
return 21;
|
||||||
|
}
|
||||||
|
st.srvevents = POLLIN;
|
||||||
|
#endif
|
||||||
for(;;){
|
for(;;){
|
||||||
if(srv->config->paused != srv->paused) break;
|
if(srv->config->paused != srv->paused) break;
|
||||||
#ifndef STDMAIN
|
#ifndef STDMAIN
|
||||||
udpadopt(&st);
|
udpadopt(&st);
|
||||||
#endif
|
#endif
|
||||||
|
/* room for every socket: a session has its socket and maybe a
|
||||||
|
parent's control connection */
|
||||||
if(st.nalloc < 2 * st.count + 1){
|
if(st.nalloc < 2 * st.count + 1){
|
||||||
int want = 2 * st.count + 64;
|
int want = 2 * st.count + 64;
|
||||||
|
#ifdef UDPEPOLL
|
||||||
|
struct epoll_event *e = realloc(st.evs, want * sizeof(*e));
|
||||||
|
|
||||||
|
if(!e) { res = 21; break; }
|
||||||
|
st.evs = e;
|
||||||
|
#else
|
||||||
struct pollfd *f = realloc(st.fds, want * sizeof(*f));
|
struct pollfd *f = realloc(st.fds, want * sizeof(*f));
|
||||||
|
|
||||||
if(f) st.fds = f;
|
if(f) st.fds = f;
|
||||||
if(f && (st.of = realloc(st.of, want * sizeof(*st.of)))) st.nalloc = want;
|
if(!f || !(st.of = realloc(st.of, want * sizeof(*st.of)))) { res = 21; break; }
|
||||||
else { res = 21; break; }
|
#endif
|
||||||
|
st.nalloc = want;
|
||||||
}
|
}
|
||||||
now = time(NULL);
|
now = time(NULL);
|
||||||
timeo = 1000;
|
timeo = 1000;
|
||||||
@ -334,6 +418,40 @@ static int udploop(struct srvstart *s){
|
|||||||
/* waiting for an older service's sessions: soon looked for again */
|
/* waiting for an older service's sessions: soon looked for again */
|
||||||
if((wait = udpolder(srv))) timeo = 20;
|
if((wait = udpolder(srv))) timeo = 20;
|
||||||
#endif
|
#endif
|
||||||
|
srvready = 0;
|
||||||
|
#ifdef UDPEPOLL
|
||||||
|
if(st.srvevents != (wait? 0 : POLLIN)){
|
||||||
|
st.srvevents = wait? 0 : POLLIN;
|
||||||
|
udpwatch(&st, EPOLL_CTL_MOD, srv->srvsock, st.srvevents, NULL);
|
||||||
|
}
|
||||||
|
for(ss = st.head; ss; ss = ss->next){
|
||||||
|
if(ss->deadline && (ss->deadline - now) * 1000 < timeo)
|
||||||
|
timeo = ss->deadline > now? (int)(ss->deadline - now) * 1000 : 0;
|
||||||
|
udpsync(&st, ss);
|
||||||
|
}
|
||||||
|
n = epoll_wait(st.ep, st.evs, st.nalloc, timeo);
|
||||||
|
if(n < 0 && errno != EINTR){
|
||||||
|
res = 91;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
now = time(NULL);
|
||||||
|
for(i = 0; i < n; i++){
|
||||||
|
uintptr_t tag = (uintptr_t)st.evs[i].data.ptr;
|
||||||
|
|
||||||
|
if(!tag){
|
||||||
|
srvready = 1;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
ss = (struct udpsession *)(tag & ~(uintptr_t)1);
|
||||||
|
if(ss->done) continue;
|
||||||
|
/* a parent's control connection: its end ends the session */
|
||||||
|
if(tag & 1) ss->done = 1;
|
||||||
|
else {
|
||||||
|
udprefresh(ss, now);
|
||||||
|
if((*st.svc->server)(&st.u, ss)) ss->done = 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
#else
|
||||||
nfds = 0;
|
nfds = 0;
|
||||||
st.fds[nfds].fd = srv->srvsock;
|
st.fds[nfds].fd = srv->srvsock;
|
||||||
st.fds[nfds].events = wait? 0 : POLLIN;
|
st.fds[nfds].events = wait? 0 : POLLIN;
|
||||||
@ -371,7 +489,9 @@ static int udploop(struct srvstart *s){
|
|||||||
}
|
}
|
||||||
else ss->done = 1;
|
else ss->done = 1;
|
||||||
}
|
}
|
||||||
if(n > 0 && !wait && st.fds[0].revents) udpclients(&st, now);
|
srvready = n > 0 && st.fds[0].revents;
|
||||||
|
#endif
|
||||||
|
if(srvready && !wait) udpclients(&st, now);
|
||||||
for(ss = st.head; ss; ss = next){
|
for(ss = st.head; ss; ss = next){
|
||||||
next = ss->next;
|
next = ss->next;
|
||||||
if(!ss->done && ss->deadline && ss->deadline <= now){
|
if(!ss->done && ss->deadline && ss->deadline <= now){
|
||||||
@ -383,6 +503,9 @@ static int udploop(struct srvstart *s){
|
|||||||
}
|
}
|
||||||
if(ss->done){
|
if(ss->done){
|
||||||
udpremove(&st, ss);
|
udpremove(&st, ss);
|
||||||
|
#ifdef UDPEPOLL
|
||||||
|
udpunwatch(&st, ss);
|
||||||
|
#endif
|
||||||
udpend(ss);
|
udpend(ss);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -392,8 +515,13 @@ static int udploop(struct srvstart *s){
|
|||||||
#endif
|
#endif
|
||||||
udpendall(st.head, 26);
|
udpendall(st.head, 26);
|
||||||
if(st.svc->keyed) destroyhashtable(&st.keys);
|
if(st.svc->keyed) destroyhashtable(&st.keys);
|
||||||
|
#ifdef UDPEPOLL
|
||||||
|
close(st.ep);
|
||||||
|
free(st.evs);
|
||||||
|
#else
|
||||||
free(st.fds);
|
free(st.fds);
|
||||||
free(st.of);
|
free(st.of);
|
||||||
|
#endif
|
||||||
return res;
|
return res;
|
||||||
}
|
}
|
||||||
#endif
|
#endif
|
||||||
|
|||||||
@ -253,6 +253,11 @@ struct udpsession {
|
|||||||
int keyed; /* found by its client's address */
|
int keyed; /* found by its client's address */
|
||||||
int done; /* to be ended, param->res saying how */
|
int done; /* to be ended, param->res saying how */
|
||||||
short events; /* polled for on remsock, 0: POLLIN */
|
short events; /* polled for on remsock, 0: POLLIN */
|
||||||
|
/* the sockets as the loop watches them (epoll): a handler which
|
||||||
|
replaces param->remsock sets regsock to INVALID_SOCKET, as the number
|
||||||
|
of the one closed may come back for the new one */
|
||||||
|
SOCKET regsock, regctrl;
|
||||||
|
short regevents;
|
||||||
void *data; /* the service's own, freed by it in end() */
|
void *data; /* the service's own, freed by it in end() */
|
||||||
};
|
};
|
||||||
struct udpservice {
|
struct udpservice {
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user