gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

use port_getn instead of port_get

Browse Source
This commit is contained in:
Dave Pacheco 2012-03-26 20:41:58 -07:00 committed by antirez
parent 05da63da0c
commit 6d31277048
1 changed files with 85 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

135
src/ae_evport.c
View File

@ -12,7 +12,7 @@
#include <stdio.h> #include <stdio.h>
int evport_debug = 0; static int evport_debug = 0;
/* /*
* This file implements the ae API using event ports, present on Solaris-based * This file implements the ae API using event ports, present on Solaris-based
@ -21,33 +21,33 @@ int evport_debug = 0;
* events that the consumer is interested in (e.g., POLLIN and POLLOUT). * events that the consumer is interested in (e.g., POLLIN and POLLOUT).
* *
* There's one tricky piece to this implementation: when we return events via * There's one tricky piece to this implementation: when we return events via
* aeApiPoll, the corresponding file descriptor becomes dissociated from the * aeApiPoll, the corresponding file descriptors become dissociated from the
* port. This is necessary because poll events are level-triggered, so if the * port. This is necessary because poll events are level-triggered, so if the
* fd didn't become dissociated, it would immediately fire another event since * fd didn't become dissociated, it would immediately fire another event since
* the underlying state hasn't changed yet. We must reassociate the file * the underlying state hasn't changed yet. We must reassociate the file
* descriptor, but only after we know that our caller has actually read from it. * descriptor, but only after we know that our caller has actually read from it.
* The ae API does not tell us exactly when that happens, but we do know that * The ae API does not tell us exactly when that happens, but we do know that
* it must happen by the time aeApiPoll is called again. Our solution is to * it must happen by the time aeApiPoll is called again. Our solution is to
* keep track of the last fd returned by aeApiPoll and reassociate it next time * keep track of the last fds returned by aeApiPoll and reassociate them next
* aeApiPoll is invoked. * time aeApiPoll is invoked.
* *
* To summarize, in this module, each fd association is EITHER (a) represented * To summarize, in this module, each fd association is EITHER (a) represented
* only via the in-kernel assocation OR (b) represented by pending_fd and * only via the in-kernel assocation OR (b) represented by pending_fds and
* pending_mask. (b) is only true for the last fd we returned from aeApiPoll, * pending_masks. (b) is only true for the last fds we returned from aeApiPoll,
* and only until we enter aeApiPoll again (at which point we restore the * and only until we enter aeApiPoll again (at which point we restore the
* in-kernel association). * in-kernel association).
*
* We currently only return one fd event per call to aeApiPoll. This could be
* extended to return more than one by extending the corresponding pending
* fields and using port_getn().
*/ */
#define MAX_EVENT_BATCHSZ 512
typedef struct aeApiState { typedef struct aeApiState {
int portfd; /* event port */ int portfd; /* event port */
int pending_fd; /* pending fd */ int npending; /* # of pending fds */
int pending_mask; /* pending fd's mask */ int pending_fds[MAX_EVENT_BATCHSZ]; /* pending fds */
int pending_masks[MAX_EVENT_BATCHSZ]; /* pending fds' masks */
} aeApiState; } aeApiState;
static int aeApiCreate(aeEventLoop *eventLoop) { static int aeApiCreate(aeEventLoop *eventLoop) {
int i;
aeApiState *state = zmalloc(sizeof(aeApiState)); aeApiState *state = zmalloc(sizeof(aeApiState));
if (!state) return -1; if (!state) return -1;
@ -57,8 +57,12 @@ static int aeApiCreate(aeEventLoop *eventLoop) {
return -1; return -1;
} }
state->pending_fd = -1; state->npending = 0;
state->pending_mask = AE_NONE;
for (i = 0; i < MAX_EVENT_BATCHSZ; i++) {
state->pending_fds[i] = -1;
state->pending_masks[i] = AE_NONE;
}
eventLoop->apidata = state; eventLoop->apidata = state;
return 0; return 0;
@ -71,11 +75,21 @@ static void aeApiFree(aeEventLoop *eventLoop) {
zfree(state); zfree(state);
} }
static int aeApiLookupPending(aeApiState *state, int fd) {
int i;
for (i = 0; i < state->npending; i++) {
if (state->pending_fds[i] == fd)
return (i);
}
return (-1);
}
/* /*
* Helper function to invoke port_associate for the given fd and mask. * Helper function to invoke port_associate for the given fd and mask.
*/ */
static int aeApiAssociate(const char *where, int portfd, int fd, int mask) static int aeApiAssociate(const char *where, int portfd, int fd, int mask) {
{
int events = 0; int events = 0;
int rv, err; int rv, err;
@ -87,7 +101,8 @@ static int aeApiAssociate(const char *where, int portfd, int fd, int mask)
if (evport_debug) if (evport_debug)
fprintf(stderr, "%s: port_associate(%d, 0x%x) = ", where, fd, events); fprintf(stderr, "%s: port_associate(%d, 0x%x) = ", where, fd, events);
rv = port_associate(portfd, PORT_SOURCE_FD, fd, events, (void *)mask); rv = port_associate(portfd, PORT_SOURCE_FD, fd, events,
(void *)(uintptr_t)mask);
err = errno; err = errno;
if (evport_debug) if (evport_debug)
@ -105,7 +120,7 @@ static int aeApiAssociate(const char *where, int portfd, int fd, int mask)
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) { static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
aeApiState *state = eventLoop->apidata; aeApiState *state = eventLoop->apidata;
int fullmask; int fullmask, pfd;
if (evport_debug) if (evport_debug)
fprintf(stderr, "aeApiAddEvent: fd %d mask 0x%x\n", fd, mask); fprintf(stderr, "aeApiAddEvent: fd %d mask 0x%x\n", fd, mask);
@ -116,8 +131,9 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
* we call port_associate() again. * we call port_associate() again.
*/ */
fullmask = mask | eventLoop->events[fd].mask; fullmask = mask | eventLoop->events[fd].mask;
pfd = aeApiLookupPending(state, fd);
if (fd == state->pending_fd) { if (pfd != -1) {
/* /*
* This fd was recently returned from aeApiPoll. It should be safe to * This fd was recently returned from aeApiPoll. It should be safe to
* assume that the consumer has processed that poll event, but we play * assume that the consumer has processed that poll event, but we play
@ -126,7 +142,7 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
*/ */
if (evport_debug) if (evport_debug)
fprintf(stderr, "aeApiAddEvent: adding to pending fd %d\n", fd); fprintf(stderr, "aeApiAddEvent: adding to pending fd %d\n", fd);
state->pending_mask |= fullmask; state->pending_masks[pfd] |= fullmask;
return 0; return 0;
} }
@ -135,12 +151,14 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) { static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
aeApiState *state = eventLoop->apidata; aeApiState *state = eventLoop->apidata;
int fullmask; int fullmask, pfd;
if (evport_debug) if (evport_debug)
fprintf(stderr, "del fd %d mask 0x%x\n", fd, mask); fprintf(stderr, "del fd %d mask 0x%x\n", fd, mask);
if (fd == state->pending_fd) { pfd = aeApiLookupPending(state, fd);
if (pfd != -1) {
if (evport_debug) if (evport_debug)
fprintf(stderr, "deleting event from pending fd %d\n", fd); fprintf(stderr, "deleting event from pending fd %d\n", fd);
@ -149,10 +167,10 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
* associated with the port. All we need to do is update * associated with the port. All we need to do is update
* pending_mask appropriately. * pending_mask appropriately.
*/ */
state->pending_mask &= ~mask; state->pending_masks[pfd] &= ~mask;
if (state->pending_mask == AE_NONE) if (state->pending_masks[pfd] == AE_NONE)
state->pending_fd = -1; state->pending_fds[pfd] = -1;
return; return;
} }
@ -194,25 +212,32 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) { static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
aeApiState *state = eventLoop->apidata; aeApiState *state = eventLoop->apidata;
struct timespec timeout, *tsp; struct timespec timeout, *tsp;
int mask; int mask, i;
port_event_t event; uint_t nevents;
port_event_t event[MAX_EVENT_BATCHSZ];
/* /*
* If we've returned an fd event before, we must reassociated that fd with * If we've returned fd events before, we must reassociate them with the
* the port now, before calling port_get(). See the block comment at the * port now, before calling port_get(). See the block comment at the top of
* top of this file for an explanation of why. * this file for an explanation of why.
*/ */
if (state->pending_mask != AE_NONE) { for (i = 0; i < state->npending; i++) {
if (aeApiAssociate("aeApiPoll", state->portfd, state->pending_fd, if (state->pending_fds[i] == -1)
state->pending_mask) != 0) { /* This fd has since been deleted. */
continue;
if (aeApiAssociate("aeApiPoll", state->portfd,
state->pending_fds[i], state->pending_masks[i]) != 0) {
/* See aeApiDelEvent for why this case is fatal. */ /* See aeApiDelEvent for why this case is fatal. */
abort(); abort();
} }
state->pending_mask = AE_NONE; state->pending_masks[i] = AE_NONE;
state->pending_fd = -1; state->pending_fds[i] = -1;
} }
state->npending = 0;
if (tvp != NULL) { if (tvp != NULL) {
timeout.tv_sec = tvp->tv_sec; timeout.tv_sec = tvp->tv_sec;
timeout.tv_nsec = tvp->tv_usec * 1000; timeout.tv_nsec = tvp->tv_usec * 1000;
@ -221,7 +246,13 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
tsp = NULL; tsp = NULL;
} }
if (port_get(state->portfd, &event, tsp) == -1) { /*
* port_getn can return with errno == ETIME having returned some events (!).
* So if we get ETIME, we check nevents, too.
*/
nevents = 1;
if (port_getn(state->portfd, event, MAX_EVENT_BATCHSZ, &nevents,
tsp) == -1 && (errno != ETIME || nevents == 0)) {
if (errno == ETIME || errno == EINTR) if (errno == ETIME || errno == EINTR)
return 0; return 0;
@ -230,23 +261,27 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
abort(); abort();
} }
mask = 0; state->npending = nevents;
if (event.portev_events & POLLIN)
mask |= AE_READABLE;
if (event.portev_events & POLLOUT)
mask |= AE_WRITABLE;
eventLoop->fired[0].fd = event.portev_object; for (i = 0; i < nevents; i++) {
eventLoop->fired[0].mask = mask; mask = 0;
if (event[i].portev_events & POLLIN)
mask |= AE_READABLE;
if (event[i].portev_events & POLLOUT)
mask |= AE_WRITABLE;
if (evport_debug) eventLoop->fired[i].fd = event[i].portev_object;
fprintf(stderr, "aeApiPoll: fd %d mask 0x%x\n", event.portev_object, eventLoop->fired[i].mask = mask;
mask);
state->pending_fd = event.portev_object; if (evport_debug)
state->pending_mask = (uintptr_t)event.portev_user; fprintf(stderr, "aeApiPoll: fd %d mask 0x%x\n",
(int)event[i].portev_object, mask);
return 1; state->pending_fds[i] = event[i].portev_object;
state->pending_masks[i] = (uintptr_t)event[i].portev_user;
}
return nevents;
} }
static char *aeApiName(void) { static char *aeApiName(void) {