Daemon.cpp

//
// This file is part of the LOTI distribution (https://github.com/levy/loti/).
// Copyright (c) 2018 Levente Mészáros.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, version 3.
//
// This program is distributed in the hope that it will be useful, but
// WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
// General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//

#include "Daemon.h"

namespace loti {

Define_Module(Daemon);

Daemon::Daemon()
    : scheduler_([this](cMessage *m, simtime_t t) { scheduleAt(t, m); },
                 [this](cMessage *m) { cancelEvent(m); }),
      transport_(socket_, kPort),
      rng_(this),
      telemetry_(this)
{
}

Daemon::~Daemon()
{
    cancelEvent(&createClockEventTimer_);
}

void Daemon::initialize(int stage)
{
    if (stage == INITSTAGE_LOCAL) {
        nodeId_ = static_cast<domain::NodeId>(getId());
        numChains_ = par("clockChainCount").intValue();
        chainFactor_ = par("clockChainFactor").intValue();
        const std::size_t keep = static_cast<std::size_t>(par("clockChainKeep").intValue());
        NodeConfig config;
        // Multi-resolution clock chains. All intervals are 0 — this module drives the ticks off
        // its single fast timer (so the fast chain keeps re-sampling the volatile
        // createClockEventInterval; coarser chains piggyback every clockChainFactor^L firings),
        // and node_->start() therefore schedules no core clock timers, only the purge timer.
        // keep > 0 enables per-chain ring pruning, bounding total clock-event storage.
        for (int ch = 0; ch < numChains_; ++ch)
            config.chains.push_back(ChainConfig{/*interval=*/0, keep});
        simtime_t expiry(par("discoveryExpiryTime").doubleValue());
        config.discovery_expiry = expiry.raw();
        // Discovery forwarding policy (Part 5): static single next hop, or the time-dependent flood.
        config.discovery_hop_limit = static_cast<std::uint32_t>(par("discoveryHopLimit").intValue());
        config.discovery_fanout = static_cast<std::size_t>(par("discoveryFanout").intValue());
        config.discovery_forward_cap = static_cast<std::uint32_t>(par("discoveryForwardCap").intValue());
        config.discovery_routing = (par("discoveryRouting").stdstringValue() == "flood")
                                       ? DiscoveryRouting::flood
                                       : DiscoveryRouting::static_shortest_path;
        node_ = std::make_unique<Node>(
            nodeId_, NodePorts{clock_, scheduler_, transport_, rng_, signer_, telemetry_, store_},
            config);
        clockEventsRetainedSignal_ = registerSignal("clockEventsRetained");
        createClockEventTimer_.setName("CreateClockEventTimer");
        scheduleCreateClockEventTimer();
        node_->start();
        WATCH(nodeId_);
    }
    else if (stage == INITSTAGE_APPLICATION_LAYER) {
        socket_.setOutputGate(gate("socketOut"));
        socket_.setCallback(this);
        socket_.bind(kPort);
    }
}

void Daemon::handleMessage(cMessage *message)
{
    if (message == &createClockEventTimer_) {
        node_->create_clock_event(0);  // the fastest chain ticks on every timer firing
        ++clockTickCount_;
        // Coarser chains piggyback: chain L ticks every chainFactor^L fast ticks.
        long modulus = 1;
        for (int ch = 1; ch < numChains_; ++ch) {
            modulus *= chainFactor_;
            if (clockTickCount_ % modulus == 0)
                node_->create_clock_event(static_cast<std::uint32_t>(ch));
        }
        emit(clockEventsRetainedSignal_, static_cast<long>(node_->clock_event_count()));
        scheduleCreateClockEventTimer();
    }
    else if (scheduler_.owns(message))
        scheduler_.fire(message);
    else if (socket_.belongsToSocket(message))
        socket_.processMessage(message);
    else
        throw cRuntimeError("Unknown message");
}

void Daemon::socketDataArrived(UdpSocket *socket, Packet *packet)
{
    const auto& chunk = packet->peekDataAsBytes();
    node_->on_packet_received(chunk->getBytes());
    delete packet;
}

void Daemon::scheduleCreateClockEventTimer()
{
    scheduleAt(simTime() + par("createClockEventInterval"), &createClockEventTimer_);
}

domain::Event Daemon::publishEvent(const domain::Bytes& data)
{
    Enter_Method_Silent();
    return node_->publish_event(data);
}

void Daemon::discoverEventChain(const domain::Event& event, domain::TimeRange range, ChainCallback& callback)
{
    Enter_Method_Silent();
    node_->discover_event_chain(event, range, callback);
}

void Daemon::discoverEventBounds(const domain::Event& event, domain::TimeRange range, BoundsCallback& callback)
{
    Enter_Method_Silent();
    node_->discover_event_bounds(event, range, callback);
}

void Daemon::discoverEventOrder(const domain::Event& event1, const domain::Event& event2,
                                domain::TimeRange range, OrderCallback& callback)
{
    Enter_Method_Silent();
    node_->discover_event_order(event1, event2, range, callback);
}

void Daemon::learnRoute(domain::NodeId destination, domain::NodeId nextHop)
{
    Enter_Method_Silent();
    node_->learn_route(destination, nextHop);
}

void Daemon::addNeighbor(domain::NodeId id, const L3Address& address)
{
    Enter_Method_Silent();
    node_->add_neighbor(id);
    transport_.set_address(id, address);
}

}