task.h

// Copyright (c) 2019 Internetstiftelsen
// Written by Göran Andersson <initgoran@gmail.com>
 
#pragma once
 
#include <set>
#include <fstream>
#include "logger.h"
#include "taskconfig.h"
#include "eventloop.h"
 
class Socket;
class ServerSocket;
class SocketConnection;
enum class PollState;
class SocketReceiver;
class WorkerProcess;
 
class Task : public Logger {
public:
    Task(const std::string &task_name);
 
    virtual ~Task();
 
    virtual double start() {
        dbg_log() << "Task starting. No timer.";
        return 0;
    }
 
    virtual double timerEvent() {
        dbg_log() << "Default timerEvent: will kill task.";
        setTimeout();
        return 0;
    }
 
    void resetTimer(double s) {
        supervisor->resetTimer(this, s);
    }
 
    bool finishedOK() const {
        return (has_started && is_finished && !was_killed &&
                !was_error && !was_timeout);
    }
 
    bool wasKilled() const {
        return was_killed;
    }
 
    bool wasError() const {
        return was_error;
    }
 
    bool wasTimeout() const {
        return was_timeout;
    }
 
    bool hasStarted() const {
        return has_started;
    }
 
    void killChildTaskWhenFinished() {
        kill_children = true;
    }
 
    virtual SocketConnection *newClient(int, const char *, uint16_t,
                                        ServerSocket *) {
        return nullptr;
    }
 
    virtual bool adoptConnection(Socket *conn);
 
    virtual void connAdded(SocketConnection *) {
    }
 
    virtual void connRemoved(SocketConnection *) {
    }
 
    virtual void serverAdded(ServerSocket *) {
    }
 
    virtual void serverRemoved(ServerSocket *) {
    }
 
    std::string result() const {
        return the_result;
    }
 
    std::set<Socket *> getMyConnections() const;
 
    bool isActive(Socket *conn) const {
        return supervisor->isActive(conn);
    }
 
    void wakeUp();
 
    bool wakeUpConnection(SocketConnection *s) {
        return supervisor->wakeUpConnection(s);
    }
 
    void cancelConnection(SocketConnection *s) {
        supervisor->cancelConnection(s);
    }
 
    std::string message() const {
        return the_message;
    }
 
    bool startObserving(Task *to) {
        return supervisor->startObserving(this, to);
    }
 
    void executeHandler(Task *receiver, const std::string &message) {
        if (supervisor->isObserving(this, receiver) && !receiver->terminated())
            receiver->handleExecution(this, message);
        else
            log() << "Will not call handleExecution since task isn't observed.";
    }
 
    double elapsed() const {
        return secondsSince(start_time);
    }
#ifndef _WIN32
    virtual Task *createWorkerTask(unsigned int wno) {
        log() << "missing createWorkerTask, cannot create worker " << wno;
        return nullptr;
    }
 
    virtual void finishWorkerTask(unsigned int ) {
    }
 
    virtual void newWorkerChannel(SocketReceiver *, unsigned int ) {
    }
 
    virtual void workerMessage(SocketReceiver *, const char *buf, size_t len) {
        log() << "Worker message: " << std::string(buf, len);
    }
#endif
 
    virtual PollState connectionReady(SocketConnection * /* conn */);
 
 
    virtual PollState msgFromConnection(SocketConnection * /* conn */,
                                        const std::string & /* msg */);
 
    uint64_t bytesSent() const {
        return tot_bytes_sent;
    }
 
    uint64_t bytesReceived() const {
        return tot_bytes_received;
    }
 
    void resetByteCount() {
        tot_bytes_sent = 0;
        tot_bytes_received = 0;
    }
 
    void notifyBytesSent(uint64_t n) {
        tot_bytes_sent += n;
    }
 
    void notifyBytesReceived(uint64_t n) {
        tot_bytes_received += n;
    }
 
protected:
 
    bool addConnection(SocketConnection *conn);
 
    bool addConnected(SocketConnection *conn);
 
    bool addServer(ServerSocket *conn);
 
    bool parseListen(const TaskConfig &tc, const std::string &log_label);
 
#ifdef USE_GNUTLS
    virtual bool tlsSetKey(ServerSocket *conn, const std::string &crt_path,
                   const std::string &key_path, const std::string &password) {
        return supervisor->tlsSetKey(conn, crt_path, key_path, password);
    }
#endif
 
    void setResult(const std::string &res);
 
    virtual void setError(const std::string &msg) {
        log() << "Task failure: " << msg;
        was_error = true;
        setResult("");
    }
 
    virtual void setTimeout() {
        was_timeout = true;
        setResult("");
    }
 
    void setMessage(const std::string &msg);
 
    virtual void taskFinished(Task *task) {
        dbg_log() << "Task " << task->label() << " died, no handler defined.";
    }
 
    virtual void taskMessage(Task *task) {
        dbg_log() << "No taskMessage handler implemented for " << task->label();
    }
 
    virtual void handleExecution(Task *sender, const std::string &message) {
        dbg_log() << "Event " << message << " from "
                  << (sender ? sender->label() : "unknown")
                  << ", no handler implemented";
    }
 
    bool terminated() const {
        return is_finished;
    }
 
    void addNewTask(Task *task, Task *parent = nullptr) {
        supervisor->addTask(task, parent);
    }
 
#ifdef USE_THREADS
    void addNewThread(Task *task, const std::string &name="ThreadLoop",
                      std::ostream *log_file = nullptr,
                      Task *parent = nullptr) {
        supervisor->spawnThread(task, name, log_file, parent);
    }
#endif
 
    void getMyTasks(std::set<Task *> &tset) {
        supervisor->getChildTasks(tset, this);
    }
 
    void abortMyTasks() {
        supervisor->abortChildTasks(this);
    }
 
    void abortTask(Task *task) {
        supervisor->abortTask(task);
    }
 
    void abortAllTasks() {
        supervisor->abort();
    }
 
    int runProcess(const char *const argv[]);
 
    virtual void processFinished(int pid, int wstatus);
 
#ifndef _WIN32
    WorkerProcess *createWorker(std::ostream *log_file = nullptr,
                                unsigned int channels = 1,
                                unsigned int wno = 0) {
        return supervisor->createWorker(this, log_file, channels, wno);
    }
 
    WorkerProcess *createWorker(const std::string &log_file_name,
                                unsigned int channels = 1,
                                unsigned int wno = 0) {
        return supervisor->createWorker(this, log_file_name, channels, wno);
    }
#endif
 
private:
    friend class EventLoop;
    void setTerminated() {
        is_finished = true;
    }
    // Called by EventLoop when task is scheduled for execution:
    double begin() {
        has_started = true;
        start_time = timeNow();
        return this->start();
    }
 
#ifdef USE_THREADS
    thread_local
#endif
    static EventLoop *supervisor;
    TimePoint start_time;
    std::string the_result;
    std::string the_message;
    uint64_t tot_bytes_sent = 0, tot_bytes_received = 0;
    bool is_finished = false;
    bool has_started = false;
    bool was_killed = false, was_error = false, was_timeout = false;
    bool kill_children = false;
    bool is_child_thread = false;
};