#include #include #include #include #include #include #include //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; again: rc = epoll_wait(mux->epoll_fd, mux->pending_events, EPOLL_WAIT_EVENTS, timeout_ms); if (rc == -1 && errno == EINTR) goto again; 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; }