<commit_before>#include <crossbow/infinio/EventProcessor.hpp>

#include <crossbow/infinio/Fiber.hpp>
#include <crossbow/logger.hpp>

#include <algorithm>
#include <cerrno>

#include <sys/epoll.h>
#include <sys/eventfd.h>

namespace crossbow {
namespace infinio {

EventProcessor::EventProcessor(uint64_t pollCycles)
        : mPollCycles(pollCycles) {
    LOG_TRACE("Creating epoll file descriptor");
    mEpoll = epoll_create1(EPOLL_CLOEXEC);
    if (mEpoll == -1) {
        throw std::system_error(errno, std::system_category());
    }
}

EventProcessor::~EventProcessor() {
    // TODO We have to join the poll thread
    // We are not allowed to call join in the same thread as the poll loop is

    LOG_TRACE("Destroying epoll file descriptor");
    if (close(mEpoll)) {
        std::error_code ec(errno, std::system_category());
        LOG_ERROR("Failed to close the epoll descriptor [error = %1% %2%]", ec, ec.message());
    }
}

void EventProcessor::registerPoll(int fd, EventPoll* poll) {
    LOG_TRACE("Register event poller");
    struct epoll_event event;
    event.data.ptr = poll;
    event.events = EPOLLIN | EPOLLET;
    if (epoll_ctl(mEpoll, EPOLL_CTL_ADD, fd, &event)) {
        throw std::system_error(errno, std::system_category());
    }

    mPoller.emplace_back(poll);
}

void EventProcessor::deregisterPoll(int fd, EventPoll* poll) {
    LOG_TRACE("Deregister event poller");
    auto i = std::find(mPoller.begin(), mPoller.end(), poll);
    if (i == mPoller.end()) {
        return;
    }
    mPoller.erase(i);

    if (epoll_ctl(mEpoll, EPOLL_CTL_DEL, fd, nullptr)) {
        throw std::system_error(errno, std::system_category());
    }
}

void EventProcessor::start() {
    LOG_TRACE("Starting event processor");
    mPollThread = std::thread([this] () {
        while (true) {
            doPoll();
        }
    });
}

void EventProcessor::executeFiber(std::function<void(Fiber&)> fun) {
    auto fiber = Fiber::create(*this, std::move(fun));
    fiber->resume();
}

void EventProcessor::doPoll() {
    for (uint32_t i = 0; i < mPollCycles; ++i) {
        for (auto poller : mPoller) {
            if (poller->poll()) {
                i = 0;
            }
        }
        while (!mTaskQueue.empty()) {
            mTaskQueue.front()();
            mTaskQueue.pop();
            i = 0;
        }
    }

    for (auto poller : mPoller) {
        poller->prepareSleep();
    }

    LOG_TRACE("Going to epoll sleep");
    struct epoll_event events[mPoller.size()];
    auto num = epoll_wait(mEpoll, events, 10, -1);<commit_msg>Swap the task queue before processing it<commit_after>#include <crossbow/infinio/EventProcessor.hpp>

#include <crossbow/infinio/Fiber.hpp>
#include <crossbow/logger.hpp>

#include <algorithm>
#include <cerrno>

#include <sys/epoll.h>
#include <sys/eventfd.h>

namespace crossbow {
namespace infinio {

EventProcessor::EventProcessor(uint64_t pollCycles)
        : mPollCycles(pollCycles) {
    LOG_TRACE("Creating epoll file descriptor");
    mEpoll = epoll_create1(EPOLL_CLOEXEC);
    if (mEpoll == -1) {
        throw std::system_error(errno, std::system_category());
    }
}

EventProcessor::~EventProcessor() {
    // TODO We have to join the poll thread
    // We are not allowed to call join in the same thread as the poll loop is

    LOG_TRACE("Destroying epoll file descriptor");
    if (close(mEpoll)) {
        std::error_code ec(errno, std::system_category());
        LOG_ERROR("Failed to close the epoll descriptor [error = %1% %2%]", ec, ec.message());
    }
}

void EventProcessor::registerPoll(int fd, EventPoll* poll) {
    LOG_TRACE("Register event poller");
    struct epoll_event event;
    event.data.ptr = poll;
    event.events = EPOLLIN | EPOLLET;
    if (epoll_ctl(mEpoll, EPOLL_CTL_ADD, fd, &event)) {
        throw std::system_error(errno, std::system_category());
    }

    mPoller.emplace_back(poll);
}

void EventProcessor::deregisterPoll(int fd, EventPoll* poll) {
    LOG_TRACE("Deregister event poller");
    auto i = std::find(mPoller.begin(), mPoller.end(), poll);
    if (i == mPoller.end()) {
        return;
    }
    mPoller.erase(i);

    if (epoll_ctl(mEpoll, EPOLL_CTL_DEL, fd, nullptr)) {
        throw std::system_error(errno, std::system_category());
    }
}

void EventProcessor::start() {
    LOG_TRACE("Starting event processor");
    mPollThread = std::thread([this] () {
        while (true) {
            doPoll();
        }
    });
}

void EventProcessor::executeFiber(std::function<void(Fiber&)> fun) {
    auto fiber = Fiber::create(*this, std::move(fun));
    fiber->resume();
}

void EventProcessor::doPoll() {
    for (uint32_t i = 0; i < mPollCycles; ++i) {
        for (auto poller : mPoller) {
            if (poller->poll()) {
                i = 0;
            }
        }
        if (!mTaskQueue.empty()) {
            decltype(mTaskQueue) taskQueue;
            taskQueue.swap(mTaskQueue);
            do {
                taskQueue.front()();
                taskQueue.pop();
                i = 0;
            } while (!taskQueue.empty());
        }
    }

    for (auto poller : mPoller) {
        poller->prepareSleep();
    }

    LOG_TRACE("Going to epoll sleep");
    struct epoll_event events[mPoller.size()];
    auto num = epoll_wait(mEpoll, events, 10, -1);
{'fraction_non_alphanumeric': 0.14169658038919328, 'fraction_numerical': 0.005290005667863216, 'mean_word_length': 2.918578830495929, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1991034', 'n_tokens_mistral': 1824, 'n_tokens_neox': 1764, 'n_words': 387}