BaseDaemon.cpp

Go to the documentation of this file.

/***************************************
        Auteur : Pierre Aubert
        Mail : pierre.aubert@lapp.in2p3.fr
        Licence : CeCILL-C
****************************************/
 
#include <cmath>        //for std::nan
#include "BaseDaemon.h"
#include "PTimer.h"
 
using namespace Swarm;

BaseDaemon::BaseDaemon()
        :p_optionParser(true, __PROGRAM_VERSION__)
{
        initialisationBaseDaemon();
}

BaseDaemon::~BaseDaemon(){
        clearCallableMethod();
}


bool BaseDaemon::parseArgument(int argc, char** argv){
        p_optionParser.parseArgument(argc, argv);
        const OptionMode & defaultMode = p_optionParser.getDefaultMode();
        PPath configurationFile;
        defaultMode.getValue(configurationFile, "daemonconfig");
        PString daemonName;
        defaultMode.getValue(daemonName, "daemonname");
        p_isFullMock = defaultMode.isOptionExist("mock");
        p_isFullMockRecord = defaultMode.isOptionExist("mockrecord");
        return load(configurationFile, daemonName);
}


void BaseDaemon::loadConfigFromNode(const ConfigNode & dico, const PString & daemonName) {
        const ConfigNode * mapMainConfig = dico.getChild("swarm");
        if(mapMainConfig != nullptr){
                p_mainConfig.recvTimeoutMs = phoenix_get_value<int>(*mapMainConfig, "recv_timeout_ms", -1);
                p_mainConfig.sendTimeoutMs = phoenix_get_value<int>(*mapMainConfig, "send_timeout_ms", -1);
                p_mainConfig.recvFlag = fromString<PRecvFlag::PRecvFlag>(phoenix_get_string(*mapMainConfig, "recv_flag", "NON_BLOCK"));
                p_mainConfig.sendFlag = fromString<PSendFlag::PSendFlag>(phoenix_get_string(*mapMainConfig, "send_flag", "NON_BLOCK"));
        }
        std::map<PString, time_t> mapTimeout;
        const ConfigNode * extraConfigParam = nullptr;
        daemon_read_configNode(p_config, p_mapDaemon, p_log, mapTimeout, extraConfigParam, dico, daemonName);
        
        const ConfigNode * mapStatDaemon = dico.getChild("statistics");
        if(mapStatDaemon != nullptr){
                p_config.setStatDaemonName(daemon_loadString(*mapStatDaemon, "stat_daemon_name"));
                extraLoad(mapStatDaemon);
        }
        p_config.setMapTimeout(mapTimeout);
        if(p_config.getStatDaemonName() != ""){
                if(!isDaemonExist(p_config.getStatDaemonName())){
                        p_log.getLogError() << "BaseDaemon::loadConfigFromNode : stat daemon '"<<p_config.getStatDaemonName()<<"' not found in configuration" << std::endl;
                }
        }
 
        p_log.getLogInfo() << "BaseDaemon::loadConfigFromNode : loading configuration successful" << std::endl;
        p_log.getLogInfo() << "BaseDaemon::loadConfigFromNode : ready to call addCallMethod to subscribe on received data" << std::endl;
        addCallMethod();
        if(extraConfigParam != nullptr){
                extraLoad(extraConfigParam);
        }
}


bool BaseDaemon::load(const PString & configFileContent, const PString & daemonName, ConfigFormat::ConfigFormat format){
        ConfigNode dico;
        daemon_load_config(p_log, dico, configFileContent, format);
        loadConfigFromNode(dico, daemonName);
        return true;
}


bool BaseDaemon::load(const PPath & fileName, const PString & daemonName){
        ConfigNode dico;
        daemon_load_config(p_log, dico, fileName);
        loadConfigFromNode(dico, daemonName);
        return true;
}


bool BaseDaemon::extraLoad(const ConfigNode * config){
        return true;
}

void BaseDaemon::addCallMethod(){
        if(p_log.isOpen()){
                p_log.getLogWarning() << "BaseDaemon::addCallMethod : no redefined addCallMethod " << std::endl;
        }else{
                std::cerr << "BaseDaemon::addCallMethod : no redefined addCallMethod (and no log file so we use std::cerr)" << std::endl;
        }
}


bool BaseDaemon::callMethod(Data & result, const PString & name, const Data & parameter){
        std::map<PString, AbstractFunction*>::iterator it(p_mapCallableMethod.find(name));
        if(it == p_mapCallableMethod.end()){
                getLog().getLogError() << "BaseDaemon::callMethod : function '"<<name<<"' not found" << std::endl;
                return false;
        }
        return p_mapCallableMethod[name]->call(p_log, result, parameter);
}

void BaseDaemon::clearCallableMethod(){
        for(std::map<PString, AbstractFunction*>::iterator it(p_mapCallableMethod.begin()); it != p_mapCallableMethod.end(); ++it){
                delete it->second;
        }
        p_mapCallableMethod.clear();
}

void BaseDaemon::stop(){
        p_log.getLogInfo() << "Stopping Daemon '"<<p_config.getName()<<"' at '"<<p_config.getHostName()<<"'" << std::endl;
        p_isRun = false;
}


OptionParser & BaseDaemon::getOptionParser(){
        return p_optionParser;
}

///Get the log of the current BaseDaemon

PLog & BaseDaemon::getLog(){
        return p_log;
}


bool BaseDaemon::isDaemonExist(const PString & name) const{
        MapDaemonConfig::const_iterator it(p_mapDaemon.find(name));
        if(it == p_mapDaemon.end()){
                //Then if the stat daemon does not exist in the map, it may be the current daemon
                if(p_config.getName() == name){
                        return true;
                }
        }
        if(it != p_mapDaemon.end()){
                return true;
        }
 
        return false;
}


void BaseDaemon::addMessageToConfirm(const Message & message){
        p_mapMessageToBeConfirmed[message.getId()] = message;
}


bool BaseDaemon::getMessageToConfirm(Message & message, size_t id) const{
        std::map<size_t, Message>::const_iterator it(p_mapMessageToBeConfirmed.find(id));
        if(it != p_mapMessageToBeConfirmed.end()){
                message = it->second;
                return true;
        }
        return false;
}


void BaseDaemon::processConfirmedMessage(size_t id, time_t currentTime){
        std::map<size_t, Message>::const_iterator it(p_mapMessageToBeConfirmed.find(id));
        if(it != p_mapMessageToBeConfirmed.end()){
                getLog().getLogInfo() << "BaseDaemon::processConfirmedMessage : adding statistics for processed transaction " << id << std::endl;
                const Message & message = it->second;
                time_t ellapsedTime = currentTime - message.getSendTime();
                for(const PString & receiver : message.getVecRecver()){
                        getLog().getLogDebug() << "BaseDaemon::processConfirmedMessage : add statistic for receiver : " << receiver << std::endl;
                        getCommunicationStat(receiver, message.getData().getType(), ellapsedTime, p_config.getStatNbBin(), p_config.getStatHistLowerBound(), p_config.getStatHistUpperBound());
                }
                // Remove the confirmed message
                getLog().getLogInfo() << "BaseDaemon::removeConfirmedMessage : remove confirmed transaction " << id << std::endl;
                p_mapMessageToBeConfirmed.erase(it);
        }else{
                getLog().getLogError() << "BaseDaemon::processConfirmedMessage : transaction " << id << " not found in p_mapMessageToBeConfirmed" << std::endl;
        }
}


void BaseDaemon::checkMessageTimeout(time_t currentTime){
        std::vector<size_t> messagesToRemove;
        for(auto it = p_mapMessageToBeConfirmed.begin(); it != p_mapMessageToBeConfirmed.end(); ++it){
                const size_t & messageId = it->first;
                const Message & message = it->second;
                time_t elapsedTime = currentTime - message.getSendTime();
                if(message.getVecRecver().size() == 0lu){
                        getLog().getLogError() << "BaseDaemon::checkMessageTimeout() : message " << messageId << " has no receiver." << std::endl;
                        continue;
                }
                std::map<PString, time_t>::const_iterator timeoutIt = p_config.getMapTimeout().find(message.getVecRecver().front());
                if(timeoutIt == p_config.getMapTimeout().end()){
                        getLog().getLogError() << "BaseDaemon::checkMessageTimeout() : no timeout found for receiver '" << message.getVecRecver().front() << "' of message " << messageId << "." << std::endl;
                        continue;
                }
                time_t timeout = timeoutIt->second;
                phoenix_assert(message.getVecRecver().size() != 0lu);
                if (elapsedTime > timeout) {
                        messagesToRemove.push_back(messageId);
                        getLog().getLogWarning() << "BaseDaemon::checkMessageTimeout() : Timeout Reached for message " << messageId << " sent to '" << message.getVecRecver().front() << "'. Removing unconfirmed transaction" << std::endl;
                }
        }
        for(const size_t & messageId : messagesToRemove){
                p_mapMessageToBeConfirmed.erase(messageId);
        }
}


size_t BaseDaemon::getMessageId(){
        ++p_messageId;
        return p_messageId;
}


AbstractDataFunction* BaseDaemon::getDataFunction(const Data & data){
        PString prototype(data.getType());
        std::map<PString, AbstractDataFunction*>::iterator it(p_mapDataFunction.find(prototype));
        if(it == p_mapDataFunction.end()){
                getLog().getLogError() << "BaseDaemon::getDataFunction : function to process data '"<<prototype<<"' not found" << std::endl;
                return nullptr;
        }
        return it->second;
}


bool BaseDaemon::processData(const Data & data){
        AbstractDataFunction* function = getDataFunction(data);
        if(function == nullptr){
                return false;
        }
        getLog().getLogDebug() << "BaseDaemon::processData : process data '"<<data.getType()<<"'" << std::endl;
        bool b = function->call(p_log, data);
        return b;
}


void BaseDaemon::updateStatAccumulator(StatAccumulator & stat, float value){
        stat.setNbEvent(stat.getNbEvent() + 1lu);
        getLog().getLogDebug() << "BaseDaemon::updateStatAccumulator : event nb is " << stat.getNbEvent() << std::endl;
        if(stat.getNbEvent() == 1lu){
                stat.setMin(value);
                stat.setMax(value);
                stat.setSum(value);
        }else{
                if(value < stat.getMin()){
                        stat.setMin(value);
                }
                if(value > stat.getMax()){
                        stat.setMax(value);
                }
                stat.setSum(stat.getSum() + value);
        }
        if(value > stat.getHistUpperBound()){
                stat.setNbEventAboveUpperBound(stat.getNbEventAboveUpperBound() + 1lu);
        }
        if(value < stat.getHistLowerBound()){
                stat.setNbEventBelowLowerBound(stat.getNbEventBelowLowerBound() + 1lu);
        }
        float binSize = (stat.getHistUpperBound() - stat.getHistLowerBound()) / stat.getVecHistogram().size();
        size_t bin = (value - stat.getHistLowerBound()) / binSize;
        if(bin < stat.getVecHistogram().size()){
                std::vector<size_t> & vecHistogram = stat.getVecHistogram();
                vecHistogram[bin] = vecHistogram[bin] + 1lu;
        }
}


StatAccumulator BaseDaemon::createNewStat(size_t nbBin, float histLowerBound, float histUpperBound){
        StatAccumulator newStat;
        newStat.setNbEvent(0lu);
        newStat.setMin(std::nan(""));
        newStat.setMax(std::nan(""));
        newStat.setSum(std::nan(""));
        newStat.setNbEventAboveUpperBound(0lu);
        newStat.setNbEventBelowLowerBound(0lu);
        newStat.setHistUpperBound(histUpperBound);
        newStat.setHistLowerBound(histLowerBound);
        newStat.setVecHistogram(std::vector<size_t>(nbBin, 0lu));
        return newStat;
}


const std::map<size_t, Message> & BaseDaemon::getMapMessageToBeConfirmed() const {
        return p_mapMessageToBeConfirmed;
}
 

void BaseDaemon::clearStat(){
        for(MapStatAccumulator::iterator it = p_config.getDaemonStatAccumulator().getMapStatComputing().begin(); it != p_config.getDaemonStatAccumulator().getMapStatComputing().end(); ++it){
                it->second.setNbEvent(0lu);
                std::fill(it->second.getVecHistogram().begin(), it->second.getVecHistogram().end(), 0lu);
        }
        for(MapDaemonStatAccumulator::iterator itMap = p_config.getDaemonStatAccumulator().getMapStatCommunication().begin(); itMap != p_config.getDaemonStatAccumulator().getMapStatCommunication().end(); ++itMap){
                for(MapStatAccumulator::iterator it = itMap->second.begin(); it != itMap->second.end(); ++it){
                        it->second.setNbEvent(0lu);
                        std::fill(it->second.getVecHistogram().begin(), it->second.getVecHistogram().end(), 0lu);
                }
        }
}


DaemonConfig & BaseDaemon::getConfig(){
        return p_config;
}


MapDaemonConfig & BaseDaemon::getMapDaemonConfig(){
        return p_mapDaemon;
}


DaemonMainConfig & BaseDaemon::getMainConfig(){
        return p_mainConfig;
}


PLog & BaseDaemon::getLogger(){
        return p_log;
}

void BaseDaemon::initialisationBaseDaemon(){
        p_optionParser.setExampleLongOption("phoenix_daemon --daemonconfig=daemon_config.toml --daemonname=main");
        p_optionParser.setExampleShortOption("phoenix_daemon -c daemon_config.toml -n main");
        p_optionParser.addOption("daemonconfig", "c", OptionType::FILENAME, true, "Toml configuration file which define all Daemons of the swarm");
        p_optionParser.addOption("daemonname", "n", OptionType::STRING, true, "Name of the current Daemon of the swarm configuration defined with --daemonconfig");
        p_optionParser.addOption("mock", "m", OptionType::NONE, false, "Activate the Daemon in mock mode for the sockets and the clock");
        p_optionParser.addOption("mockrecord", "r", OptionType::NONE, false, "Activate the Daemon in mock record mode for the sockets and the clock");
        p_isRun = false;
        p_messageId = 0lu;
        p_isFullMock = false;
        p_isFullMockRecord = false;
}


std::map<PString, StatAccumulator> BaseDaemon::createNewCommunicationStatMap(const PString & dataType, size_t nbBin, float histLowerBound, float histUpperBound){
        std::map<PString, StatAccumulator> newStatMap;
        newStatMap[dataType] = createNewStat(nbBin, histLowerBound, histUpperBound);
        return newStatMap;
}


void BaseDaemon::getCommunicationStat(const PString & destName, const PString & dataType, float latency, size_t nbBin, float histLowerBound, float histUpperBound){
        std::map<PString, std::map<PString, StatAccumulator> > & mapCommunication = p_config.getDaemonStatAccumulator().getMapStatCommunication();
        MapDaemonStatAccumulator::iterator it = mapCommunication.find(destName);
        if(it == mapCommunication.end()){
                mapCommunication[destName] = createNewCommunicationStatMap(dataType, nbBin, histLowerBound, histUpperBound);
        }
        else{
                std::map<PString, StatAccumulator>::iterator itDataType = it->second.find(dataType);
                if(itDataType == it->second.end()){
                        it->second[dataType] = createNewStat(nbBin, histLowerBound, histUpperBound);
                }
        }
        updateStatAccumulator(p_config.getDaemonStatAccumulator().getMapStatCommunication().find(destName)->second[dataType], latency);
}


VecStat BaseDaemon::fillVecStat(const StatAccumulator & accumulator, time_t startTimestamp, time_t endTimestamp){
        VecStat vecStat;
        
        // Compute event values
        vecStat.getNbEvent().push_back(accumulator.getNbEvent());
        vecStat.getMin().push_back(accumulator.getMin());
        vecStat.getMax().push_back(accumulator.getMax());
        vecStat.getAverage().push_back(accumulator.getSum() / accumulator.getNbEvent());
 
        // Compute time values
        vecStat.getStartTimestamp().push_back(startTimestamp);
        vecStat.getEndTimestamp().push_back(endTimestamp);
        time_t rate = endTimestamp - startTimestamp;
        vecStat.getRate().push_back(accumulator.getNbEvent()*1e9 / rate);
        vecStat.getRateEventAboveUpperBound().push_back(accumulator.getNbEventAboveUpperBound()*1e9 / rate);
        vecStat.getRateEventBelowLowerBound().push_back(accumulator.getNbEventBelowLowerBound()*1e9 / rate);
        
        vecStat.getVecRateQuantile().resize(accumulator.getVecHistogram().size());
        for(size_t i = 0; i < accumulator.getVecHistogram().size(); ++i){
                vecStat.getVecRateQuantile()[i].push_back(accumulator.getVecHistogram()[i]*1e9 / rate);
        }
        return vecStat;
}


void BaseDaemon::fillDaemonStat(Stat & stat, time_t startTimestamp, time_t endTimestamp){
        // Fill the computing statistics
        MapStatAccumulator & mapStatComputing = p_config.getDaemonStatAccumulator().getMapStatComputing();
        for(MapStatAccumulator::iterator it = mapStatComputing.begin(); it != mapStatComputing.end(); ++it){
                StatAccumulator & accumulator = it->second;
                VecStat vecStat = fillVecStat(accumulator, startTimestamp, endTimestamp);
                stat.getMapStatComputing()[it->first] = vecStat;
        }
 
        // Fill the communication statistics
        MapDaemonStatAccumulator & mapStatCommunication = p_config.getDaemonStatAccumulator().getMapStatCommunication();
        for(MapDaemonStatAccumulator::iterator itMap = mapStatCommunication.begin(); itMap != mapStatCommunication.end(); ++itMap){
                const PString & destName = itMap->first;
                std::map<PString, VecStat> & mapVecStat = stat.getMapStatCommunication()[destName];
                for(MapStatAccumulator::iterator it = itMap->second.begin(); it != itMap->second.end(); ++it){
                        StatAccumulator & accumulator = it->second;
                        VecStat vecStat = fillVecStat(accumulator, startTimestamp, endTimestamp);
                        mapVecStat[it->first] = vecStat;
                }
        }
}