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 on Windows: AFD polls completed to a port

Browse Source 
A loop waits for sockets with the polls of AFD, the driver under
Windows' sockets, as epoll and kqueue do elsewhere: each socket's poll is
given to \Device\Afd once and completes to the loop's I/O completion
port, which a message from another thread is posted to as well. WSAPoll()
stays the backend with LOOP_POLL and for Watcom and Windows CE.

A poll reports once and has to be made again. Edge-triggered sockets, the
registrations of fibers' sockets, are polled again for what an operation
has found them not to be ready for: loopagain(), called where readiness
known is found gone, a no-op with epoll and kqueue. A poll in flight for
less than wanted is cancelled and made again as it completes; one of a
socket deleted is freed as it completes.

With it the sockets of fibers stay registered on Windows too (SOCKREGS):
the Windows socket functions tell the loop of a socket found not ready
and of one closed, as the Unix ones do, and registrations are kept by the
handle divided by 4.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Vladimir Dubrovin 2026-10-27 14:40:42 +03:00
parent b7037f1159
commit e09e20e71c
4 changed files with 382 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
src/common.c
View File

@ -346,7 +346,10 @@ void (*fiberclosefunc)(SOCKET s) = NULL;
return socket(domain, type, protocol); return socket(domain, type, protocol);
} }
SOCKET WINAPI def_accept(void* state, SOCKET s, struct sockaddr * addr, int * addrlen){ 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){ int WINAPI def_bind(void* state, SOCKET s, const struct sockaddr *addr, int addrlen){
return bind(s, addr, addrlen); return bind(s, addr, addrlen);
@ -355,7 +358,10 @@ void (*fiberclosefunc)(SOCKET s) = NULL;
return listen(s, backlog); return listen(s, backlog);
} }
int WINAPI def_connect(void* state, SOCKET s, const struct sockaddr *name, int namelen){ 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){ int WINAPI def_getpeername(void* state, SOCKET s, struct sockaddr * name, int * namelen){
return getpeername(s, name, namelen); return getpeername(s, name, namelen);
@ -373,22 +379,35 @@ void (*fiberclosefunc)(SOCKET s) = NULL;
return 0; return 0;
} }
int WINAPI def_send(void* state, SOCKET s, const char *msg, int len, int flags){ 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){ 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){ 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){ 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){ int WINAPI def_shutdown(void* state, SOCKET s, int how){
return shutdown(s, how); return shutdown(s, how);
} }
int WINAPI def_closesocket(void* state, SOCKET s){ int WINAPI def_closesocket(void* state, SOCKET s){
if(fiberclosefunc) fiberclosefunc(s);
return closesocket(s); return closesocket(s);
} }
#else #else

309
src/loop.c
View File

@ -38,7 +38,7 @@ uint64_t looptime(void){
/* ---- what wakes the loop ---------------------------------------------- */ /* ---- what wakes the loop ---------------------------------------------- */
#ifndef LOOP_EPOLL #if !defined LOOP_EPOLL && !defined LOOP_AFD
static void setnonblock(SOCKET s){ static void setnonblock(SOCKET s){
#ifdef _WIN32 #ifdef _WIN32
unsigned long ul = 1; unsigned long ul = 1;
@ -50,7 +50,10 @@ static void setnonblock(SOCKET s){
#endif #endif
static int wakeinit(struct loop *l){ 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); int fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
if(fd < 0) return 1; if(fd < 0) return 1;
@ -111,7 +114,8 @@ static int wakeinit(struct loop *l){
} }
static void wakefree(struct loop *l){ static void wakefree(struct loop *l){
#ifdef _WIN32 #ifdef LOOP_AFD
#elif defined _WIN32
closesocket(l->wakerd); closesocket(l->wakerd);
closesocket(l->wakewr); closesocket(l->wakewr);
#else #else
@ -121,7 +125,9 @@ static void wakefree(struct loop *l){
} }
static void wake(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; uint64_t one = 1;
if(write(l->wakewr, &one, sizeof(one)) != sizeof(one)){} if(write(l->wakewr, &one, sizeof(one)) != sizeof(one)){}
#elif defined _WIN32 #elif defined _WIN32
@ -132,9 +138,12 @@ static void wake(struct loop *l){
} }
static void wakedrain(struct loop *l){ static void wakedrain(struct loop *l){
#ifndef LOOP_AFD
char buf[64]; char buf[64];
#endif
#ifdef _WIN32 #ifdef LOOP_AFD
#elif defined _WIN32
while(recv(l->wakerd, buf, sizeof(buf), 0) > 0); while(recv(l->wakerd, buf, sizeof(buf), 0) > 0);
#else #else
while(read(l->wakerd, buf, sizeof(buf)) > 0); while(read(l->wakerd, buf, sizeof(buf)) > 0);
@ -227,11 +236,191 @@ static int pollarm(struct loop *l, struct loopio *io, short events){
} }
#endif #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){ static int arm(struct loop *l, struct loopio *io, short events){
#ifdef LOOP_EPOLL #ifdef LOOP_EPOLL
return eparm(l, io, events); return eparm(l, io, events);
#elif defined LOOP_KQUEUE #elif defined LOOP_KQUEUE
return kqarm(l, io, events); return kqarm(l, io, events);
#elif defined LOOP_AFD
return afdarm(l, io, events);
#else #else
return pollarm(l, io, events); return pollarm(l, io, events);
#endif #endif
@ -486,6 +675,47 @@ int loopinit(struct loop *l){
return 1; 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 #else
if(pollgrow(l)){ if(pollgrow(l)){
wakefree(l); wakefree(l);
@ -504,6 +734,20 @@ int loopinit(struct loop *l){
void loopfree(struct loop *l){ void loopfree(struct loop *l){
#if defined LOOP_EPOLL || defined LOOP_KQUEUE #if defined LOOP_EPOLL || defined LOOP_KQUEUE
close(l->fd); 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 #else
free(l->pfds); free(l->pfds);
free(l->pios); free(l->pios);
@ -519,6 +763,9 @@ int loopadd(struct loop *l, struct loopio *io){
io->armed = 0; io->armed = 0;
io->idx = -1; io->idx = -1;
#ifdef LOOP_AFD
io->afd = NULL;
#endif
io->events = 0; io->events = 0;
return loopmod(l, io, events); 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; 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. */ /* m->func is to be called by the loop: from any thread. */
void looppost(struct loop *l, struct loopmsg *m){ void looppost(struct loop *l, struct loopmsg *m){
int first; int first;
@ -674,6 +932,39 @@ int looponce(struct loop *l, int maxwait){
readyadd(l, io, revents); 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 #else
{ {
unsigned j; unsigned j;
@ -707,6 +998,14 @@ int looponce(struct loop *l, int maxwait){
if(io) io->func(l, io, l->readyev[i]); 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; if(n >= 0) n = l->nready;
l->nready = 0; l->nready = 0;
timersturn(l); timersturn(l);

23
src/loop.h
View File

@ -7,9 +7,10 @@
/* An event loop: one thread waiting for many sockets and timers at once, /* 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 calling back what waits for them. Sockets are waited for with epoll on
Linux, kqueue on the BSDs and macOS, poll() (WSAPoll() on Windows) Linux, kqueue on the BSDs and macOS, the polls of AFD, the driver under
elsewhere or with LOOP_POLL. A loop is run by one thread: only Windows' sockets, completed to an I/O completion port on Windows, poll()
looppost() and loopstop() may be called by another. */ 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 #ifndef _LOOP_H
#define _LOOP_H #define _LOOP_H
@ -18,6 +19,8 @@
#define LOOP_EPOLL #define LOOP_EPOLL
#elif !defined LOOP_POLL && (defined __APPLE__ || defined __FreeBSD__ || defined __NetBSD__ || defined __OpenBSD__ || defined __DragonFly__) #elif !defined LOOP_POLL && (defined __APPLE__ || defined __FreeBSD__ || defined __NetBSD__ || defined __OpenBSD__ || defined __DragonFly__)
#define LOOP_KQUEUE #define LOOP_KQUEUE
#elif !defined LOOP_POLL && defined _WIN32 && !defined __WATCOMC__ && !defined _WINCE
#define LOOP_AFD
#elif !defined LOOP_POLL #elif !defined LOOP_POLL
#define LOOP_POLL #define LOOP_POLL
#endif #endif
@ -60,8 +63,12 @@ struct loopio {
short events; /* POLLIN, POLLOUT */ short events; /* POLLIN, POLLOUT */
short armed; /* what the system waits for */ short armed; /* what the system waits for */
int edge; /* called when it becomes ready, not while it is: 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 */ 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 (*func)(struct loop *l, struct loopio *io, short revents);
void *data; void *data;
}; };
@ -77,13 +84,18 @@ struct loopmsg {
struct loop { struct loop {
#if defined LOOP_EPOLL || defined LOOP_KQUEUE #if defined LOOP_EPOLL || defined LOOP_KQUEUE
int fd; 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 #else
struct pollfd *pfds; struct pollfd *pfds;
struct loopio **pios; struct loopio **pios;
unsigned npfds, maxpfds; unsigned npfds, maxpfds;
#endif #endif
/* what wakes the loop from another thread: an eventfd, the read end /* 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; SOCKET wakerd, wakewr;
_3proxy_mutex_t mutex; _3proxy_mutex_t mutex;
struct loopmsg *posted, **postedtail; struct loopmsg *posted, **postedtail;
@ -109,6 +121,7 @@ void loopfree(struct loop *l);
int loopadd(struct loop *l, struct loopio *io); int loopadd(struct loop *l, struct loopio *io);
int loopmod(struct loop *l, struct loopio *io, short events); int loopmod(struct loop *l, struct loopio *io, short events);
void loopdel(struct loop *l, struct loopio *io); 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 looptimerset(struct loop *l, struct looptimer *t, uint64_t deadline);
void looptimerlater(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); void looptimerdel(struct loop *l, struct looptimer *t);

53
src/mainloop.c
View File

@ -43,15 +43,23 @@ struct fiberslot {
int abandoned; int abandoned;
}; };
/* With epoll and kqueue a socket a fiber waits on is registered once, for /* With epoll, kqueue and AFD a socket a fiber waits on is registered once,
as long as it is open, and the loop remembers what it was reported ready for as long as it is open, and the loop remembers what it was reported
for until an operation finds it not to be: a wait for what a socket is ready for until an operation finds it not to be: a wait for what a socket
ready for costs nothing, as a read or write which can go on is not is ready for costs nothing, as a read or write which can go on is not
preceded by any. */ preceded by any. AFD reports it once, until the loop is told so. */
#if defined LOOP_EPOLL || defined LOOP_KQUEUE #if defined LOOP_EPOLL || defined LOOP_KQUEUE || defined LOOP_AFD
#define SOCKREGS #define SOCKREGS
#endif #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 fiberwait;
struct sockreg { 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. */ /* The registration of a socket of this loop, made the first time. */
static struct sockreg *sockreg(struct looper *lp, SOCKET s){ static struct sockreg *sockreg(struct looper *lp, SOCKET s){
struct sockreg *r; struct sockreg *r;
unsigned x = REGIDX(s);
if((unsigned)s >= lp->nregs){ if(x >= lp->nregs){
unsigned n = lp->nregs? lp->nregs : 1024; unsigned n = lp->nregs? lp->nregs : 1024;
struct sockreg **regs; struct sockreg **regs;
while(n <= (unsigned)s) n *= 2; while(n <= x) n *= 2;
if(!(regs = realloc(lp->regs, n * sizeof(*regs)))) return NULL; if(!(regs = realloc(lp->regs, n * sizeof(*regs)))) return NULL;
memset(regs + lp->nregs, 0, (n - lp->nregs) * sizeof(*regs)); memset(regs + lp->nregs, 0, (n - lp->nregs) * sizeof(*regs));
lp->regs = regs; lp->regs = regs;
lp->nregs = n; lp->nregs = n;
} }
if((r = lp->regs[s])) return r; if((r = lp->regs[x])) return r;
if(!(r = malloc(sizeof(*r)))) return NULL; if(!(r = malloc(sizeof(*r)))) return NULL;
memset(r, 0, sizeof(*r)); memset(r, 0, sizeof(*r));
r->io.fd = s; r->io.fd = s;
@ -271,16 +280,18 @@ static struct sockreg *sockreg(struct looper *lp, SOCKET s){
free(r); free(r);
return NULL; return NULL;
} }
return lp->regs[s] = r; return lp->regs[x] = r;
} }
/* fiberblockfunc: an operation found s not ready for events */ /* fiberblockfunc: an operation found s not ready for events */
static void fiberblocked(SOCKET s, short events){ static void fiberblocked(SOCKET s, short events){
struct looper *lp = curlooper; struct looper *lp = curlooper;
struct sockreg *r;
if(lp && (unsigned)s < lp->nregs && lp->regs[s]){ if(lp && REGIDX(s) < lp->nregs && (r = lp->regs[REGIDX(s)])){
lp->regs[s]->ready &= ~events; r->ready &= ~events;
lp->regs[s]->stale &= ~events; r->stale &= ~events;
loopagain(&lp->l, &r->io, events);
} }
} }
@ -289,13 +300,13 @@ static void fiberclosed(SOCKET s){
struct looper *lp = curlooper; struct looper *lp = curlooper;
struct sockreg *r; 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; int e = errno;
loopdel(&lp->l, &r->io); loopdel(&lp->l, &r->io);
lp->regs[s] = NULL; lp->regs[REGIDX(s)] = NULL;
free(r); 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); have = regs[i]->ready & (want | POLLERR | POLLHUP);
if(have & regs[i]->stale){ if(have & regs[i]->stale){
struct pollfd p; struct pollfd p;
short was = regs[i]->ready;
p.fd = fds[i].fd; p.fd = fds[i].fd;
p.events = POLLIN | POLLOUT; p.events = POLLIN | POLLOUT;
p.revents = 0; p.revents = 0;
if(poll(&p, 1, 0) >= 0) regs[i]->ready = p.revents & READYBITS; 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, /* poll() there tells a socket shut down for sending, by us,
as hung up, the loop does not: its end of reading is 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; regs[i]->ready &= ~POLLHUP;
#endif #endif
/* found not ready for what was reported, waited for or
not */
if(was & ~regs[i]->ready)
loopagain(&lp->l, &regs[i]->io, was & ~regs[i]->ready);
regs[i]->stale = 0; regs[i]->stale = 0;
have = regs[i]->ready & (want | POLLERR | POLLHUP); have = regs[i]->ready & (want | POLLERR | POLLHUP);
} }