refactor internal AE api, avoiding eventLoop

Signed-off-by: Jim Brunner <brunnerj@amazon.com>

Author: JimB123

src/ae.c

@@ -73,6 +73,10 @@
         assert(pthread_mutex_unlock(&(eventLoop)->poll_mutex) == 0); \
     }
 
+/* Regardless of the flags used in AE, the only flags understood by the backend
+ * implementations are AE_READABLE & AE_WRITABLE. */
+#define BACKEND_MASK(mask) ((mask) & (AE_READABLE | AE_WRITABLE))
+
 aeEventLoop *aeCreateEventLoop(int setsize) {
     aeEventLoop *eventLoop;
     int i;
@@ -98,7 +102,7 @@ aeEventLoop *aeCreateEventLoop(int setsize) {
     pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
     if (pthread_mutex_init(&eventLoop->poll_mutex, &attr) != 0) goto err;
 
-    if (aeApiCreate(eventLoop) == -1) goto err;
+    if ((eventLoop->apidata = aeApiCreate(setsize)) == NULL) goto err;
     /* Events with mask == AE_NONE are not set. So let's initialize the
      * vector with it. */
     for (i = 0; i < setsize; i++) eventLoop->events[i].mask = AE_NONE;
@@ -144,7 +148,7 @@ int aeResizeSetSize(aeEventLoop *eventLoop, int setsize) {
 
     if (setsize == eventLoop->setsize) goto done;
     if (eventLoop->maxfd >= setsize) goto err;
-    if (aeApiResize(eventLoop, setsize) == -1) goto err;
+    if (aeApiResize(eventLoop->apidata, setsize) == -1) goto err;
 
     eventLoop->events = zrealloc(eventLoop->events, sizeof(aeFileEvent) * setsize);
     eventLoop->fired = zrealloc(eventLoop->fired, sizeof(aeFiredEvent) * setsize);
@@ -163,7 +167,7 @@ int aeResizeSetSize(aeEventLoop *eventLoop, int setsize) {
 }
 
 void aeDeleteEventLoop(aeEventLoop *eventLoop) {
-    aeApiFree(eventLoop);
+    aeApiFree(eventLoop->apidata);
     zfree(eventLoop->events);
     zfree(eventLoop->fired);
 
@@ -192,7 +196,10 @@ int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask, aeFileProc *proc
     }
     aeFileEvent *fe = &eventLoop->events[fd];
 
-    if (aeApiAddEvent(eventLoop, fd, mask) == -1) goto done;
+    int backend_add_mask = BACKEND_MASK(mask) & ~fe->mask; // just the meaningful additions
+    if (backend_add_mask) {
+        if (aeApiAddEvent(eventLoop->apidata, fd, BACKEND_MASK(fe->mask), backend_add_mask) == -1) goto done;
+    }
     fe->mask |= mask;
     if (mask & AE_READABLE) fe->rfileProc = proc;
     if (mask & AE_WRITABLE) fe->wfileProc = proc;
@@ -220,7 +227,8 @@ void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask) {
     /* Only remove attached events */
     mask = mask & fe->mask;
 
-    fe->mask = fe->mask & (~mask);
+    int old_mask = fe->mask;
+    fe->mask = fe->mask & ~mask;
     if (fd == eventLoop->maxfd && fe->mask == AE_NONE) {
         /* Update the max fd */
         int j;
@@ -233,10 +241,9 @@ void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask) {
     /* Check whether there are events to be removed.
      * Note: user may remove the AE_BARRIER without
      * touching the actual events. */
-    if (mask & (AE_READABLE | AE_WRITABLE)) {
-        /* Must be invoked after the eventLoop mask is modified,
-         * which is required by evport and epoll */
-        aeApiDelEvent(eventLoop, fd, mask);
+    int backend_del_mask = BACKEND_MASK(mask); // just the meaningful deletions
+    if (backend_del_mask) {
+        aeApiDelEvent(eventLoop->apidata, fd, BACKEND_MASK(old_mask), backend_del_mask);
     }
 
 done:
@@ -390,7 +397,7 @@ static int processTimeEvents(aeEventLoop *eventLoop) {
 int aePoll(aeEventLoop *eventLoop, struct timeval *tvp) {
     AE_LOCK(eventLoop);
 
-    int ret = aeApiPoll(eventLoop, tvp);
+    int ret = aeApiPoll(eventLoop->apidata, eventLoop->fired, eventLoop->events, eventLoop->setsize, eventLoop->maxfd, tvp);
 
     AE_UNLOCK(eventLoop);
     return ret;
@@ -449,7 +456,7 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags) {
             }
             /* Call the multiplexing API, will return only on timeout or when
              * some event fires. */
-            numevents = aeApiPoll(eventLoop, tvp);
+            numevents = aeApiPoll(eventLoop->apidata, eventLoop->fired, eventLoop->events, eventLoop->setsize, eventLoop->maxfd, tvp);
         }
 
         /* Don't process file events if not requested. */

src/ae.h

@@ -65,6 +65,14 @@
 struct timeval; /* forward declaration */
 struct aeEventLoop;
 
+/* Opaque per-backend polling state (epoll/kqueue/evport/select).
+ *
+ * The concrete struct aeApiState is defined privately by each polling backend
+ * and its layout varies between them. ae.c only ever holds and passes a typed
+ * pointer to it, so the forward declaration here lets the event loop and the
+ * inner aeApi* interface use "aeApiState *" instead of an untyped "void *". */
+typedef struct aeApiState aeApiState;
+
 /* Types and data structures */
 typedef void aeFileProc(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);
 typedef long long aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);
@@ -109,7 +117,7 @@ typedef struct aeEventLoop {
     aeFiredEvent *fired; /* Fired events */
     aeTimeEvent *timeEventHead;
     int stop;
-    void *apidata; /* This is used for polling API specific data */
+    aeApiState *apidata; /* Polling API specific state (owned by the backend) */
     aeBeforeSleepProc *beforesleep;
     aeAfterSleepProc *aftersleep;
     aeCustomPollProc *custompoll;

src/ae_epoll.c

@@ -36,63 +36,55 @@ typedef struct aeApiState {
     struct epoll_event *events;
 } aeApiState;
 
-static int aeApiCreate(aeEventLoop *eventLoop) {
+static aeApiState *aeApiCreate(int setsize) {
     aeApiState *state = zmalloc(sizeof(aeApiState));
 
-    if (!state) return -1;
-    state->events = zmalloc(sizeof(struct epoll_event) * eventLoop->setsize);
+    if (!state) return NULL;
+    state->events = zmalloc(sizeof(struct epoll_event) * setsize);
     if (!state->events) {
         zfree(state);
-        return -1;
+        return NULL;
     }
     state->epfd = epoll_create(1024); /* 1024 is just a hint for the kernel */
     if (state->epfd == -1) {
         zfree(state->events);
         zfree(state);
-        return -1;
+        return NULL;
     }
     anetCloexec(state->epfd);
-    eventLoop->apidata = state;
-    return 0;
+    return state;
 }
 
-static int aeApiResize(aeEventLoop *eventLoop, int setsize) {
-    aeApiState *state = eventLoop->apidata;
-
+static int aeApiResize(aeApiState *state, int setsize) {
     state->events = zrealloc(state->events, sizeof(struct epoll_event) * setsize);
     return 0;
 }
 
-static void aeApiFree(aeEventLoop *eventLoop) {
-    aeApiState *state = eventLoop->apidata;
-
+static void aeApiFree(aeApiState *state) {
     close(state->epfd);
     zfree(state->events);
     zfree(state);
 }
 
-static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
-    aeApiState *state = eventLoop->apidata;
+static int aeApiAddEvent(aeApiState *state, int fd, int curr_mask, int add_mask) {
     struct epoll_event ee = {0}; /* avoid valgrind warning */
     /* If the fd was already monitored for some event, we need a MOD
      * operation. Otherwise, we need an ADD operation. */
-    int op = eventLoop->events[fd].mask == AE_NONE ? EPOLL_CTL_ADD : EPOLL_CTL_MOD;
+    int op = (curr_mask & (AE_READABLE | AE_WRITABLE)) ? EPOLL_CTL_MOD : EPOLL_CTL_ADD;
 
     ee.events = 0;
-    mask |= eventLoop->events[fd].mask; /* Merge old events */
+    int mask = curr_mask | add_mask;
     if (mask & AE_READABLE) ee.events |= EPOLLIN;
     if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
     ee.data.fd = fd;
     if (epoll_ctl(state->epfd, op, fd, &ee) == -1) return -1;
     return 0;
 }
 
-static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
-    aeApiState *state = eventLoop->apidata;
+static void aeApiDelEvent(aeApiState *state, int fd, int curr_mask, int del_mask) {
     struct epoll_event ee = {0}; /* avoid valgrind warning */
 
-    /* We rely on the fact that our caller has already updated the mask in the eventLoop. */
-    mask = eventLoop->events[fd].mask;
+    int mask = curr_mask & ~del_mask;
 
     ee.events = 0;
     if (mask & AE_READABLE) ee.events |= EPOLLIN;
@@ -107,11 +99,12 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
     }
 }
 
-static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
-    aeApiState *state = eventLoop->apidata;
+static int aeApiPoll(aeApiState *state, aeFiredEvent *fired, aeFileEvent *events, int setsize, int maxfd, struct timeval *tvp) {
+    AE_NOTUSED(events);
+    AE_NOTUSED(maxfd);
     int retval, numevents = 0;
 
-    retval = epoll_wait(state->epfd, state->events, eventLoop->setsize,
+    retval = epoll_wait(state->epfd, state->events, setsize,
                         tvp ? (tvp->tv_sec * 1000 + (tvp->tv_usec + 999) / 1000) : -1);
     if (retval > 0) {
         int j;
@@ -125,8 +118,8 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
             if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
             if (e->events & EPOLLERR) mask |= AE_WRITABLE | AE_READABLE;
             if (e->events & EPOLLHUP) mask |= AE_WRITABLE | AE_READABLE;
-            eventLoop->fired[j].fd = e->data.fd;
-            eventLoop->fired[j].mask = mask;
+            fired[j].fd = e->data.fd;
+            fired[j].mask = mask;
         }
     } else if (retval == -1 && errno != EINTR) {
         panic("aeApiPoll: epoll_wait, %s", strerror(errno));

src/ae_evport.c

@@ -71,15 +71,16 @@ typedef struct aeApiState {
     int pending_masks[MAX_EVENT_BATCHSZ]; /* pending fds' masks */
 } aeApiState;
 
-static int aeApiCreate(aeEventLoop *eventLoop) {
+static aeApiState *aeApiCreate(int setsize) {
+    AE_NOTUSED(setsize);
     int i;
     aeApiState *state = zmalloc(sizeof(aeApiState));
-    if (!state) return -1;
+    if (!state) return NULL;
 
     state->portfd = port_create();
     if (state->portfd == -1) {
         zfree(state);
-        return -1;
+        return NULL;
     }
     anetCloexec(state->portfd);
 
@@ -90,20 +91,17 @@ static int aeApiCreate(aeEventLoop *eventLoop) {
         state->pending_masks[i] = AE_NONE;
     }
 
-    eventLoop->apidata = state;
-    return 0;
+    return state;
 }
 
-static int aeApiResize(aeEventLoop *eventLoop, int setsize) {
-    (void)eventLoop;
-    (void)setsize;
+static int aeApiResize(aeApiState *state, int setsize) {
+    AE_NOTUSED(state);
+    AE_NOTUSED(setsize);
     /* Nothing to resize here. */
     return 0;
 }
 
-static void aeApiFree(aeEventLoop *eventLoop) {
-    aeApiState *state = eventLoop->apidata;
-
+static void aeApiFree(aeApiState *state) {
     close(state->portfd);
     zfree(state);
 }
@@ -144,19 +142,16 @@ static int aeApiAssociate(const char *where, int portfd, int fd, int mask) {
     return rv;
 }
 
-static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
-    aeApiState *state = eventLoop->apidata;
-    int fullmask, pfd;
-
-    if (evport_debug) fprintf(stderr, "aeApiAddEvent: fd %d mask 0x%x\n", fd, mask);
+static int aeApiAddEvent(aeApiState *state, int fd, int curr_mask, int add_mask) {
+    if (evport_debug) fprintf(stderr, "aeApiAddEvent: fd %d mask 0x%x\n", fd, add_mask);
 
     /*
      * Since port_associate's "events" argument replaces any existing events, we
      * must be sure to include whatever events are already associated when
      * we call port_associate() again.
      */
-    fullmask = mask | eventLoop->events[fd].mask;
-    pfd = aeApiLookupPending(state, fd);
+    int mask = curr_mask | add_mask;
+    int pfd = aeApiLookupPending(state, fd);
 
     if (pfd != -1) {
         /*
@@ -166,18 +161,17 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
          * re-associated as usual when aeApiPoll is called again.
          */
         if (evport_debug) fprintf(stderr, "aeApiAddEvent: adding to pending fd %d\n", fd);
-        state->pending_masks[pfd] |= fullmask;
+        state->pending_masks[pfd] |= mask;
         return 0;
     }
 
-    return (aeApiAssociate("aeApiAddEvent", state->portfd, fd, fullmask));
+    return (aeApiAssociate("aeApiAddEvent", state->portfd, fd, mask));
 }
 
-static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
-    aeApiState *state = eventLoop->apidata;
+static void aeApiDelEvent(aeApiState *state, int fd, int curr_mask, int del_mask) {
     int fullmask, pfd;
 
-    if (evport_debug) fprintf(stderr, "del fd %d mask 0x%x\n", fd, mask);
+    if (evport_debug) fprintf(stderr, "del fd %d mask 0x%x\n", fd, del_mask);
 
     pfd = aeApiLookupPending(state, fd);
 
@@ -189,7 +183,7 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
          * associated with the port.  All we need to do is update
          * pending_mask appropriately.
          */
-        state->pending_masks[pfd] &= ~mask;
+        state->pending_masks[pfd] &= ~del_mask;
 
         if (state->pending_masks[pfd] == AE_NONE) state->pending_fds[pfd] = -1;
 
@@ -199,13 +193,10 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
     /*
      * The fd is currently associated with the port.  Like with the add case
      * above, we must look at the full mask for the file descriptor before
-     * updating that association.  We don't have a good way of knowing what the
-     * events are without looking into the eventLoop state directly.  We rely on
-     * the fact that our caller has already updated the mask in the eventLoop.
      */
 
-    fullmask = eventLoop->events[fd].mask;
-    if (fullmask == AE_NONE) {
+    int mask = curr_mask & ~del_mask;
+    if (mask == AE_NONE) {
         /*
          * We're removing *all* events, so use port_dissociate to remove the
          * association completely.  Failure here indicates a bug.
@@ -216,7 +207,7 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
             perror("aeApiDelEvent: port_dissociate");
             abort(); /* will not return */
         }
-    } else if (aeApiAssociate("aeApiDelEvent", state->portfd, fd, fullmask) != 0) {
+    } else if (aeApiAssociate("aeApiDelEvent", state->portfd, fd, mask) != 0) {
         /*
          * ENOMEM is a potentially transient condition, but the kernel won't
          * generally return it unless things are really bad.  EAGAIN indicates
@@ -228,8 +219,10 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
     }
 }
 
-static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
-    aeApiState *state = eventLoop->apidata;
+static int aeApiPoll(aeApiState *state, aeFiredEvent *fired, aeFileEvent *events, int setsize, int maxfd, struct timeval *tvp) {
+    AE_NOTUSED(events);
+    AE_NOTUSED(setsize);
+    AE_NOTUSED(maxfd);
     struct timespec timeout, *tsp;
     uint_t mask, i;
     uint_t nevents;
@@ -282,8 +275,8 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
         if (event[i].portev_events & POLLIN) mask |= AE_READABLE;
         if (event[i].portev_events & POLLOUT) mask |= AE_WRITABLE;
 
-        eventLoop->fired[i].fd = event[i].portev_object;
-        eventLoop->fired[i].mask = mask;
+        fired[i].fd = event[i].portev_object;
+        fired[i].mask = mask;
 
         if (evport_debug) fprintf(stderr, "aeApiPoll: fd %d mask 0x%x\n", (int)event[i].portev_object, mask);