bughz/3proxy
1
0
Fork 0
mirror of https://github.com/3proxy/3proxy.git synced 2026-11-05 09:25:49 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Event loops: sockets registered once, readiness kept until not found

Browse Source 
A fiber waiting on a socket registered it with its loop and took it
off again on every wait: two epoll_ctl() calls a wait, some 39 for an
exchange of the mix benchmark, about an eighth of the proxy's time. A
socket is now registered once, edge-triggered for reading and writing,
until it is closed, and the loop keeps what it was reported ready for
until an operation finds it not to be: read, write, accept, connect,
splice() from a pipe and TLS tell it so through fiberblockfunc, the
close through fiberclosefunc. A wait for what a socket is known to be
ready for is answered at once, with no system call and no switch; only
with none of them ready does the fiber wait in its loop. Readiness is
trusted once: answered again with nothing reported or found since, what
was there may have been taken without its end being found, and the
system is asked with poll(). epoll and kqueue only; with poll() each
wait still registers its sockets.

What takes readability for an end rather than reading, the control
connection of a UDP association, the client waiting for a SOCKS BIND,
the server between requests of a kept-alive HTTP connection, asks
first with sockreadable(), a peek below any TLS layer.

kqueue: the end of either side, the peer's or our own shutdown, is
readable or writable, not a hang up, which poll() reports only with
both: kept, it made a wait for the answer after our shutdown return
at once, and sockmap give up with 93.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Vladimir Dubrovin 2026-10-25 12:41:47 +03:00
parent 43f40569d4
commit 7b873b80b1
10 changed files with 263 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
src/common.c
View File

@ -332,6 +332,15 @@ int
#endif
/* Set while clients are served by fibers on event loops, which keep their
sockets registered and remember what each was found ready for: an
operation finding a socket not ready, and a socket about to be closed,
are told to its loop. errno is left as it was. */
void (*fiberblockfunc)(SOCKET s, short events) = NULL;
void (*fiberclosefunc)(SOCKET s) = NULL;
#define BLOCKED(s, ev) do { if(fiberblockfunc && (errno == EAGAIN || errno == EWOULDBLOCK)) \
fiberblockfunc((s), (ev)); } while(0)
#ifdef _WIN32
SOCKET WINAPI def_socket(void* state, int domain, int type, int protocol){
return socket(domain, type, protocol);
@ -387,7 +396,10 @@ int
return socket(domain, type, protocol);
}
SOCKET def_accept(void* state, SOCKET s, struct sockaddr * addr, socklen_t* addrlen){
return accept(s, addr, addrlen);
SOCKET r = accept(s, addr, addrlen);
if(r == INVALID_SOCKET) BLOCKED(s, POLLIN);
return r;
}
int def_bind(void* state, SOCKET s, const struct sockaddr *addr, socklen_t addrlen){
return bind(s, addr, addrlen);
@ -402,7 +414,10 @@ int
return listen(s, backlog);
}
int def_connect(void* state, SOCKET s, const struct sockaddr *name, socklen_t namelen){
return connect(s, name, namelen);
int r = connect(s, name, namelen);
if(r && errno == EINPROGRESS && fiberblockfunc) fiberblockfunc(s, POLLOUT);
return r;
}
int def_getsockopt(void* state, SOCKET s, int level, int optname, void * optval, socklen_t * optlen){
return getsockopt(s, level, optname, optval, optlen);
@ -415,6 +430,7 @@ int
return 0;
}
/* a peer gone is an error of the send, not SIGPIPE, where the system can
say so for a call: SIGPIPE is ignored too, for TLS and splice() */
#ifdef MSG_NOSIGNAL
@ -423,21 +439,34 @@ int
#define SENDFLAGS 0
#endif
ssize_t def_send(void* state, SOCKET s, const void *msg, size_t len, int flags){
return send(s, msg, len, flags | SENDFLAGS);
ssize_t r = send(s, msg, len, flags | SENDFLAGS);
if(r < 0) BLOCKED(s, POLLOUT);
return r;
}
ssize_t def_sendto(void* state, SOCKET s, const void *msg, size_t len, int flags, const struct sockaddr *to, socklen_t tolen){
return sendto(s, msg, len, flags | SENDFLAGS, to, tolen);
ssize_t r = sendto(s, msg, len, flags | SENDFLAGS, to, tolen);
if(r < 0) BLOCKED(s, POLLOUT);
return r;
}
ssize_t def_recv(void* state, SOCKET s, void *buf, size_t len, int flags){
return recv(s, buf, len, flags);
ssize_t r = recv(s, buf, len, flags);
if(r < 0) BLOCKED(s, POLLIN);
return r;
}
ssize_t def_recvfrom(void* state, SOCKET s, void * buf, size_t len, int flags, struct sockaddr * from, socklen_t* fromlen){
return recvfrom(s, buf, len, flags, from, fromlen);
ssize_t r = recvfrom(s, buf, len, flags, from, fromlen);
if(r < 0) BLOCKED(s, POLLIN);
return r;
}
int def_shutdown(void* state, SOCKET s, int how){
return shutdown(s, how);
}
int def_closesocket(void* state, SOCKET s){
if(fiberclosefunc) fiberclosefunc(s);
return close(s);
}
#endif
@ -463,6 +492,22 @@ struct sockfuncs so = {
def_closesocket
};
/* Whether s, found readable, is: data or its end there. A fiber's wait
may answer with what a socket was reported ready for, until an
operation finds it not to be: whatever takes readability for an end,
rather than reading, asks first. A peek below any layer over the socket,
which a TLS read would consume. */
int sockreadable(SOCKET s){
char c;
if(!fiberblockfunc) return 1;
if(recv(s, &c, 1, MSG_PEEK) < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)){
fiberblockfunc(s, POLLIN);
return 0;
}
return 1;
}
/* Waits as poll() does for fds of sockets made with sf, once their layers
know of nothing ready already. A layer may have a socket waited for
other than as asked, a TLS read for it to be writable: whatever it gets

11
src/loop.c
View File

@ -148,13 +148,14 @@ static void wakedrain(struct loop *l){
static int kqarm(struct loop *l, struct loopio *io, short events){
struct kevent ch[2];
int n = 0;
unsigned short add = EV_ADD | (io->edge? EV_CLEAR : 0);
if((events & POLLIN) && !(io->armed & POLLIN))
EV_SET(ch + n++, io->fd, EVFILT_READ, EV_ADD, 0, 0, io);
EV_SET(ch + n++, io->fd, EVFILT_READ, add, 0, 0, io);
else if(!(events & POLLIN) && (io->armed & POLLIN))
EV_SET(ch + n++, io->fd, EVFILT_READ, EV_DELETE, 0, 0, io);
if((events & POLLOUT) && !(io->armed & POLLOUT))
EV_SET(ch + n++, io->fd, EVFILT_WRITE, EV_ADD, 0, 0, io);
EV_SET(ch + n++, io->fd, EVFILT_WRITE, add, 0, 0, io);
else if(!(events & POLLOUT) && (io->armed & POLLOUT))
EV_SET(ch + n++, io->fd, EVFILT_WRITE, EV_DELETE, 0, 0, io);
if(n && kevent(l->fd, ch, n, NULL, 0, NULL) && (events & ~io->armed)) return 1;
@ -172,6 +173,7 @@ static int eparm(struct loop *l, struct loopio *io, short events){
ev.data.ptr = io;
if(events & POLLIN) ev.events |= EPOLLIN;
if(events & POLLOUT) ev.events |= EPOLLOUT;
if(io->edge) ev.events |= EPOLLET;
/* not registered while nothing is waited for: an error or a hang up
would be reported all the same */
if(!events) op = EPOLL_CTL_DEL;
@ -665,7 +667,10 @@ int looponce(struct loop *l, int maxwait){
if(ev[i].flags & EV_ERROR) revents |= POLLERR;
else if(ev[i].filter == EVFILT_READ) revents |= POLLIN;
else if(ev[i].filter == EVFILT_WRITE) revents |= POLLOUT;
if(ev[i].flags & EV_EOF) revents |= POLLHUP;
/* The end of either side, the peer's or our own shutdown,
is no hang up, which poll() reports only with both:
readable or writable, the read or the write telling
what it is */
readyadd(l, io, revents);
}
}

2
src/loop.h
View File

@ -59,6 +59,8 @@ struct loopio {
SOCKET fd;
short events; /* POLLIN, POLLOUT */
short armed; /* what the system waits for */
int edge; /* called when it becomes ready, not while it is:
epoll and kqueue only */
int idx; /* poll(): its place in the array, -1 if none */
void (*func)(struct loop *l, struct loopio *io, short revents);
void *data;

169
src/mainloop.c
View File

@ -40,11 +40,32 @@ struct fiberslot {
struct fiberslot *next;
};
/* With epoll and kqueue a socket a fiber waits on is registered once, for
as long as it is open, and the loop remembers what it was reported ready
for until an operation finds it not to be: a wait for what a socket is
ready for costs nothing, as a read or write which can go on is not
preceded by any. */
#if defined LOOP_EPOLL || defined LOOP_KQUEUE
#define SOCKREGS
#endif
struct fiberwait;
struct sockreg {
struct loopio io; /* edge-triggered, for reading and writing */
short ready; /* reported ready for, not found not to be since */
short want; /* what the fiber waiting on it waits for */
unsigned char spins; /* waits answered at once since it was reported */
struct fiberwait *w; /* the wait of the fiber waiting on it */
};
struct looper {
struct loop l;
uint64_t threadid;
struct fiberslot *idle; /* the loop's thread only */
unsigned nidle;
struct sockreg **regs; /* by descriptor, the loop's thread only */
unsigned nregs;
};
/* what srvspawn() posts to a loop */
@ -185,6 +206,143 @@ static void fiberwaittimer(struct loop *l, struct looptimer *t){
}
}
#ifdef SOCKREGS
#define READYBITS (POLLIN | POLLOUT | POLLERR | POLLHUP)
static void sockregio(struct loop *l, struct loopio *io, short revents){
struct sockreg *r = (struct sockreg *)io;
struct fiberwait *w = r->w;
r->ready |= revents;
r->spins = 0;
/* nothing of r is touched after the fiber runs: it may close it */
if(w && !w->woken && (revents & (r->want | POLLERR | POLLHUP))){
w->woken = 1;
fiberrun(w->lp, w->f);
}
}
/* The registration of a socket of this loop, made the first time. */
static struct sockreg *sockreg(struct looper *lp, SOCKET s){
struct sockreg *r;
if((unsigned)s >= lp->nregs){
unsigned n = lp->nregs? lp->nregs : 1024;
struct sockreg **regs;
while(n <= (unsigned)s) n *= 2;
if(!(regs = realloc(lp->regs, n * sizeof(*regs)))) return NULL;
memset(regs + lp->nregs, 0, (n - lp->nregs) * sizeof(*regs));
lp->regs = regs;
lp->nregs = n;
}
if((r = lp->regs[s])) return r;
if(!(r = malloc(sizeof(*r)))) return NULL;
memset(r, 0, sizeof(*r));
r->io.fd = s;
r->io.events = POLLIN | POLLOUT;
r->io.edge = 1;
r->io.func = sockregio;
r->io.data = r;
if(loopadd(&lp->l, &r->io)){
free(r);
return NULL;
}
return lp->regs[s] = r;
}
/* fiberblockfunc: an operation found s not ready for events */
static void fiberblocked(SOCKET s, short events){
struct looper *lp = curlooper;
if(lp && (unsigned)s < lp->nregs && lp->regs[s]){
lp->regs[s]->ready &= ~events;
lp->regs[s]->spins = 0;
}
}
/* fiberclosefunc: s is closed, a socket of the same number another */
static void fiberclosed(SOCKET s){
struct looper *lp = curlooper;
struct sockreg *r;
if(lp && (unsigned)s < lp->nregs && (r = lp->regs[s])){
int e = errno;
loopdel(&lp->l, &r->io);
lp->regs[s] = NULL;
free(r);
errno = e;
}
}
/* fiberpoll() with the sockets registered: answered at once with what they
are known to be ready for; otherwise the fiber waits until they are, or
the time is out. What a socket was reported ready for is trusted once:
answered at once again with nothing reported or found since, what was
there may have been taken without its end being found, or by an
operation past the socket functions, and the system is asked. */
static int fiberpollregs(struct looper *lp, struct pollfd *fds, unsigned nfds, int timeout){
struct sockreg *ownregs[FIBERWAITS], **regs = ownregs;
struct fiberwait w;
unsigned i;
int res = 0;
if(nfds > FIBERWAITS && !(regs = malloc(nfds * sizeof(*regs)))) return -1;
for(i = 0; i < nfds; i++){
short want = fds[i].events & (POLLIN | POLLOUT), have;
fds[i].revents = 0;
regs[i] = NULL;
if(fds[i].fd == INVALID_SOCKET || (int)fds[i].fd < 0) continue;
if(!(regs[i] = sockreg(lp, fds[i].fd))){
res = -1;
goto DONE;
}
have = regs[i]->ready & (want | POLLERR | POLLHUP);
if(have && regs[i]->spins){
struct pollfd p;
p.fd = fds[i].fd;
p.events = POLLIN | POLLOUT;
p.revents = 0;
if(poll(&p, 1, 0) >= 0) regs[i]->ready = p.revents & READYBITS;
regs[i]->spins = 0;
have = regs[i]->ready & (want | POLLERR | POLLHUP);
}
fds[i].revents = have;
if(have) res++;
}
if(res){
for(i = 0; i < nfds; i++) if(fds[i].revents) regs[i]->spins++;
goto DONE;
}
memset(&w, 0, sizeof(w));
w.lp = lp;
w.f = fibercurrent();
for(i = 0; i < nfds; i++) if(regs[i]){
regs[i]->w = &w;
regs[i]->want = fds[i].events & (POLLIN | POLLOUT);
}
if(timeout > 0){
w.t.func = fiberwaittimer;
w.t.data = &w;
looptimerset(&lp->l, &w.t, looptime() + (uint64_t)timeout);
}
if(nfds || timeout > 0) fiberyield();
if(timeout > 0) looptimerdel(&lp->l, &w.t);
for(i = 0; i < nfds; i++) if(regs[i]){
regs[i]->w = NULL;
regs[i]->spins = 0;
fds[i].revents = regs[i]->ready & (fds[i].events & (POLLIN | POLLOUT) | POLLERR | POLLHUP);
if(fds[i].revents) res++;
}
DONE:
if(regs != ownregs) free(regs);
return res;
}
#endif
/* sockpoll() in a fiber of a loop: what poll() would return, waiting in the
loop. -2 elsewhere, or for no wait at all, for sockpoll() to poll. */
static int fiberpoll(struct pollfd *fds, unsigned nfds, int timeout){
@ -196,6 +354,9 @@ static int fiberpoll(struct pollfd *fds, unsigned nfds, int timeout){
int res = 0;
if(!lp || !fibercurrent() || !timeout) return -2;
#ifdef SOCKREGS
return fiberpollregs(lp, fds, nfds, timeout);
#endif
memset(&w, 0, sizeof(w));
w.lp = lp;
w.f = fibercurrent();
@ -313,7 +474,13 @@ static int loopsstart(void){
}
if(!i) loopsfailed = 1;
nloopers = i;
if(i) fiberpollfunc = fiberpoll;
if(i){
#ifdef SOCKREGS
fiberblockfunc = fiberblocked;
fiberclosefunc = fiberclosed;
#endif
fiberpollfunc = fiberpoll;
}
_3proxy_mutex_unlock(&loopers_mutex);
return !i;
}

2
src/proxy.c
View File

@ -571,7 +571,7 @@ for(;;){
if(res<=0) {
RETURN(555);
}
if((fds[1].revents & (POLLIN|POLLHUP|POLLERR|POLLNVAL))) {
if((fds[1].revents & (POLLIN|POLLHUP|POLLERR|POLLNVAL)) && sockreadable(param->remsock)) {
if(param->transparent || (!param->redirected && param->redirtype == R_HTTP)) RETURN(555);
ckeepalive = 0;
param->srv->so._shutdown(param->sostate, param->remsock, SHUT_RDWR);

3
src/proxy.h
View File

@ -334,6 +334,9 @@ int nsbroken(void);
SOCKET nssocket(struct srvparam *srv, struct sockfuncs *sf, void *state, int side, int domain, int type, int protocol);
int sockpoll(struct sockfuncs *sf, void *state, struct pollfd *fds, unsigned nfds, int timeout);
extern int (*fiberpollfunc)(struct pollfd *fds, unsigned nfds, int timeout);
extern void (*fiberblockfunc)(SOCKET s, short events);
extern void (*fiberclosefunc)(SOCKET s);
int sockreadable(SOCKET s);
void nonblock(SOCKET sock);
int sysresolves(int mode);
int acceptagain(void);

4
src/sockmap.c
View File

@ -382,6 +382,8 @@ log("done send to client from buf");
log("send to server from pipe");
#endif
res = splice(pipecli[0], NULL, param->remsock, NULL, MIN(MAXSPLICE, inclientpipe), SPLICE_F_NONBLOCK|SPLICE_F_MOVE);
/* from a pipe holding data: the socket is full */
if(res < 0 && errno == EAGAIN && fiberblockfunc) fiberblockfunc(param->remsock, POLLOUT);
#ifdef WITHLOG
log("server from pipe splice finished\n");
#if WITHLOG > 1
@ -419,6 +421,8 @@ log(logbuf);
log("send to client from pipe");
#endif
res = splice(pipesrv[0], NULL, param->clisock, NULL, MIN(MAXSPLICE, inserverpipe), SPLICE_F_NONBLOCK|SPLICE_F_MOVE);
/* from a pipe holding data: the socket is full */
if(res < 0 && errno == EAGAIN && fiberblockfunc) fiberblockfunc(param->clisock, POLLOUT);
#ifdef WITHLOG
log("client from pipe splice finished\n");
#if WITHLOG > 1

14
src/socks.c
View File

@ -356,13 +356,23 @@ fflush(stderr);
fds[0].fd = param->remsock;
fds[1].fd = param->clisock;
fds[0].events = fds[1].events = POLLIN;
s = INVALID_SOCKET;
do {
res = sockpoll(&param->srv->so, param->sostate, fds, 2, param->srv->timeouts[CONNECTION_L] * 1000);
if (res < 1 || fds[1].revents) break;
if (res < 1) break;
/* the client sending or closing ends the wait */
if (fds[1].revents && sockreadable(param->clisock)) {
res = 0;
break;
}
if (!fds[0].revents) {
errno = EAGAIN;
continue;
}
sasize = sizeof(param->sinsr);
s = param->srv->so._accept(param->sostate, param->remsock, (struct sockaddr *)&param->sinsr, &sasize);
} while (s == INVALID_SOCKET && acceptagain());
if (res < 1 || fds[1].revents) {
if (res < 1) {
res = 460;
break;
}

12
src/ssl.c
View File

@ -144,6 +144,8 @@ static ssize_t ssl_send(void *state, SOCKET s, const void *msg, size_t len, int
err = SSL_get_error((SSL *)((ssl_conn*)sslq->conn)->ssl, res);
if (err == SSL_ERROR_WANT_WRITE || err == SSL_ERROR_WANT_READ){
sslq->wwant = (err == SSL_ERROR_WANT_READ)? POLLIN : POLLOUT;
/* TLS found the socket not ready, past the socket functions */
if(fiberblockfunc) fiberblockfunc(s, sslq->wwant);
seterrno3(EAGAIN);
return -1;
}
@ -170,6 +172,8 @@ static ssize_t ssl_sendto(void *state, SOCKET s, const void *msg, size_t len, in
err = SSL_get_error((SSL *)((ssl_conn*)sslq->conn)->ssl, res);
if (err == SSL_ERROR_WANT_WRITE || err == SSL_ERROR_WANT_READ){
sslq->wwant = (err == SSL_ERROR_WANT_READ)? POLLIN : POLLOUT;
/* TLS found the socket not ready, past the socket functions */
if(fiberblockfunc) fiberblockfunc(s, sslq->wwant);
seterrno3(EAGAIN);
return -1;
}
@ -195,6 +199,8 @@ static ssize_t ssl_recvfrom(void *state, SOCKET s, void *msg, size_t len, int f
err = SSL_get_error((SSL *)((ssl_conn*)sslq->conn)->ssl, res);
if (err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE) {
sslq->rwant = (err == SSL_ERROR_WANT_WRITE)? POLLOUT : POLLIN;
/* TLS found the socket not ready, past the socket functions */
if(fiberblockfunc) fiberblockfunc(s, sslq->rwant);
seterrno3(EAGAIN);
return -1;
}
@ -219,6 +225,8 @@ static ssize_t ssl_recv(void *state, SOCKET s, void *msg, size_t len, int flags)
err = SSL_get_error((SSL *)((ssl_conn*)sslq->conn)->ssl, res);
if (err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE) {
sslq->rwant = (err == SSL_ERROR_WANT_WRITE)? POLLOUT : POLLIN;
/* TLS found the socket not ready, past the socket functions */
if(fiberblockfunc) fiberblockfunc(s, sslq->rwant);
seterrno3(EAGAIN);
return -1;
}
@ -314,10 +322,12 @@ SSL_CONN ssl_handshake_to_server(SOCKET s, char * hostname, SSL_CONFIG *config,
if(sslerr == SSL_ERROR_WANT_READ){
fds[0].fd = s;
fds[0].events = POLLIN;
if(fiberblockfunc) fiberblockfunc(s, POLLIN);
}
else if(sslerr == SSL_ERROR_WANT_WRITE){
fds[0].fd = s;
fds[0].events = POLLOUT;
if(fiberblockfunc) fiberblockfunc(s, POLLOUT);
}
else break;
if(sockpoll(&sso, sso.state, fds, 1, config->timeout*1000) <= 0 || !(fds[0].revents & (POLLOUT|POLLIN))) break;
@ -389,10 +399,12 @@ SSL_CONN ssl_handshake_to_client(SOCKET s, SSL_CONFIG *config, X509 *server_cert
if(sslerr == SSL_ERROR_WANT_READ){
fds[0].fd = s;
fds[0].events = POLLIN;
if(fiberblockfunc) fiberblockfunc(s, POLLIN);
}
else if(sslerr == SSL_ERROR_WANT_WRITE){
fds[0].fd = s;
fds[0].events = POLLOUT;
if(fiberblockfunc) fiberblockfunc(s, POLLOUT);
}
else break;
if(sockpoll(&sso, sso.state, fds, 1, config->timeout*1000) <= 0 || !(fds[0].revents & (POLLOUT|POLLIN))) break;

4
src/udpsockmap.c
View File

@ -412,8 +412,8 @@ int udpsockmap(struct clientparam *param, int timeo)
(struct sockaddr *)&sin, SASIZE(&sin));
}
if ((p.ctrl >= 0 && p.fds[p.ctrl].revents) ||
(p.ctrlsrv >= 0 && p.fds[p.ctrlsrv].revents)) return 0;
if ((p.ctrl >= 0 && p.fds[p.ctrl].revents && sockreadable(param->ctrlsock)) ||
(p.ctrlsrv >= 0 && p.fds[p.ctrlsrv].revents && sockreadable(param->ctrlsocksrv))) return 0;
}
return 0;
}