jay/component/loop.cpp

// SPDX-FileCopyrightText: 2024 Yury Gubich <blue@macaw.me>
// SPDX-License-Identifier: GPL-3.0-or-later

#include "loop.h"

#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>
#include <string.h>
#include <csignal>

Loop::Loop(const Shared::Logger& logger):
    Shared::Loggable(logger, {"Loop"}),
    wakePipe(),
    handlers(),
    handlersToAdd(),
    descriptorsToRemove(),
    mutex(),
    running(false)
{
    if (pipe(wakePipe) < 0) {
        fatal(std::string("pipe: ") + strerror(errno));
        std::exit(1);
    }

    fcntl(wakePipe[0], F_SETFL, O_NONBLOCK);
    fcntl(wakePipe[1], F_SETFL, O_NONBLOCK);

    registerInstance(this);
}

Loop::~Loop() {
    unregisterInstance(this);

    close(wakePipe[0]);
    close(wakePipe[1]);
}

void Loop::run() {
    if (running.exchange(true)) {
        warn("an attempt to run already running event loop, skipping");
        return;
    }

    debug("entering the loop");
    while (running) {
        syncHandlers();

        fd_set readfds;
        int maxFD = setFDsAndFindMax(&readfds);

        int result = select(maxFD + 1, &readfds, nullptr, nullptr, nullptr);

        if (result < 0) {
            if (errno == EINTR) {
                debug("interrupted");
                // drain(&readfds); looks like it's not a good idea to drain here
                continue;
            };
            fatal(std::string("select: ") + strerror(errno));
            running = false;
            break;
        }
        
        drain(&readfds);
        runCallbacks(&readfds);
    }

    debug("left the loop");
}

void Loop::stop() {
    if (!running.exchange(false))
        return;

    debug("stopping the loop");
    wake();
}

void Loop::addDescriptor(int descriptor, const Callback& handler) {
    std::lock_guard lock(mutex);

    if (handlers.count(descriptor) != 0) {
        warn("an attempt to add descriptor " + std::to_string(descriptor) + " for the second time");
        return;
    }

    if (!handlersToAdd.emplace(descriptor, handler).second) {
        warn("an attempt to add descriptor " + std::to_string(descriptor) + " for the second time");
        return;
    }

    wake();
}

void Loop::removeDescriptor(int descriptor) {
    std::lock_guard lock(mutex);
            
    if (handlersToAdd.erase(descriptor) != 0)
        return;

    if (handlers.count(descriptor) == 0) {
        warn("an attempt to remove unknown descriptor " + std::to_string(descriptor));
        return;
    }

    if (!descriptorsToRemove.insert(descriptor).second) {
        warn("an attempt to remove descriptor " + std::to_string(descriptor) + " for the second time");
        return;
    }

    wake();
}

void Loop::wake() {
    char w = 'w';
    if (write(wakePipe[1], &w, 1) < 0 && errno != EAGAIN)
        error("failed to wake up event loop: " + std::string(strerror(errno)));
}

void Loop::drain(fd_set* readfds) {
    if (!FD_ISSET(wakePipe[0], readfds))
        return;

    char buf[64];
    while (true) {
        ssize_t r = read(wakePipe[0], buf, sizeof(buf));
        if (r > 0)
            continue;

        if (r == -1) {
            if (errno == EAGAIN) 
                break;
            
            error("failed to drain wake pipe: " + std::string(strerror(errno)));
            break;
        }

        if (r == 0) 
            break; // pipe closed
    }
}

void Loop::runCallbacks(fd_set* readfds) {
    for (const std::pair<const int, Callback>& pair : handlers) {
        if (!FD_ISSET(pair.first, readfds))
            continue;
    
        try {
            pair.second();
        } catch (const std::exception& e) {
            error("exception in event loop from handler with file descriptor " + std::to_string(pair.first) + ":\n" + e.what());
        } catch (...) {
            error("unhandled throw in event loop from handler with file descriptor " + std::to_string(pair.first));
        }
    }
}

int Loop::setFDsAndFindMax(fd_set* readfds) {
    FD_ZERO(readfds);
    FD_SET(wakePipe[0], readfds);

    int maxFD = wakePipe[0];
    for (const std::pair<const int, Callback>& pair : handlers) {
        FD_SET(pair.first, readfds);
        maxFD = std::max(maxFD, pair.first);
    }

    return maxFD;
}

void Loop::syncHandlers() {
    std::lock_guard lock(mutex);

    for (int descriptor : descriptorsToRemove)
        handlers.erase(descriptor);

    for (const std::pair<const int, Callback>& pair : handlersToAdd)
        handlers.insert(pair);

    descriptorsToRemove.clear();
    handlersToAdd.clear();
}

void Loop::registerInstance(Loop* loop) {
    std::lock_guard lock(instanceMutex);
    if (instances.size() == 0) {
        std::signal(SIGINT, signalHandler);
        std::signal(SIGTERM, signalHandler);
    }

    instances.insert(loop);
}

void Loop::unregisterInstance(Loop* loop) {
    std::lock_guard lock(instanceMutex);
    instances.erase(loop);

    if (instances.size() == 0) {
        std::signal(SIGINT, SIG_DFL);
        std::signal(SIGTERM, SIG_DFL);
    }
}

void Loop::signalHandler(int signum) {
    for (Loop* loop : instances)
        loop->stop();
}