eventloop.h

// Copyright (c) 2018 IIS (The Internet Foundation in Sweden)
// Written by Göran Andersson <initgoran@gmail.com>
 
#pragma once
 
#include <set>
#include <map>
 
#include "logger.h"
#include "engine.h"
#ifdef USE_THREADS
#include "msgqueue.h"
#endif
 
class Task;
class SocketConnection;
class ServerSocket;
class WorkerProcess;
 
class EventLoop : public Logger {
public:
    EventLoop(std::string log_label = "MainLoop") :
        Logger(log_label),
        engine("NetworkEngine"),
        name(log_label) {
#ifdef USE_THREADS
        do_init(nullptr);
#else
        do_init();
#endif
    }
 
    ~EventLoop();
 
    void addTask(Task *task, Task *parent = nullptr);
 
    bool run(double timeout_s);
 
    void runUntilComplete();
 
#ifdef USE_THREADS
    static void runTask(Task *task, const std::string &name = "MainLoop",
                        std::ostream *log_file = nullptr,
                        EventLoop *parent = nullptr);
#else
    static void runTask(Task *task, const std::string &name = "MainLoop",
                        std::ostream *log_file = nullptr);
#endif
 
    void waitForThreadsToFinish();
 
    void abortTask(Task *task);
 
    void abort() {
        interrupt();
        do_abort = true;
    }
 
    void getChildTasks(std::set<Task *> &tset, Task *parent) const;
 
    void abortChildTasks(Task *parent);
 
    void wakeUpTask(Task *t);
 
    bool wakeUpConnection(SocketConnection *s) {
        return engine.wakeUpConnection(s);
    }
 
    void cancelConnection(SocketConnection *s) {
        engine.cancelConnection(s);
    }
 
    std::set<Socket *> findConnByTask(const Task *t) const;
 
    bool isActive(const Socket *conn) const {
        return engine.connActive(conn);
    }
 
    void resetTimer(Task *task, double s);
 
    bool addConnection(SocketConnection *conn);
 
    bool addConnected(SocketConnection *conn);
 
    bool addServer(ServerSocket *conn);
#ifdef USE_GNUTLS
    bool tlsSetKey(ServerSocket *conn, const std::string &crt_path,
                   const std::string &key_path, const std::string &password) {
        return engine.tlsSetKey(conn, crt_path, key_path, password);
    }
 
    bool setCABundle(const std::string &path) {
        return engine.setCABundle(path);
    }
#endif
 
    bool running(Task *task);
 
    void taskDeleted(Task *task);
 
    static void interrupt() {
        Engine::yield();
    }
 
#ifdef USE_THREADS
    void spawnThread(Task *task, const std::string &name="ThreadLoop",
                     std::ostream *log_file = nullptr,
                     Task *parent = nullptr);
#endif
#ifdef _WIN32
    int externalCommand(Task *owner, const char *const argv[]) {
        // Not implemented
        exit(1);
    }
#else
    int externalCommand(Task *owner, const char *const argv[]);
 
    static void daemonize();
 
    WorkerProcess *createWorker(Task *parent, std::ostream *log_file,
                                unsigned int channels, unsigned int wno);
 
    WorkerProcess *createWorker(Task *parent, const std::string &log_file_name,
                                unsigned int channels, unsigned int wno);
 
    void killChildProcesses(int signum);
 
    static void setLogFilename(const std::string &filename) {
        openFileOnSIGHUP = filename;
    }
#endif
 
    void notifyTaskFinished(Task *task) {
        auto ret = finishedTasks.insert(task);
        if (ret.second)
            engine.yield();
    }
 
    void notifyTaskMessage(Task *task) {
        auto ret = messageTasks.insert(task);
        if (ret.second)
            engine.yield();
    }
 
    bool aborted() const {
        return do_abort;
    }
 
    void addSignalHandler(int signum, void (*handler)(int, EventLoop &));
 
    bool startObserving(Task *from, Task *to);
 
    bool isObserving(Task *observer, Task *task) const {
        auto p = observed_by.find(observer);
        return p != observed_by.end() &&
            p->second.find(task) != p->second.end();
    }
 
private:
#ifndef _WIN32
#ifdef USE_THREADS
    thread_local
#endif
    static std::map<int, int> terminatedPIDs;
    std::map<int, Task *> pidOwner;
#ifdef USE_THREADS
    thread_local
#endif
    static std::string openFileOnSIGHUP;
#endif
#ifdef USE_THREADS
    void do_init(EventLoop *parent);
 
    EventLoop(std::string log_label, EventLoop *parent) :
        Logger(log_label),
        engine("NetworkEngine"),
        name(log_label) {
        do_init(parent);
    }
    std::map<Task *, std::thread> threads;
    static MsgQueue<Task *> finished_threads;
    std::map<Task *, Task *> threadTaskObserver;
    void collect_thread(Task *t);
    EventLoop *parent_loop;
#else
    void do_init();
#endif
 
#ifdef USE_THREADS
    //thread_local
#endif
    static volatile int got_signal;
#ifdef USE_THREADS
    thread_local
#endif
    static volatile int terminatedPIDtmp[100];
 
    static void signalHandler(int signum);
    void removeAllTasks();
    void check_finished();
    Task *nextTimerToExecute();
    void _removeTimer(Task *task);
    void _removeTask(Task *task, bool killed = false);
    Engine engine;
 
    // Map each task to its parent (or, if it has no parent, to nullptr)
    std::map<Task *, Task *> tasks;
 
    // These are used to keep track of "observation" between tasks.
    // E.g. a parent task is observing its children.
    std::map<Task *, std::set<Task *> > observed_by;
    std::map<Task *, std::set<Task *> > observing;
 
    std::set<Task *> finishedTasks, messageTasks;
    std::multimap<int, void (*)(int, EventLoop &)> userSignalHandler;
    std::multimap<TimePoint, Task *> timer_queue;
    std::string name;
    bool do_abort = false;
};