<commit_before>#include <atomic>
#include <chrono>
#include <ctime>
#include <deque>
#include <functional>
#include <iomanip>
#include <iostream>
#include <string>
#include <tuple>

#include <cassert>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <unistd.h>

#include "logging.hpp"

namespace io
{
    class selectable: public std::enable_shared_from_this<selectable>
    {
        public:
            virtual int fd() = 0;
            virtual void do_read() = 0;
            virtual void do_write() = 0;
            virtual ~selectable() {};

            void do_close();
    };

    typedef std::tuple<std::chrono::milliseconds, std::function<void()>> delayed_call;

    class loop
    {
        std::map<int, std::shared_ptr<io::selectable>> selectables;

        std::deque<delayed_call> delayed_calls;

        int epoll_fd;
        std::atomic<bool> running;

        public:
        loop() : epoll_fd(-1), running(false) {};
        ~loop();

        void init();
        void add_selectable(std::shared_ptr<io::selectable> selectable);
        void remove_selectable(int fd);
        void call_soon(std::function<void()> func);
        void call_later(std::chrono::milliseconds delay, std::function<void()> func);
        void stop();

        static loop* instance() {
            static loop ev;
            return &ev;
        };

        void run();
        void run_delayed_calls(const std::chrono::milliseconds &elapsed);
    };

    std::string error(int errnum)
    {
        std::string msg;
        char buf[512];
        msg = strerror_r(errnum, buf, 512);
        return msg + " (" + std::to_string(errnum) + ")";
    }

    std::tuple<int, std::string> write(int fd, std::string buf)
    {
        int written = ::write(fd, buf.c_str(), buf.size());

        if (written < 0) {
            int xerrno = errno;
            logger(2, "write error " << io::error(xerrno));
            close(fd);
        } else {
            buf = buf.substr(written, std::string::npos);
        }

        return std::make_tuple(written, buf);
    }

    std::string consume(int fd)
    {
        std::string msg;
        ssize_t rsize;
        char buf[512];

        while (true) {
            memset(&buf, 0, 512);
            rsize = read(fd, buf, sizeof(buf));
            logger(5, "read " << rsize << " bytes from " << fd);
            if (rsize == -1) {
                if (errno != EAGAIN) {
                    perror ("read");
                    // FIXME: raise exception. Would be cool to make sure if
                    // msg.size() that we... handle that in some way.
                }
                break;
            } else if (rsize == 0) {
                break;
            } else {
                msg += buf;
            }<commit_msg>Don't add a raw char* here, it'll only read up the NULL byte, which means not being able to handle actually receiving a NULL byte.<commit_after>#include <atomic>
#include <chrono>
#include <ctime>
#include <deque>
#include <functional>
#include <iomanip>
#include <iostream>
#include <string>
#include <tuple>

#include <cassert>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <unistd.h>

#include "logging.hpp"

namespace io
{
    class selectable: public std::enable_shared_from_this<selectable>
    {
        public:
            virtual int fd() = 0;
            virtual void do_read() = 0;
            virtual void do_write() = 0;
            virtual ~selectable() {};

            void do_close();
    };

    typedef std::tuple<std::chrono::milliseconds, std::function<void()>> delayed_call;

    class loop
    {
        std::map<int, std::shared_ptr<io::selectable>> selectables;

        std::deque<delayed_call> delayed_calls;

        int epoll_fd;
        std::atomic<bool> running;

        public:
        loop() : epoll_fd(-1), running(false) {};
        ~loop();

        void init();
        void add_selectable(std::shared_ptr<io::selectable> selectable);
        void remove_selectable(int fd);
        void call_soon(std::function<void()> func);
        void call_later(std::chrono::milliseconds delay, std::function<void()> func);
        void stop();

        static loop* instance() {
            static loop ev;
            return &ev;
        };

        void run();
        void run_delayed_calls(const std::chrono::milliseconds &elapsed);
    };

    std::string error(int errnum)
    {
        std::string msg;
        char buf[512];
        msg = strerror_r(errnum, buf, 512);
        return msg + " (" + std::to_string(errnum) + ")";
    }

    std::tuple<int, std::string> write(int fd, std::string buf)
    {
        int written = ::write(fd, buf.c_str(), buf.size());

        if (written < 0) {
            int xerrno = errno;
            logger(2, "write error " << io::error(xerrno));
            close(fd);
        } else {
            buf = buf.substr(written, std::string::npos);
        }

        return std::make_tuple(written, buf);
    }

    std::string consume(int fd)
    {
        std::string msg;
        ssize_t rsize;
        char buf[512];

        while (true) {
            memset(&buf, 0, 512);
            rsize = read(fd, buf, sizeof(buf));
            logger(5, "read " << rsize << " bytes from " << fd);
            if (rsize == -1) {
                if (errno != EAGAIN) {
                    perror ("read");
                    // FIXME: raise exception. Would be cool to make sure if
                    // msg.size() that we... handle that in some way.
                }
                break;
            } else if (rsize == 0) {
                break;
            } else {
                msg += std::string(buf, rsize);
            }
{'fraction_non_alphanumeric': 0.13936002797691904, 'fraction_numerical': 0.007693652736492394, 'mean_word_length': 1.9168791432942376, 'pattern_counts': {'":': 0, '<': 71, '<?xml version=': 0, '>': 53, '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': '6052403', 'n_tokens_mistral': 1803, 'n_tokens_neox': 1719, 'n_words': 439}