atoms/epoll_poller.cc

Go to the documentation of this file.

#include <atoms/epoll_poller.hh>
#include <atoms/debug.hh>
#include <sys/epoll.h>
#include <unistd.h>
#include <errno.h>

using namespace std;
using namespace boost;

namespace atoms {

DebugLogger EpollPoller::LOGGER("org.slamb.atoms.EpollPoller");
const int EpollPoller::CREATE_SIZE = 128;
const int EpollPoller::NUM_EVENTS  = 128;

EpollPoller::EpollPoller() {
    epfd = epoll_create(CREATE_SIZE);
    if (epfd < 0) {
        if (errno == ENOSYS) {
            throw UnsupportedError("No epoll support in kernel.");
        }
        assert(false); 
    }
    events.reset(new epoll_event[NUM_EVENTS]);
}

EpollPoller::~EpollPoller() {
    if (close(epfd) < 0) {
        assert(false); 
    }
}

void EpollPoller::add(int fd, Event eventMask, const EventHandler &onEvent) {
    map<int, boost::shared_ptr<DescriptorInfo> >::const_iterator it =
        descriptorMap.find(fd);
    boost::shared_ptr<DescriptorInfo> d;
    bool edgeTriggered = (eventMask & eEdgeTriggered) != 0;
    int op;

    assert(   (eventMask &  eAddable) != 0
           && (eventMask & ~eAddable & ~eEdgeTriggered) == 0);

    if (it != descriptorMap.end()) {
        // Adding an event to an existing descriptor
        d = it->second;
        op = EPOLL_CTL_MOD;
        assert(d->edgeTriggered == edgeTriggered);
    } else {
        // Adding an entirely new descriptor
        op = EPOLL_CTL_ADD;
        d.reset(new DescriptorInfo);
        d->stale = false;
        d->edgeTriggered = edgeTriggered;
        descriptorMap.insert(make_pair(fd,d));
    }

    // New events
    if ((eventMask & eInput) != 0) {
        if (!d->eventHandlerMap.insert(make_pair(eInput, onEvent)).second) {
            LOGGER.logFatal("Descriptor %d already being monitored for input in epfd %d",
                            fd, epfd);
        } else {
            LOGGER.log(DebugLogger::lInfo, "Descriptor %d added for input on epfd %d",
                       fd, epfd);
        }
    }
    if ((eventMask & eOutput) != 0) {
        if (!d->eventHandlerMap.insert(make_pair(eOutput, onEvent)).second) {
            LOGGER.logFatal("Descriptor %d already being monitored for output on epfd %d",
                            fd, epfd);
        } else {
            LOGGER.log(DebugLogger::lInfo, "Descriptor %d added for output on epfd %d",
                       fd, epfd);
        }
    }
    if ((eventMask & ePriorityInput) != 0) {
        if (!d->eventHandlerMap.insert(
            make_pair(ePriorityInput, onEvent)).second) {
            LOGGER.logFatal("Descriptor %d already being monitored for "
                            "priority input on epfd %d", fd, epfd);
        } else {
            LOGGER.log(DebugLogger::lInfo,
                       "Descriptor %d added for priority input on epfd %d", fd, epfd);
        }
    }

    // Construct the structure epoll_ctl likes
    struct epoll_event event;
    event.events = edgeTriggered ? EPOLLET : 0;
    map<Event,EventHandler>::const_iterator ehIt;
    for (ehIt = d->eventHandlerMap.begin();
         ehIt != d->eventHandlerMap.end(); ++ehIt) {
        if (ehIt->first == eInput)
            event.events |= EPOLLIN;
        else if (ehIt->first == eOutput)
            event.events |= EPOLLOUT;
        else if (ehIt->first == ePriorityInput)
            event.events |= EPOLLPRI;
    }
    event.data.ptr = d.get();

    if (epoll_ctl(epfd, op, fd, &event) < 0) {
        assert(false); 
    }

}

void EpollPoller::remove(int fd, Event eventMask) {
    map<int, boost::shared_ptr<DescriptorInfo> >::iterator it =
        descriptorMap.find(fd);
    if (it == descriptorMap.end()) {
        LOGGER.logFatal("Descriptor %d was not in set for epfd %d", fd, epfd);
    }
    boost::shared_ptr<DescriptorInfo> d = it->second;

    assert(   (eventMask &  eAddable) != 0
           && (eventMask & ~eAddable) == 0);

    map<Event,EventHandler> oldEventHandlerMap(d->eventHandlerMap);
    if ((eventMask & eInput) != 0) {
        if (d->eventHandlerMap.erase(eInput) == 0) {
            LOGGER.logFatal("Descriptor %d not registered for input on epfd %d", fd, epfd);
        } else {
            LOGGER.log(DebugLogger::lInfo,
                       "Descriptor %d deregistered for input on epfd %d", fd, epfd);
        }
    }
    if ((eventMask & eOutput) != 0) {
        if (d->eventHandlerMap.erase(eOutput) == 0) {
            LOGGER.logFatal("Descriptor %d not registered for output on epfd %d", fd, epfd);
        } else {
            LOGGER.log(DebugLogger::lInfo,
                       "Descriptor %d deregistered for output on epfd %d", fd, epfd);
        }
    }
    if ((eventMask & ePriorityInput) != 0) {
        if (d->eventHandlerMap.erase(ePriorityInput) == 0) {
            LOGGER.logFatal("Descriptor %d not registered for prio input on epfd %d", fd, epfd);
        } else {
            LOGGER.log(DebugLogger::lInfo,
                       "Descriptor %d deregistered for prio input on epfd %d", fd, epfd);
        }
    }

    if (d->eventHandlerMap.empty()) {
        // If no event types are left, remove our mapping
        // removeInternal(it, false);
    } else {
        struct epoll_event event;
        event.events = d->edgeTriggered ? EPOLLET : 0;
        map<Event,EventHandler>::const_iterator ehIt;
        for (ehIt = d->eventHandlerMap.begin();
             ehIt != d->eventHandlerMap.end(); ++ehIt) {
            if (ehIt->first == eInput)
                event.events |= EPOLLIN;
            else if (ehIt->first == eOutput)
                event.events |= EPOLLOUT;
            else if (ehIt->first == ePriorityInput)
                event.events |= EPOLLPRI;
        }
        event.data.ptr = d.get();

        if (epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &event) < 0) {
            assert(false); 
        }
    }
}

void EpollPoller::remove(int fd, bool isClosing) {
    map<int, boost::shared_ptr<DescriptorInfo> >::iterator it =
        descriptorMap.find(fd);
    if (it == descriptorMap.end()) {
        LOGGER.logFatal("Descriptor %d was not in set", fd);
    } else {
        LOGGER.log(DebugLogger::lInfo,
                   "Descriptor %d deregistered entirely", fd);
    }
    removeInternal(it, isClosing);
}

void EpollPoller::removeInternal(
        map<int, boost::shared_ptr<DescriptorInfo> >::iterator it,
        bool isClosing) {
    int fd = it->first;
    boost::shared_ptr<DescriptorInfo> d = it->second;
    descriptorMap.erase(it);
    d->stale = true;
    staleDescriptors.push_back(d);
    if (!isClosing) {
        // It doesn't seem like it should be necessary for us to fill out
        // an epoll_event here, but epoll_Ctl gives us EFAULT if we pass
        // it NULL
        struct epoll_event event;
        event.events = 0;
        event.data.ptr = d.get();
        if (epoll_ctl(epfd, EPOLL_CTL_DEL, fd, &event) < 0) {
            LOGGER.logFatal("epoll_ctl(EPOLL_CTL_DEL) failed: %d (%s)",
                            errno, strerror(errno));
        }
    }
}

void EpollPoller::poll(const Timespan &interval) {
    staleDescriptors.clear();
    int retval;
    int milliseconds = -1;
    const struct timespec *ts = interval.getTimespec();
    if (ts != NULL) {
        milliseconds = ts->tv_sec * 1000 + ts->tv_nsec / 1000000;
    }
    {
        DebugLogger::Mon dm(LOGGER, DebugLogger::lInfo, "Polling on epfd %d", epfd);
        retval = epoll_wait(epfd, events.get(), NUM_EVENTS, milliseconds);
    }
    if (retval < 0 && errno != EINTR) {
        LOGGER.logFatal("epoll_wait failed: %d (%s)", errno, strerror(errno));
    }
    for (int i = 0; i < retval; i++) {
        DescriptorInfo *d = reinterpret_cast<DescriptorInfo*>
            (events[i].data.ptr);
        if (d->stale)
            continue;
        int eventMask = 0;
        if ((events[i].events & EPOLLIN ) != 0) eventMask |= eInput;
        if ((events[i].events & EPOLLOUT) != 0) eventMask |= eOutput;
        if ((events[i].events & EPOLLPRI) != 0) eventMask |= ePriorityInput;
        if ((events[i].events & EPOLLERR) != 0) eventMask |= eError;
        if ((events[i].events & EPOLLHUP) != 0) eventMask |= eHangup;

        map<Event, EventHandler>::const_iterator it;
        for (it = d->eventHandlerMap.begin(); it != d->eventHandlerMap.end();
             ++it) {
            if ((it->first & eventMask) != 0)
                it->second(static_cast<Event>(eventMask));
        }
    }
}

} // namespace atoms

---