experiment.h

Go to the documentation of this file.

 
#pragma once
#include <comm/frameregistry.h>
#include <utils/deadline.h>
 
#include "kern.h"
#include "evented.h"
#include "logger.h"
 
extern class Experiment {
    struct ELog : Logger {
        const uint32_t &time;
        virtual Buffer<uint8_t> pre() override {
            Buffer<uint8_t> ret = 256;
            ret.len = snprintf((char*)ret.buf, ret.size,
                    "Experiment Logger (@%" PRIu32 "ms): ", time);
            return ret;
        }
        ELog(const uint32_t &time, Sink<Buffer<uint8_t>> &underlying)
                : Logger(underlying), time(time) {
        }
        void start(uint32_t) { info("started\n"); }
        void stop(uint32_t) { info("stopped\n"); }
        void timeout(uint32_t) { warn("timed out!\n"); }
    };
public:
    enum State {
        IDLE,
        RUN,
    };
    enum Event {
        INIT,
        STOP,
        TIMEOUT,
    };
    Experiment(FrameRegistry *fr=nullptr) {
        k.every(1, *this, &Experiment::tick);
        if (fr) registerWith(*fr);
        onEvent(INIT).call(log, &ELog::start);
        onEvent(STOP).call(log, &ELog::stop);
        onEvent(TIMEOUT).call(log, &ELog::timeout);
    }
    void registerWith(FrameRegistry &reg) {
        reg.setHandler(1, *this, &Experiment::handleFrame);
    }
    Schedule::Evented::Registry& onEvent(Event e) {
        switch (e) {
            case INIT: return init;
            case STOP: return stop;
            default: return timeout;
        }
    }
    Schedule::Recurring::Registry& during(State s) {
        switch (s) {
            case IDLE: return idle;
            default: return running;
        }
    }
    const State& state{state_};
    const uint32_t& time{time_};
    void setHeartbeatTimeout(uint32_t ms, Sink<Frame> &notify) {
        heartbeat.timeout = ms;
        heartbeat.reset(time);
        // notify other side if we timeout
        struct TMP :Schedule::Schedulable{
                Sink<Frame> &s;
                TMP(Sink<Frame>&s): s(s) {}
                void call() override{
                    auto f = Frame{1}.pack(false);
                    s.trypush(std::move(f));
                }};
        timeout.call(new TMP{notify});
    }
    ELog log{time, k.log};
 
private:
    State state_{};
    bool alive{};
    void statemachine() {
        State old = state;
        // very simple state machine
        state_ = alive ? RUN : IDLE;
        if (state != old) switch(old) { // handle state transitions == events
        case IDLE:
            time_ = 0;
            k.schedule(time, init);
            idle.reset(); running.reset();
            break;
        case RUN:
            k.schedule(time, stop);
            break;
        }
        switch (state) { // handle states == recurring
        case IDLE:
            k.schedule(time, idle);
            break;
        case RUN :
           k.schedule(time, running);
           if (heartbeat && heartbeat(time)) {
               k.schedule(time, timeout);
               alive = false;
           }
           break;
        }
    }
    Schedule::Evented::Registry init{}, stop{}, timeout{};
    Schedule::Recurring::Registry idle{}, running{};
    uint32_t time_{};
    class _: Deadline {
        friend Experiment;
        uint32_t timeout{};
        operator bool() {
            return timeout;
        }
        bool reset(uint32_t now) {
            when = now + timeout;
            return true;
        }
    } heartbeat{};
 
    void tick(uint32_t, uint32_t dt) {
        statemachine();
        time_ += dt;
    }
 
    void handleFrame(Frame &f) {
        assert(f.id == 1);
        struct {
            uint8_t alive:1;
            uint8_t heartbeat:1;
            uint8_t _:6;
        } b = f.unpack<decltype(b)>();
        if (b.heartbeat) {
            (bool)heartbeat && heartbeat.reset(time);
        } else {
            alive = b.alive;
        }
    }
} e;