diff --git a/src/common.c b/src/common.c index 666af55..5016d14 100644 --- a/src/common.c +++ b/src/common.c @@ -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 diff --git a/src/loop.c b/src/loop.c index 3fc3393..23a3ddd 100644 --- a/src/loop.c +++ b/src/loop.c @@ -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); } } diff --git a/src/loop.h b/src/loop.h index 02dc95d..f12e7b9 100644 --- a/src/loop.h +++ b/src/loop.h @@ -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; diff --git a/src/mainloop.c b/src/mainloop.c index 36d2737..b5d6701 100644 --- a/src/mainloop.c +++ b/src/mainloop.c @@ -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; } diff --git a/src/proxy.c b/src/proxy.c index 257485e..ab2076d 100644 --- a/src/proxy.c +++ b/src/proxy.c @@ -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); diff --git a/src/proxy.h b/src/proxy.h index de5e66b..998e7f1 100644 --- a/src/proxy.h +++ b/src/proxy.h @@ -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); diff --git a/src/sockmap.c b/src/sockmap.c index 2e6309e..c2b547f 100644 --- a/src/sockmap.c +++ b/src/sockmap.c @@ -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 diff --git a/src/socks.c b/src/socks.c index e5086ff..865d59c 100644 --- a/src/socks.c +++ b/src/socks.c @@ -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(¶m->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 *)¶m->sinsr, &sasize); } while (s == INVALID_SOCKET && acceptagain()); - if (res < 1 || fds[1].revents) { + if (res < 1) { res = 460; break; } diff --git a/src/ssl.c b/src/ssl.c index a3abef2..25f588f 100644 --- a/src/ssl.c +++ b/src/ssl.c @@ -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; diff --git a/src/udpsockmap.c b/src/udpsockmap.c index 2fce3cf..ae4dd11 100644 --- a/src/udpsockmap.c +++ b/src/udpsockmap.c @@ -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; }