Only do one event at a time to a callback.

src/iomux_epoll.c

@@ -162,7 +162,7 @@ static int iomux_epoll_run(struct IOMux *mux_, struct timeval *timeout)
 
         rc = epoll_wait(mux->epoll_fd, mux->pending_events, EPOLL_WAIT_EVENTS, timeout_ms);
 
-    while (rc == -1 && errno == EINTR)
+    } while (rc == -1 && errno == EINTR);
 
     assert(rc >= 0);
     if(rc < 0)
@@ -177,18 +177,27 @@ static int iomux_epoll_run(struct IOMux *mux_, struct timeval *timeout)
 
     for (i = 0; i < rc; i++) {
         struct IOMuxFD *fd = mux->pending_events[i].data.ptr;
-        if (fd != NULL) { //calling _unregister from a callback could have removed it
-            unsigned int events;
+        unsigned int events;
 
-            events = epoll_2_mux_events(mux->pending_events[i].events);
+        events = epoll_2_mux_events(mux->pending_events[i].events);
+
+        //we want to deliver only one event at a time to simplify
+        //application programming.
+        fd = mux->pending_events[i].data.ptr;
+        
+        if (events & MUX_EV_READ && fd != NULL) { 
             assert(fd->callback != NULL);
-            fd->callback(mux_, fd, events);
-            //be sure not to use fd after the callback, it might been removed.
-
-            mux->pending_events[i].data.ptr = NULL; //clear from pending list
-
-            event_cnt++;
+            fd->callback(mux_, fd, MUX_EV_READ);
         }
+
+        //now re-check, in case the IOMuxFD was removed
+        fd = mux->pending_events[i].data.ptr;
+        if (events & MUX_EV_WRITE && fd != NULL) { 
+            assert(fd->callback != NULL);
+            fd->callback(mux_, fd, MUX_EV_WRITE);
+        }
+
+        event_cnt += events != 0;
     }
 
     mux->pending = 0;

src/iomux_select.c

@@ -150,7 +150,7 @@ static int iomux_select_run(struct IOMux *mux_, struct timeval *timeout)
         write_set = mux->master_write_set;
         rc = select(mux->max_fd + 1, &read_set, &write_set, NULL, timeout); //we can use timeout directly
 
-    while (rc == -1 && errno == EINTR)
+    } while (rc == -1 && errno == EINTR);
 
     assert(rc >= 0);
     if (rc < 0)
@@ -164,25 +164,26 @@ static int iomux_select_run(struct IOMux *mux_, struct timeval *timeout)
     for (i = 0; rc > 0 && i < mux->num_descriptors; i++) {
         unsigned int events = 0;
 
-        if (mux->descriptors[i] == NULL) { //callback might have deleted it
-            continue; 
+        //deliver only one event at a time , to simplify application
+        //programming
+        if (mux->descriptors[i] != NULL &&
+                FD_ISSET(mux->descriptors[i]->fd, &read_set)) {
+            rc--;
+            assert(mux->descriptors[i]->callback != NULL);
+            mux->descriptors[i]->callback(mux_, mux->descriptors[i], MUX_EV_READ);
+            events = 1;
         }
 
-        if (FD_ISSET(mux->descriptors[i]->fd, &read_set)) {
+        //need to re-check in case the callback deleted the IOMuxFD
+        if (mux->descriptors[i] != NULL &&
+                FD_ISSET(mux->descriptors[i]->fd, &write_set)) {
             rc--;
-            events |= MUX_EV_READ;
-        }
-
-        if (FD_ISSET(mux->descriptors[i]->fd, &write_set)) {
-            rc--;
-            events |= MUX_EV_WRITE;
+            assert(mux->descriptors[i]->callback != NULL);
+            mux->descriptors[i]->callback(mux_, mux->descriptors[i], MUX_EV_WRITE);
+            events = 1;
         }
         
-        if (events) {
-            assert(mux->descriptors[i]->callback != NULL);
-            mux->descriptors[i]->callback(mux_, mux->descriptors[i], events);
-            event_cnt++;
-        }
+        event_cnt += events;
     }
 
     return event_cnt;