libucore/src/iomux_epoll.c

#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;

    do {    

        rc = epoll_wait(mux->epoll_fd, mux->pending_events, EPOLL_WAIT_EVENTS, timeout_ms);

    } while (rc == -1 && errno == EINTR);

    assert(rc >= 0);
    if(rc < 0)
        return errno;

    mux->pending = rc;

    event_cnt = iomux_timers_run(mux_); //process timers

    if (rc == 0) //no fd's were ready, no more work to do
        return event_cnt;

    for (i = 0; i < rc; i++) {
        struct IOMuxFD *fd = mux->pending_events[i].data.ptr;
        unsigned int 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.
        if (events & MUX_EV_READ && fd != NULL) { 
            assert(fd->callback != NULL);
            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;

    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;
}