diff --git a/src/common.c b/src/common.c index 9cb9a41..af3dd1b 100644 --- a/src/common.c +++ b/src/common.c @@ -346,7 +346,10 @@ void (*fiberclosefunc)(SOCKET s) = NULL; return socket(domain, type, protocol); } SOCKET WINAPI def_accept(void* state, SOCKET s, struct sockaddr * addr, int * addrlen){ - return accept(s, addr, addrlen); + SOCKET r = accept(s, addr, addrlen); + + if(r == INVALID_SOCKET) BLOCKED(s, POLLIN); + return r; } int WINAPI def_bind(void* state, SOCKET s, const struct sockaddr *addr, int addrlen){ return bind(s, addr, addrlen); @@ -355,7 +358,10 @@ void (*fiberclosefunc)(SOCKET s) = NULL; return listen(s, backlog); } int WINAPI def_connect(void* state, SOCKET s, const struct sockaddr *name, int namelen){ - return connect(s, name, namelen); + int r = connect(s, name, namelen); + + if(r) BLOCKED(s, POLLOUT); + return r; } int WINAPI def_getpeername(void* state, SOCKET s, struct sockaddr * name, int * namelen){ return getpeername(s, name, namelen); @@ -373,22 +379,35 @@ void (*fiberclosefunc)(SOCKET s) = NULL; return 0; } int WINAPI def_send(void* state, SOCKET s, const char *msg, int len, int flags){ - return send(s, msg, len, flags); + int r = send(s, msg, len, flags); + + if(r < 0) BLOCKED(s, POLLOUT); + return r; } int WINAPI def_sendto(void* state, SOCKET s, const char *msg, int len, int flags, const struct sockaddr *to, int tolen){ - return sendto(s, msg, len, flags, to, tolen); + int r = sendto(s, msg, len, flags, to, tolen); + + if(r < 0) BLOCKED(s, POLLOUT); + return r; } int WINAPI def_recv(void* state, SOCKET s, char *buf, int len, int flags){ - return recv(s, buf, len, flags); + int r = recv(s, buf, len, flags); + + if(r < 0) BLOCKED(s, POLLIN); + return r; } int WINAPI def_recvfrom(void* state, SOCKET s, char * buf, int len, int flags, struct sockaddr * from, int * fromlen){ - return recvfrom(s, buf, len, flags, from, fromlen); + int r = recvfrom(s, buf, len, flags, from, fromlen); + + if(r < 0) BLOCKED(s, POLLIN); + return r; } int WINAPI def_shutdown(void* state, SOCKET s, int how){ return shutdown(s, how); } int WINAPI def_closesocket(void* state, SOCKET s){ + if(fiberclosefunc) fiberclosefunc(s); return closesocket(s); } #else diff --git a/src/loop.c b/src/loop.c index 23a3ddd..d561ce2 100644 --- a/src/loop.c +++ b/src/loop.c @@ -38,7 +38,7 @@ uint64_t looptime(void){ /* ---- what wakes the loop ---------------------------------------------- */ -#ifndef LOOP_EPOLL +#if !defined LOOP_EPOLL && !defined LOOP_AFD static void setnonblock(SOCKET s){ #ifdef _WIN32 unsigned long ul = 1; @@ -50,7 +50,10 @@ static void setnonblock(SOCKET s){ #endif static int wakeinit(struct loop *l){ -#ifdef LOOP_EPOLL +#ifdef LOOP_AFD + /* a completion posted to the port */ + l->wakerd = l->wakewr = INVALID_SOCKET; +#elif defined LOOP_EPOLL int fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); if(fd < 0) return 1; @@ -111,7 +114,8 @@ static int wakeinit(struct loop *l){ } static void wakefree(struct loop *l){ -#ifdef _WIN32 +#ifdef LOOP_AFD +#elif defined _WIN32 closesocket(l->wakerd); closesocket(l->wakewr); #else @@ -121,7 +125,9 @@ static void wakefree(struct loop *l){ } static void wake(struct loop *l){ -#ifdef LOOP_EPOLL +#ifdef LOOP_AFD + PostQueuedCompletionStatus(l->iocp, 0, 0, NULL); +#elif defined LOOP_EPOLL uint64_t one = 1; if(write(l->wakewr, &one, sizeof(one)) != sizeof(one)){} #elif defined _WIN32 @@ -132,9 +138,12 @@ static void wake(struct loop *l){ } static void wakedrain(struct loop *l){ +#ifndef LOOP_AFD char buf[64]; +#endif -#ifdef _WIN32 +#ifdef LOOP_AFD +#elif defined _WIN32 while(recv(l->wakerd, buf, sizeof(buf), 0) > 0); #else while(read(l->wakerd, buf, sizeof(buf)) > 0); @@ -227,11 +236,191 @@ static int pollarm(struct loop *l, struct loopio *io, short events){ } #endif +#ifdef LOOP_AFD +/* AFD's own, as the driver has them: not in the SDK */ +#define IOCTL_AFD_POLL 0x00012024 +#define AFD_POLL_RECEIVE 0x0001 +#define AFD_POLL_SEND 0x0004 +#define AFD_POLL_DISCONNECT 0x0008 +#define AFD_POLL_ABORT 0x0010 +#define AFD_POLL_LOCAL_CLOSE 0x0020 +#define AFD_POLL_ACCEPT 0x0080 +#define AFD_POLL_CONNECT_FAIL 0x0100 +#define AFDCANCELLED ((LONG)0xC0000120) + +#ifndef SIO_BASE_HANDLE +#define SIO_BASE_HANDLE 0x48000022 +#endif +#ifndef SIO_BSP_HANDLE +#define SIO_BSP_HANDLE 0x4800001B +#endif +#ifndef SIO_BSP_HANDLE_SELECT +#define SIO_BSP_HANDLE_SELECT 0x4800001C +#endif +#ifndef SIO_BSP_HANDLE_POLL +#define SIO_BSP_HANDLE_POLL 0x4800001D +#endif + +struct afdhandle { + HANDLE handle; + ULONG events; + LONG status; +}; + +struct afdinfo { + LARGE_INTEGER timeout; + ULONG n; + ULONG exclusive; + struct afdhandle h[1]; +}; + +/* A poll of a socket, the loop's: with the driver until it completes, + which may be after the socket is deleted. */ +struct afdpoll { + OVERLAPPED ov; /* what the completion points to */ + struct afdinfo info; + struct loopio *io; /* NULL once deleted: freed as it completes */ + HANDLE base; /* the socket below any layered provider */ + short polled; /* what the poll with the driver is for */ + short reported; /* edge: reported, not found otherwise since */ + int pending; /* with the driver */ + int cancelled; /* to be made again for more as it completes */ + int closed; /* the socket closed under it */ +}; + +/* The socket the driver knows, under any layered provider: a provider + which hides it may still give it for polls or select(). */ +static HANDLE afdbase(SOCKET s){ + static const DWORD ioctls[] = {SIO_BASE_HANDLE, SIO_BSP_HANDLE_POLL, SIO_BSP_HANDLE_SELECT, SIO_BSP_HANDLE}; + SOCKET b; + DWORD n; + unsigned i; + + for(i = 0; i < sizeof(ioctls)/sizeof(ioctls[0]); i++){ + b = INVALID_SOCKET; + if(WSAIoctl(s, ioctls[i], NULL, 0, &b, sizeof(b), &n, NULL, NULL) != SOCKET_ERROR && + b != INVALID_SOCKET) + return (HANDLE)b; + } + return (HANDLE)s; +} + +static int afdsubmit(struct loop *l, struct afdpoll *p, short want){ + ULONG ev = AFD_POLL_LOCAL_CLOSE | AFD_POLL_ABORT | AFD_POLL_CONNECT_FAIL; + DWORD n; + + if(want & POLLIN) ev |= AFD_POLL_RECEIVE | AFD_POLL_ACCEPT | AFD_POLL_DISCONNECT; + if(want & POLLOUT) ev |= AFD_POLL_SEND; + memset(&p->ov, 0, sizeof(p->ov)); + memset(&p->info, 0, sizeof(p->info)); + p->info.timeout.QuadPart = 0x7fffffffffffffffLL; + p->info.n = 1; + p->info.h[0].handle = p->base; + p->info.h[0].events = ev; + /* completed at once or not, the completion comes to the port */ + if(!DeviceIoControl(l->afd, IOCTL_AFD_POLL, &p->info, sizeof(p->info), + &p->info, sizeof(p->info), &n, &p->ov) && GetLastError() != ERROR_IO_PENDING) + return 1; + p->pending = 1; + p->polled = want; + l->afdpending++; + return 0; +} + +/* The poll of io made for what it waits for: edge-triggered, for what + was not reported, nothing once it failed. One for less in flight is + cancelled, to be made again as it completes. errno is kept. */ +static int afdupdate(struct loop *l, struct loopio *io, short events){ + struct afdpoll *p = io->afd; + short want = events; + DWORD e = GetLastError(); + int res = 0; + + if(p->closed) return 0; + if(io->edge) want = (p->reported & (POLLERR | POLLHUP))? 0 : want & ~p->reported; + if(p->pending){ + if((want & ~p->polled) && !p->cancelled){ + CancelIoEx(l->afd, &p->ov); + p->cancelled = 1; + } + } + else if(want) res = afdsubmit(l, p, want); + SetLastError(e); + return res; +} + +/* io not waited for: its poll freed now, or as it completes */ +static void afddrop(struct loop *l, struct loopio *io){ + struct afdpoll *p = io->afd; + DWORD e = GetLastError(); + + if(!p) return; + io->afd = NULL; + p->io = NULL; + if(!p->pending) free(p); + else if(!p->cancelled){ + CancelIoEx(l->afd, &p->ov); + p->cancelled = 1; + } + SetLastError(e); +} + +static int afdarm(struct loop *l, struct loopio *io, short events){ + struct afdpoll *p; + + if(!events){ + afddrop(l, io); + io->armed = 0; + return 0; + } + if(!io->afd){ + DWORD e = GetLastError(); + + if(!(p = malloc(sizeof(*p)))) return 1; + memset(p, 0, sizeof(*p)); + p->base = afdbase(io->fd); + SetLastError(e); + p->io = io; + io->afd = p; + } + if(afdupdate(l, io, events)){ + afddrop(l, io); + io->armed = 0; + return 1; + } + io->armed = events; + return 0; +} + +/* What a completed poll found, as poll() has it. */ +static short afdrevents(struct afdpoll *p){ + LONG st = (LONG)p->ov.Internal; + ULONG ev; + short r = 0; + + if(st == AFDCANCELLED) return 0; + if(st < 0) return POLLERR; + if(p->info.n < 1) return 0; + ev = p->info.h[0].events; + if(ev & AFD_POLL_LOCAL_CLOSE){ + p->closed = 1; + return 0; + } + if(ev & (AFD_POLL_RECEIVE | AFD_POLL_ACCEPT | AFD_POLL_DISCONNECT)) r |= POLLIN; + if(ev & AFD_POLL_SEND) r |= POLLOUT; + if(ev & AFD_POLL_ABORT) r |= POLLHUP; + if(ev & AFD_POLL_CONNECT_FAIL) r |= POLLERR; + return r; +} +#endif + static int arm(struct loop *l, struct loopio *io, short events){ #ifdef LOOP_EPOLL return eparm(l, io, events); #elif defined LOOP_KQUEUE return kqarm(l, io, events); +#elif defined LOOP_AFD + return afdarm(l, io, events); #else return pollarm(l, io, events); #endif @@ -486,6 +675,47 @@ int loopinit(struct loop *l){ return 1; } } +#elif defined LOOP_AFD + { + typedef struct { + USHORT len, max; + PWSTR buf; + } afdstr; + typedef struct { + ULONG len; + HANDLE root; + afdstr *name; + ULONG attrs; + PVOID sd, sqos; + } afdattr; + typedef struct { + ULONG_PTR status, info; + } afdiosb; + typedef LONG (NTAPI *ntcreatefile)(HANDLE *, ACCESS_MASK, afdattr *, afdiosb *, + PLARGE_INTEGER, ULONG, ULONG, ULONG, ULONG, PVOID, ULONG); + static WCHAR name[] = L"\\Device\\Afd\\3proxy"; + afdstr str; + afdattr attr; + afdiosb iosb; + ntcreatefile create = (ntcreatefile)(void (*)(void))GetProcAddress(GetModuleHandleW(L"ntdll.dll"), "NtCreateFile"); + + str.len = sizeof(name) - sizeof(WCHAR); + str.max = sizeof(name); + str.buf = name; + memset(&attr, 0, sizeof(attr)); + attr.len = sizeof(attr); + attr.name = &str; + l->afd = l->iocp = NULL; + if(!create || + !(l->iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 1)) || + create(&l->afd, SYNCHRONIZE, &attr, &iosb, NULL, 0, FILE_SHARE_READ | FILE_SHARE_WRITE, 1 /* FILE_OPEN */, 0, NULL, 0) < 0 || + !CreateIoCompletionPort(l->afd, l->iocp, 0, 0) || + !SetFileCompletionNotificationModes(l->afd, FILE_SKIP_SET_EVENT_ON_HANDLE)){ + if(l->afd && l->afd != INVALID_HANDLE_VALUE) CloseHandle(l->afd); + if(l->iocp) CloseHandle(l->iocp); + return 1; + } + } #else if(pollgrow(l)){ wakefree(l); @@ -504,6 +734,20 @@ int loopinit(struct loop *l){ void loopfree(struct loop *l){ #if defined LOOP_EPOLL || defined LOOP_KQUEUE close(l->fd); +#elif defined LOOP_AFD + /* the polls of what was deleted, cancelled, freed as they complete */ + while(l->afdpending){ + OVERLAPPED_ENTRY ev[LOOPREADY]; + ULONG n, i; + + if(!GetQueuedCompletionStatusEx(l->iocp, ev, LOOPREADY, &n, 1000, FALSE)) break; + for(i = 0; i < n; i++) if(ev[i].lpOverlapped){ + l->afdpending--; + free(ev[i].lpOverlapped); + } + } + CloseHandle(l->afd); + CloseHandle(l->iocp); #else free(l->pfds); free(l->pios); @@ -519,6 +763,9 @@ int loopadd(struct loop *l, struct loopio *io){ io->armed = 0; io->idx = -1; +#ifdef LOOP_AFD + io->afd = NULL; +#endif io->events = 0; return loopmod(l, io, events); } @@ -540,6 +787,17 @@ void loopdel(struct loop *l, struct loopio *io){ for(i = 0; i < l->nready; i++) if(l->ready[i] == io) l->ready[i] = NULL; } +/* An operation found io not ready for events, waited for edge-triggered: + reported again once it is. With epoll and kqueue it is anyway. */ +void loopagain(struct loop *l, struct loopio *io, short events){ +#ifdef LOOP_AFD + if(io->afd && (io->afd->reported & events)){ + io->afd->reported &= ~events; + afdupdate(l, io, io->events); + } +#endif +} + /* m->func is to be called by the loop: from any thread. */ void looppost(struct loop *l, struct loopmsg *m){ int first; @@ -674,6 +932,39 @@ int looponce(struct loop *l, int maxwait){ readyadd(l, io, revents); } } +#elif defined LOOP_AFD + { + OVERLAPPED_ENTRY ev[LOOPREADY]; + ULONG got = 0; + + if(GetQueuedCompletionStatusEx(l->iocp, ev, LOOPREADY, &got, timeo < 0? INFINITE : (DWORD)timeo, FALSE)) + n = (int)got; + else n = (GetLastError() == WAIT_TIMEOUT)? 0 : -1; + for(i = 0; i < n; i++){ + struct afdpoll *p = (struct afdpoll *)ev[i].lpOverlapped; + struct loopio *io; + short revents; + + if(!p){ + woken = 1; + continue; + } + p->pending = 0; + p->cancelled = 0; + l->afdpending--; + if(!(io = p->io)){ + free(p); + continue; + } + revents = afdrevents(p); + if(io->edge) p->reported |= revents; + revents &= io->events | POLLERR | POLLHUP; + /* polled again once called, as poll() would be: now, if + it is not to be */ + if(revents) readyadd(l, io, revents); + else if(afdupdate(l, io, io->events)) readyadd(l, io, POLLERR); + } + } #else { unsigned j; @@ -707,6 +998,14 @@ int looponce(struct loop *l, int maxwait){ if(io) io->func(l, io, l->readyev[i]); } +#ifdef LOOP_AFD + /* what was called and is still waited for, polled again */ + for(i = 0; i < l->nready; i++){ + struct loopio *io = l->ready[i]; + + if(io && io->afd) afdupdate(l, io, io->events); + } +#endif if(n >= 0) n = l->nready; l->nready = 0; timersturn(l); diff --git a/src/loop.h b/src/loop.h index f12e7b9..e32428b 100644 --- a/src/loop.h +++ b/src/loop.h @@ -7,9 +7,10 @@ /* An event loop: one thread waiting for many sockets and timers at once, calling back what waits for them. Sockets are waited for with epoll on - Linux, kqueue on the BSDs and macOS, poll() (WSAPoll() on Windows) - elsewhere or with LOOP_POLL. A loop is run by one thread: only - looppost() and loopstop() may be called by another. */ + Linux, kqueue on the BSDs and macOS, the polls of AFD, the driver under + Windows' sockets, completed to an I/O completion port on Windows, poll() + elsewhere or with LOOP_POLL (WSAPoll() on Windows). A loop is run by one + thread: only looppost() and loopstop() may be called by another. */ #ifndef _LOOP_H #define _LOOP_H @@ -18,6 +19,8 @@ #define LOOP_EPOLL #elif !defined LOOP_POLL && (defined __APPLE__ || defined __FreeBSD__ || defined __NetBSD__ || defined __OpenBSD__ || defined __DragonFly__) #define LOOP_KQUEUE +#elif !defined LOOP_POLL && defined _WIN32 && !defined __WATCOMC__ && !defined _WINCE +#define LOOP_AFD #elif !defined LOOP_POLL #define LOOP_POLL #endif @@ -60,8 +63,12 @@ struct loopio { 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 */ + epoll, kqueue and AFD only; with AFD once + for each readiness, until loopagain() */ int idx; /* poll(): its place in the array, -1 if none */ +#ifdef LOOP_AFD + struct afdpoll *afd; /* its poll, NULL if none */ +#endif void (*func)(struct loop *l, struct loopio *io, short revents); void *data; }; @@ -77,13 +84,18 @@ struct loopmsg { struct loop { #if defined LOOP_EPOLL || defined LOOP_KQUEUE int fd; +#elif defined LOOP_AFD + HANDLE iocp; /* where the polls complete, and the loop is woken */ + HANDLE afd; /* the driver, opened for the polls */ + unsigned afdpending; /* polls with the driver */ #else struct pollfd *pfds; struct loopio **pios; unsigned npfds, maxpfds; #endif /* what wakes the loop from another thread: an eventfd, the read end - of a pipe, or on Windows a UDP socket sending to itself */ + of a pipe, or on Windows a TCP connection to itself; with AFD a + completion posted instead */ SOCKET wakerd, wakewr; _3proxy_mutex_t mutex; struct loopmsg *posted, **postedtail; @@ -109,6 +121,7 @@ void loopfree(struct loop *l); int loopadd(struct loop *l, struct loopio *io); int loopmod(struct loop *l, struct loopio *io, short events); void loopdel(struct loop *l, struct loopio *io); +void loopagain(struct loop *l, struct loopio *io, short events); void looptimerset(struct loop *l, struct looptimer *t, uint64_t deadline); void looptimerlater(struct loop *l, struct looptimer *t, uint64_t deadline); void looptimerdel(struct loop *l, struct looptimer *t); diff --git a/src/mainloop.c b/src/mainloop.c index 05e4547..dd9b5eb 100644 --- a/src/mainloop.c +++ b/src/mainloop.c @@ -43,15 +43,23 @@ struct fiberslot { int abandoned; }; -/* 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 +/* With epoll, kqueue and AFD 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. AFD reports it once, until the loop is told so. */ +#if defined LOOP_EPOLL || defined LOOP_KQUEUE || defined LOOP_AFD #define SOCKREGS #endif +/* where a socket's registration is kept: Windows' sockets are handles, + multiples of 4 */ +#ifdef _WIN32 +#define REGIDX(s) ((unsigned)((s) >> 2)) +#else +#define REGIDX(s) ((unsigned)(s)) +#endif + struct fiberwait; struct sockreg { @@ -248,18 +256,19 @@ static void sockregio(struct loop *l, struct loopio *io, short revents){ /* The registration of a socket of this loop, made the first time. */ static struct sockreg *sockreg(struct looper *lp, SOCKET s){ struct sockreg *r; + unsigned x = REGIDX(s); - if((unsigned)s >= lp->nregs){ + if(x >= lp->nregs){ unsigned n = lp->nregs? lp->nregs : 1024; struct sockreg **regs; - while(n <= (unsigned)s) n *= 2; + while(n <= x) 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 = lp->regs[x])) return r; if(!(r = malloc(sizeof(*r)))) return NULL; memset(r, 0, sizeof(*r)); r->io.fd = s; @@ -271,16 +280,18 @@ static struct sockreg *sockreg(struct looper *lp, SOCKET s){ free(r); return NULL; } - return lp->regs[s] = r; + return lp->regs[x] = r; } /* fiberblockfunc: an operation found s not ready for events */ static void fiberblocked(SOCKET s, short events){ struct looper *lp = curlooper; + struct sockreg *r; - if(lp && (unsigned)s < lp->nregs && lp->regs[s]){ - lp->regs[s]->ready &= ~events; - lp->regs[s]->stale &= ~events; + if(lp && REGIDX(s) < lp->nregs && (r = lp->regs[REGIDX(s)])){ + r->ready &= ~events; + r->stale &= ~events; + loopagain(&lp->l, &r->io, events); } } @@ -289,13 +300,13 @@ static void fiberclosed(SOCKET s){ struct looper *lp = curlooper; struct sockreg *r; - if(lp && (unsigned)s < lp->nregs && (r = lp->regs[s])){ + if(lp && REGIDX(s) < lp->nregs && (r = lp->regs[REGIDX(s)])){ int e = errno; loopdel(&lp->l, &r->io); - lp->regs[s] = NULL; + lp->regs[REGIDX(s)] = NULL; free(r); - errno = e; + seterrno3(e); } } @@ -325,17 +336,23 @@ static int fiberpollregs(struct looper *lp, struct pollfd *fds, unsigned nfds, i have = regs[i]->ready & (want | POLLERR | POLLHUP); if(have & regs[i]->stale){ struct pollfd p; + short was = regs[i]->ready; p.fd = fds[i].fd; p.events = POLLIN | POLLOUT; p.revents = 0; if(poll(&p, 1, 0) >= 0) regs[i]->ready = p.revents & READYBITS; -#ifdef LOOP_KQUEUE +#if defined LOOP_KQUEUE || defined LOOP_AFD /* poll() there tells a socket shut down for sending, by us, as hung up, the loop does not: its end of reading is - readable */ + readable. WSAPoll() tells the peer's end so too, which + AFD reports readable once asked again */ regs[i]->ready &= ~POLLHUP; #endif + /* found not ready for what was reported, waited for or + not */ + if(was & ~regs[i]->ready) + loopagain(&lp->l, ®s[i]->io, was & ~regs[i]->ready); regs[i]->stale = 0; have = regs[i]->ready & (want | POLLERR | POLLHUP); }