Initial import of libucore

src/iomux_epoll.c

@@ -0,0 +1,224 @@
+#include <sys/epoll.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <errno.h>
+#include <string.h>
+#include <assert.h>
+#include "iomux_impl.h"
+
+//epoll (linux specific) IO mux. 
+//We stuff the struct IOMuxFD pointer in the event.data.ptr slot
+//for the kernel to keep track of, so we don't have to.
+///we do not use edge triggered mode
+
+//number of events we wait for in one go
+#define EPOLL_WAIT_EVENTS (256)
+
+struct IOMuxEpoll {
+    //number of file descriptors we're watching
+    size_t num_descriptors;
+    //the epoll file descriptor
+    int epoll_fd;
+
+    struct epoll_event pending_events[EPOLL_WAIT_EVENTS];
+    //number of pending events
+    int pending;
+};
+
+static inline uint32_t mux_2_epoll_events(unsigned int events)
+{
+    uint32_t epoll_events = 0;
+
+    if(events & MUX_EV_READ) 
+        epoll_events |= EPOLLIN;
+
+    if(events & MUX_EV_WRITE) 
+        epoll_events |= EPOLLOUT;
+
+    return epoll_events;
+}
+
+static inline unsigned int epoll_2_mux_events(uint32_t events)
+{
+    unsigned int mux_events = 0;
+
+    //epoll always waits for EPOLLHUP and EPOLLERR , we map
+    //those to read events which will hopefully do the right thing
+    if(events & (EPOLLIN | EPOLLHUP | EPOLLERR))
+        mux_events |= MUX_EV_READ;
+
+    if(events & EPOLLOUT) 
+        mux_events |= MUX_EV_WRITE;
+
+    return mux_events;
+}
+
+static int iomux_epoll_update_events(struct IOMux *mux_, struct IOMuxFD *fd)
+{
+    struct IOMuxEpoll *mux = mux_->instance;
+    struct epoll_event e_event = {0, {0}};
+    int rc;
+   
+    //if we EPOLL_CTL_MOD with events of 0, epoll
+    //will still wait for HUP/ERR , which can lead to
+    //strange things. (e.g. updating fd->what to 0 on a pipe
+    //in order to try to "pause" reading can lead to missing or false
+    //read events.
+    assert(fd->what != 0);
+    if(fd->what == 0)
+        return EINVAL;
+
+
+    e_event.events = mux_2_epoll_events(fd->what);
+    e_event.data.ptr = fd;
+    
+    rc = epoll_ctl(mux->epoll_fd, EPOLL_CTL_MOD, fd->fd, &e_event);
+    assert(rc == 0);
+    if(rc != 0)
+        return errno;
+
+    return 0;
+}
+
+static int iomux_epoll_register_fd(struct IOMux *mux_, struct IOMuxFD *fd)
+{
+    struct IOMuxEpoll *mux = mux_->instance;
+    struct epoll_event e_event = {0, {0}};
+    int rc;
+
+    if(fd->what == 0 || fd->callback == NULL)
+        return EINVAL;
+
+    e_event.events = mux_2_epoll_events(fd->what);
+    e_event.data.ptr = fd;
+
+    rc = epoll_ctl(mux->epoll_fd, EPOLL_CTL_ADD, fd->fd, &e_event);
+    assert(rc == 0);
+    if(rc != 0) {
+        return errno;
+    }
+
+    mux->num_descriptors++;
+
+    return 0;
+}
+static int iomux_epoll_unregister_fd(struct IOMux *mux_, struct IOMuxFD *fd)
+{
+    struct IOMuxEpoll *mux = mux_->instance;
+    struct epoll_event e_event = {0,{ 0}};
+    int i;
+    int rc;
+
+    rc = epoll_ctl(mux->epoll_fd, EPOLL_CTL_DEL, fd->fd, &e_event);
+    assert(rc == 0);
+    if(rc != 0)
+        return errno;
+
+    //if the fd is pending, remove it from pending_events
+    //This ensures that it's safe to unregister a descriptor
+    //that is currently pending a callback
+    for(i = 0; i < mux->pending; i++) {
+        struct IOMuxFD *fdi = mux->pending_events[i].data.ptr;
+        if(fdi == NULL)
+            continue;
+
+        if(fd == fdi) {
+            mux->pending_events[i].data.ptr = NULL;
+            break;
+        }
+    }
+ 
+    mux->num_descriptors--;
+
+    return rc;
+}
+static inline int timeval_to_ms(const struct timeval *t)
+{
+    int ms = t->tv_sec * 1000;
+
+    ms += (t->tv_usec + 999) / 1000; //round upwards
+
+    return ms;
+}
+
+static int iomux_epoll_run(struct IOMux *mux_, struct timeval *timeout)
+{
+    int i;
+    struct IOMuxEpoll *mux = mux_->instance;
+    int rc;
+    int timeout_ms;
+    int event_cnt;
+
+    if(mux->num_descriptors == 0 && timeout == NULL)
+        return 0;
+
+    //epoll takes the timeout in miliseconds
+    if(timeout)
+        timeout_ms = timeval_to_ms(timeout);
+    else 
+        timeout_ms = -1;
+    
+    rc = epoll_wait(mux->epoll_fd, mux->pending_events, EPOLL_WAIT_EVENTS, timeout_ms);
+    assert(rc >= 0);
+    if(rc < 0)
+        return errno;
+
+    mux->pending = rc;
+
+    event_cnt = iomux_timers_run(mux_); //process timers
+
+    if(rc == 0) //Just the timeout
+        return event_cnt;
+
+    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;
+
+            events = epoll_2_mux_events(mux->pending_events[i].events);
+            assert(fd->callback != NULL);
+            fd->callback(mux_, fd, events);
+            //be sure not to use fd after the callback, it might been removed.
+
+            event_cnt++;
+        }
+    }
+
+    mux->pending = 0;
+
+    return event_cnt;
+}
+
+static void iomux_epoll_delete(struct IOMux *mux)
+{
+    assert(mux != NULL && mux->instance != NULL);
+    if(mux != NULL && mux->instance != NULL) {
+        struct IOMuxEpoll *mux_epoll = mux->instance;
+        close(mux_epoll->epoll_fd);
+        free(mux_epoll);
+        mux->instance = NULL;
+    }
+}
+    
+int iomux_epoll_init(struct IOMux *mux)
+{
+    struct IOMuxEpoll *mux_epoll = calloc(1, sizeof *mux_epoll);
+    if(mux_epoll == NULL)
+        return ENOMEM;
+
+    mux_epoll->epoll_fd = epoll_create(128);
+    if(mux_epoll->epoll_fd == -1) {
+        free(mux_epoll);
+        return errno;
+    }
+
+    mux->instance           = mux_epoll;
+    mux->run_impl           = iomux_epoll_run;
+    mux->delete_impl        = iomux_epoll_delete;
+    mux->register_fd_impl   = iomux_epoll_register_fd;
+    mux->unregister_fd_impl = iomux_epoll_unregister_fd;
+    mux->update_events_impl = iomux_epoll_update_events;
+
+    return 0;
+}
+