<commit_before>// Copyright (C) 2017 <EMAIL>. All rights reserved.
// Distributed under the terms and conditions of the Apache License. 
// See accompanying files LICENSE.

#include "EventLoop.h"
#include <assert.h>
#include "Common/Define.h"
#include "Common/Logging.h"
#include "Select.h"
#include "AsyncSelect.h"
#include "AsyncEvent.h"

EventLoop::EventLoop(IOMode type)
{
    timeout_ = 50;
    createIOPoller(type);
    events_.rehash(64);
    fired_.reserve(64);
}

EventLoop::~EventLoop()
{
    delete poller_;
    poller_ = NULL;
}

void EventLoop::createIOPoller(IOMode type)
{
    switch(type)
    {
    case IOSelect:
        poller_ = new SelectPoller();
        break;
    case IOAsyncSelect:
        poller_ = new AsyncSelectPoller();
        break;
    case IOEventSelect:
        poller_ = new AsyncEventPoller();
        break;
    default:
        CHECK(false) << "invalid multiplexer type: " << type;
    }
}

void EventLoop::AddEvent(SOCKET fd, int mask, EventProc func)
{
    assert(poller_ != NULL);
    auto iter = events_.find(fd);
    if (iter == events_.end())
    {
        EventEntry entry;
        entry.fd = fd;
        entry.mask =  mask;
        events_[fd] = entry;
    }
    EventEntry* entry = &events_[fd];
    entry->mask |= mask;
    if (mask & EV_READABLE)
    {
        entry->readProc = func;
    }
    if (mask & EV_WRITABLE)
    {
        entry->writeProc = func;
    }
    poller_->AddFd(fd, mask);
}

void EventLoop::DelEvent(SOCKET fd, int mask)
{
    auto iter = events_.find(fd);
    if (iter == events_.end())
    {
        LOG(ERROR) << "event entry not found " << fd;
        return;
    }
    EventEntry* entry = &iter->second;
    entry->mask = entry->mask & (~mask);
    if (mask & EV_READABLE)
    {
        entry->readProc = NULL;
    }
    if (mask & EV_WRITABLE)
    {
        entry->writeProc = NULL;
    }
    if (entry->mask == EV_NONE)
    {
        events_.erase(fd);
    }
}

EventEntry* EventLoop::GetEntry(SOCKET fd)
{
    auto iter = events_.find(fd);
    if (iter != events_.end())
    {
        return &iter->second;
    }
    return NULL;
}

void EventLoop::AddFiredEvent(SOCKET fd, int mask, int ec)
{
    FiredEvent event;
    event.fd = fd;
    event.mask = mask;
    event.ec = ec;
    fired_.push_back(event);
}

void EventLoop::Run()
{
    while (true)
    {
        RunOne();
    }
}

void EventLoop::RunOne()
{
    if (events_.empty())
    {
        return ;
    }

    fired_.clear();
    poller_->Poll(this, 100);
    for (unsigned i = 0; i < fired_.size(); i++)
    {
        const FiredEvent* event = &fired_[i];
        auto iter = events_.find(event->fd);
        if (iter == events_.end())
            continue;
        EventEntry* entry = &iter->second;
        if ((event->mask & EV_READABLE) && entry->readProc)
        {
            entry->readProc(event->fd, event->mask, event->ec);
        }
        if ((event->mask & EV_WRITABLE) && entry->writeProc)
        {
            entry->writeProc(event->fd, event->mask, event->ec);
        }
    }
}<commit_msg>Fix poller<commit_after>// Copyright (C) 2017 <EMAIL>. All rights reserved.
// Distributed under the terms and conditions of the Apache License. 
// See accompanying files LICENSE.

#include "EventLoop.h"
#include <assert.h>
#include "Common/Define.h"
#include "Common/Logging.h"
#include "Select.h"
#include "AsyncSelect.h"
#include "AsyncEvent.h"

EventLoop::EventLoop(IOMode type)
{
    timeout_ = 50;
    createIOPoller(type);
    events_.rehash(64);
    fired_.reserve(64);
}

EventLoop::~EventLoop()
{
    delete poller_;
    poller_ = NULL;
}

void EventLoop::createIOPoller(IOMode type)
{
    switch(type)
    {
    case IOSelect:
        poller_ = new SelectPoller();
        break;
    case IOAsyncSelect:
        poller_ = new AsyncSelectPoller();
        break;
    case IOEventSelect:
        poller_ = new AsyncEventPoller();
        break;
    default:
        CHECK(false) << "invalid multiplexer type: " << type;
    }
}

void EventLoop::AddEvent(SOCKET fd, int mask, EventProc func)
{
    assert(poller_ != NULL);
    auto iter = events_.find(fd);
    if (iter == events_.end())
    {
        EventEntry entry;
        entry.fd = fd;
        entry.mask =  mask;
        events_[fd] = entry;
    }
    EventEntry* entry = &events_[fd];
    entry->mask |= mask;
    if (mask & EV_READABLE)
    {
        entry->readProc = func;
    }
    if (mask & EV_WRITABLE)
    {
        entry->writeProc = func;
    }
    poller_->AddFd(fd, mask);
}

void EventLoop::DelEvent(SOCKET fd, int mask)
{
    auto iter = events_.find(fd);
    if (iter == events_.end())
    {
        LOG(ERROR) << "event entry not found " << fd;
        return;
    }
    EventEntry* entry = &iter->second;
    entry->mask = entry->mask & (~mask);
    if (mask & EV_READABLE)
    {
        entry->readProc = NULL;
        poller_->DelFd(fd, EV_READABLE);
    }
    if (mask & EV_WRITABLE)
    {
        entry->writeProc = NULL;
        poller_->DelFd(fd, EV_WRITABLE);
    }
}

EventEntry* EventLoop::GetEntry(SOCKET fd)
{
    auto iter = events_.find(fd);
    if (iter != events_.end())
    {
        return &iter->second;
    }
    return NULL;
}

void EventLoop::AddFiredEvent(SOCKET fd, int mask, int ec)
{
    FiredEvent event;
    event.fd = fd;
    event.mask = mask;
    event.ec = ec;
    fired_.push_back(event);
}

void EventLoop::Run()
{
    while (true)
    {
        RunOne();
    }
}

void EventLoop::RunOne()
{
    if (events_.empty())
    {
        return ;
    }

    fired_.clear();
    poller_->Poll(this, timeout_);
    for (unsigned i = 0; i < fired_.size(); i++)
    {
        const FiredEvent* event = &fired_[i];
        auto iter = events_.find(event->fd);
        if (iter == events_.end())
        {
            continue;
        }
        EventEntry* entry = &iter->second;
        if ((event->mask & EV_READABLE) && entry->readProc)
        {
            entry->readProc(event->fd, event->mask, event->ec);
        }
        if ((event->mask & EV_WRITABLE) && entry->writeProc)
        {
            entry->writeProc(event->fd, event->mask, event->ec);
        }
        if (entry->mask == EV_NONE)
        {
            events_.erase(event->fd);
        }
    }
}
{'fraction_non_alphanumeric': 0.14453187730849526, 'fraction_numerical': 0.004014774369680424, 'mean_word_length': 2.2086553323029365, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 62, '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': '3796743', 'n_tokens_mistral': 2129, 'n_tokens_neox': 2100, 'n_words': 467}