atoms/kqueue_poller.cc

Go to the documentation of this file.

#include <atoms/kqueue_poller.hh>
#include <sys/types.h>
#include <sys/event.h>
#include <sys/time.h>
#include <boost/cast.hpp>
#include <atoms/debug.hh>

using namespace std;
using namespace boost;

namespace atoms {

const int KqueuePoller::NUM_EVENTS = 16;
DebugLogger KqueuePoller::LOGGER("org.slamb.atoms.KqueuePoller");

KqueuePoller::KqueuePoller() : events(new struct kevent[NUM_EVENTS]) {
    if ((kq = kqueue()) < 0) {
        LOGGER.logFatal("Unexpected kqueue error: %d (%s)",
                        errno, strerror(errno));
    }
    LOGGER.log(DebugLogger::lInfo, "Created kqueue fd %d", kq);
}

KqueuePoller::~KqueuePoller() {
    if (::close(kq) < 0) {
        LOGGER.logFatal("Unexpected close error on kqueue fd %d: %d (%s)",
                        kq, errno, strerror(errno));
    }
}

void KqueuePoller::add(int fd, Event eventMask, const EventHandler &onEvent) {
    map<int, shared_ptr<DescriptorInfo> >::const_iterator it =
        descriptorMap.find(fd);
    struct kevent changes[2];
    unsigned int numChanges = 0;
    bool edgeTriggered = (eventMask & eEdgeTriggered) != 0;
    assert(   (eventMask &  eAddable) != 0
           && (eventMask & ~eAddable & ~eEdgeTriggered) == 0);

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

    u_short flags = EV_ADD;
    if ((eventMask & eEdgeTriggered) != 0) {
        flags |= EV_CLEAR;
    }
    if ((eventMask & eInput) != 0) {
        EV_SET(&changes[numChanges++], fd, EVFILT_READ, flags, 0, 0, d.get());
        d->eventHandlerMap.insert(make_pair(eInput, onEvent));
    }
    if ((eventMask & eOutput) != 0) {
        EV_SET(&changes[numChanges++], fd, EVFILT_WRITE, flags, 0, 0, d.get());
        d->eventHandlerMap.insert(make_pair(eOutput, onEvent));
    }
    struct timespec ts;
    ts.tv_sec = 0L;
    ts.tv_nsec = 0L;
    if (kevent(kq, changes, numChanges, NULL, 0, &ts) < 0) {
        LOGGER.logFatal("Unexpected kevent error: %d (%s)",
                        errno, strerror(errno));
    }
}

void KqueuePoller::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", fd);
    }
    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", fd);
        } else {
            LOGGER.log(DebugLogger::lInfo,
                       "Descriptor %d deregistered for input", fd);
        }
    }
    if ((eventMask & eOutput) != 0) {
        if (d->eventHandlerMap.erase(eOutput) == 0) {
            LOGGER.logFatal("Descriptor %d not registered for output", fd);
        } else {
            LOGGER.log(DebugLogger::lInfo,
                       "Descriptor %d deregistered for output", fd);
        }
    }

    struct kevent changes[2];
    int numChanges = 0;
    if ((eventMask & eInput) != 0) {
        EV_SET(&changes[numChanges++], fd, EVFILT_READ, EV_DELETE, 0, 0, 0);
    }
    if ((eventMask & eOutput) != 0) {
        EV_SET(&changes[numChanges++], fd, EVFILT_WRITE, EV_DELETE, 0, 0, 0);
    }
    assert(numChanges != 0);
    struct timespec ts;
    ts.tv_sec = 0L;
    ts.tv_nsec = 0L;
    if (kevent(kq, changes, numChanges, NULL, 0, &ts) < 0) {
        LOGGER.logFatal("Unexpected kevent error: %d (%s)",
                        errno, strerror(errno));
    }

    if (d->eventHandlerMap.empty()) {
        boost::shared_ptr<DescriptorInfo> d = it->second;
        descriptorMap.erase(it);
        d->stale = true;
        staleDescriptors.push_back(d);
    }
}

void KqueuePoller::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);
    }
    boost::shared_ptr<DescriptorInfo> d = it->second;
    if (!isClosing) {
        map<Event,EventHandler>::const_iterator it,
                                                end = d->eventHandlerMap.end();
        struct kevent changes[2];
        int numChanges = 0;
        if ((it = d->eventHandlerMap.find(eInput)) != end) {
            EV_SET(&changes[numChanges++], fd, EVFILT_READ, EV_DELETE, 0, 0, 0);
        }
        if ((it = d->eventHandlerMap.find(eOutput)) != end) {
            EV_SET(&changes[numChanges++], fd, EVFILT_WRITE, EV_DELETE,
                   0, 0, 0);
        }
        assert(numChanges != 0);
        struct timespec ts;
        ts.tv_sec = 0L;
        ts.tv_nsec = 0L;
        if (kevent(kq, changes, numChanges, NULL, 0, &ts) < 0) {
            LOGGER.logFatal("Unexpected kevent error: %d (%s)",
                            errno, strerror(errno));
        }
    }
    descriptorMap.erase(it);
    d->stale = true;
    staleDescriptors.push_back(d);
}

void KqueuePoller::poll(const Timespan &interval) {
    staleDescriptors.clear();
    const struct timespec *ts_p = interval.getTimespec();
    int retval;
    if ((retval = kevent(kq, NULL, 0, events.get(), NUM_EVENTS, ts_p)) < 0) {
        LOGGER.logFatal("Unexpected kevent error: %d (%s)",
                        errno, strerror(errno));
    }
    for (int i = 0; i < retval; i++) {
        DescriptorInfo *d = reinterpret_cast<DescriptorInfo*>(events[i].udata);
        if (d->stale)
            continue;
        Event e;
        if (events[i].filter == EVFILT_READ) {
            LOGGER.log(DebugLogger::lDebug, "Input on fd %d",
                       numeric_cast<int>(events[i].ident));
            e = eInput;
        } else /* events[i].filter == EVFILT_WRITE */ {
            LOGGER.log(DebugLogger::lDebug, "Output on fd %d",
                       numeric_cast<int>(events[i].ident));
            e = eOutput;
        }
        map<Event, EventHandler>::const_iterator it;
        for (it = d->eventHandlerMap.begin(); it != d->eventHandlerMap.end();
             ++it) {
            if ((it->first & e) != 0) {
                DebugLogger::Mon dm(LOGGER, DebugLogger::lDebug,
                                    "Callback on fd %d",
                                    numeric_cast<int>(events[i].ident));
                it->second(e);
            }
        }
    }
}

} // namespace atoms

---