Daemon_impl.h

Go to the documentation of this file.

/***************************************
        Auteur : Pierre Aubert
        Mail : pierre.aubert@lapp.in2p3.fr
        Licence : CeCILL-C
****************************************/
 
#ifndef __DAEMON_H_IMPL__
#define __DAEMON_H_IMPL__
 
#include "phoenix_assert.h"
#include "Daemon.h"
 
using namespace Swarm;

template<class _TBackend>
Daemon<_TBackend>::Daemon()
        :BaseDaemon()
{
        initialisationDaemon();
}

template<class _TBackend>
Daemon<_TBackend>::~Daemon(){
        
}


template<class _TBackend>
void Daemon<_TBackend>::setSocketMode(PSocketMode::PSocketMode mode){
        p_backend.socketManager.setMode(mode);
        switch(mode){
                case PSocketMode::MOCK:
                        p_isFullMock = true;
                        p_isFullMockRecord = false;
                        break;
                case PSocketMode::MOCK_RECORD:
                        p_isFullMock = false;
                        p_isFullMockRecord = true;
                        break;
                default:
                        p_isFullMock = false;
                        p_isFullMockRecord = false;
                        break;
        }
}


template<class _TBackend>
void Daemon<_TBackend>::setClockMode(PClockMode::PClockMode mode){
        p_backend.clock.setMode(mode);
        switch(mode){
                case PClockMode::MOCK:
                        p_isFullMock = true;
                        p_isFullMockRecord = false;
                        break;
                case PClockMode::MOCK_RECORD:
                        p_isFullMock = false;
                        p_isFullMockRecord = true;
                        break;
                default:
                        p_isFullMock = false;
                        p_isFullMockRecord = false;
                        break;
        }
}


template<class _TBackend>
PRecvStatus::PRecvStatus Daemon<_TBackend>::checkRecvStatus(PRecvStatus::PRecvStatus recvStatus){
        if(recvStatus == PRecvStatus::SOCKET_NOT_AVAILABLE){
//              p_log.errorAndThrow<Phoenix::SocketStatusException>("Daemon::checkRecvStatus : Socket not available", "Socket not available");
                p_log.getLogWarning() << "Daemon::checkRecvStatus : Socket not available" << std::endl;
        }
        if(recvStatus == PRecvStatus::SIGNAL_INTERRUPTION){
                p_log.criticalAndThrow<Phoenix::SocketStatusException>("Daemon::checkRecvStatus : Socket Backend caught a signal, aborting", "Socket Backend caught a signal, aborting");
        }
        if(recvStatus == PRecvStatus::BROKEN_BACKEND){
                p_log.criticalAndThrow<Phoenix::SocketStatusException>("Daemon::checkRecvStatus : Back-end is in invalid state", "Back-end is in invalid state");
        }
        if(recvStatus == PRecvStatus::BROKEN_SOCKET){
                p_log.criticalAndThrow<Phoenix::SocketStatusException>("Daemon::checkRecvStatus : Socket is in invalid state", "Socket is in invalid state");
        }
        if(recvStatus == PRecvStatus::CANNOT_DESERIALIZE_DATA){
                p_log.errorAndThrow<Phoenix::SocketStatusException>("Daemon::checkRecvStatus : Cannot deserialize data", "Cannot deserialize data");
        }
        return recvStatus;
}
 


template<class _TBackend>
void Daemon<_TBackend>::checkSendStatus(PSendStatus::PSendStatus sendStatus){
        if(sendStatus == PSendStatus::SOCKET_NOT_AVAILABLE){
                p_log.errorAndThrow<Phoenix::SocketStatusException>("Daemon::checkSendStatus : Socket not available", "Socket not available");
        }
        if(sendStatus == PSendStatus::NO_ROUTE_TO_RECEIVER){
                p_log.errorAndThrow<Phoenix::SocketStatusException>("Daemon::checkSendStatus : Receiver not reachable", "Receiver not reachable");
        }
        if(sendStatus == PSendStatus::SIGNAL_INTERRUPTION){
                p_log.criticalAndThrow<Phoenix::SocketStatusException>("Daemon::checkSendStatus : Socket Backend caught a signal, aborting.", "Socket Backend caught a signal, aborting.");
        }
        if(sendStatus == PSendStatus::BROKEN_BACKEND){
                p_log.criticalAndThrow<Phoenix::SocketStatusException>("Daemon::checkSendStatus : Back-end is in invalid state", "Back-end is in invalid state");
        }
        if(sendStatus == PSendStatus::BROKEN_SOCKET){
                p_log.criticalAndThrow<Phoenix::SocketStatusException>("Daemon::checkSendStatus : Socket is in invalid state", "Socket is in invalid state");
        }
        if(sendStatus == PSendStatus::CANNOT_SERIALIZE_DATA){
                p_log.errorAndThrow<Phoenix::SocketStatusException>("Daemon::checkSendStatus : Cannot serialize data", "Cannot serialize data");
        }
}


template<class _TBackend>
bool Daemon<_TBackend>::run(){
        p_isRun = true;
        //Let's initialise all connexions
        initialisationDaemonSocket();
        p_startTimestamp = p_backend.clock.now();
        p_statTimer.setEllapsedTime(p_config.getStatTimerPeriodMs());
        p_statTimer.setStartTime(p_startTimestamp);
        while(p_isRun){
                //Let's get the last received message
                //If there is not message, we treat the rest of the loop
                Message message;
                PRecvStatus::PRecvStatus recvStatus = checkRecvStatus(p_backend.socketManager.recvData("pull", message, p_mainConfig.recvFlag));
                if(recvStatus == PRecvStatus::OK){
                        processInputMessage(message);
                }
                extraLoopProcessing();
 
                time_t currentTime = p_backend.clock.now();
                checkMessageTimeout(currentTime);
                if(p_statTimer.isTime(currentTime) || !p_isRun){
                        sendStatToStatDaemon(currentTime);
                }
        }
        //Let's execute a last command before stop
        executeOnStop();
        return true;
}


template<class _TBackend>
bool Daemon<_TBackend>::sendMessage(const Message & message){
        bool b(true);
        const PVecString & vecDestination = message.getVecRecver();
        for(PVecString::const_iterator it(vecDestination.begin()); it != vecDestination.end(); ++it){
                b &= sendMessage(*it, message);
        }
        return b;
}


template<class _TBackend>
bool Daemon<_TBackend>::sendMessage(const PString & destinationName, const Message & message){
        if(message.getIsConfirmationNeeded()){
                addMessageToConfirm(message);
        }
        checkSendStatus(p_backend.socketManager.sendData(destinationName, message, p_mainConfig.sendFlag));
        return true;
}


template<class _TBackend>
bool Daemon<_TBackend>::sendData(const PString & destinationName, const Data & data, bool isConfirmationNeeded){
        Message message;
        message.setSendTime(p_backend.clock.now());
        message.setData(data);
        message.setId(getMessageId());
        message.setIsConfirmationNeeded(isConfirmationNeeded);
        message.setType(MessageType::RESULT_DATA);
        message.setSender(p_config.getName());
        message.getVecRecver().push_back(destinationName);
        return sendMessage(destinationName, message);
}

template<class _TBackend>
void Daemon<_TBackend>::extraLoopProcessing(){}

template<class _TBackend>
void Daemon<_TBackend>::executeOnStop(){}

template<class _TBackend>
void Daemon<_TBackend>::initialisationDaemon(){
        
}

template<class _TBackend>
void Daemon<_TBackend>::initialisationDaemonSocket(){
        if(p_config.getName() == ""){
                getLog().criticalAndThrow<Phoenix::ConfigException>("Daemon::initialisationDaemonSocket", "daemon has no name");
        }
        if(p_isFullMockRecord){
                setSocketMode(PSocketMode::MOCK_RECORD);
                setClockMode(PClockMode::MOCK_RECORD);
        }else{
                if(p_isFullMock){
                        setSocketMode(PSocketMode::MOCK);
                        setClockMode(PClockMode::MOCK);
                }
        }
        //Let's send when we are a client and recv when we are a server
        //By doing this way, we have one server per Daemon and as many clients as other Daemon
        getLog().getLogInfo() << "Daemon<_TBackend>::initialisationDaemonSocket() : initialise pull socket for daemon '" << p_config.getName() << "'" << std::endl;
        phoenix_assert(p_backend.socketManager.addServerSocket("pull",
                        PSocketParam{p_config.getHostName(), p_config.getReceivingPort(), p_mainConfig.recvTimeoutMs, p_mainConfig.sendTimeoutMs}, p_backend.extraServerParam,
                        "./" + p_config.getName() + "_", p_backend.extraMockServerParam));
        
        //Now let's initialise connexions to other
        for(MapDaemonConfig::iterator it(p_mapDaemon.begin()); it != p_mapDaemon.end(); ++it){
                if(it->second.getName() == p_config.getName()){continue;}
                getLog().getLogInfo() << "Daemon<_TBackend>::initialisationDaemonSocket() : initialise send socket for neighbour Daemon '"<<it->second.getName()<<"'" << std::endl;
                phoenix_assert(p_backend.socketManager.addClientSocket(it->second.getName(),
                        PSocketParam{it->second.getHostName(), it->second.getReceivingPort(), p_mainConfig.recvTimeoutMs, p_mainConfig.sendTimeoutMs}, p_backend.extraClientParam,
                        "./" + p_config.getName() + "_", p_backend.extraMockClientParam));
                // p_backend.socketManager.getSocket(it->second.getName())->waitUntilConnection(500000lu, 30lu);        //0.5 s per check and 30 checks
        }
        getLog().getLogInfo() << "Daemon<_TBackend>::initialisationDaemonSocket() : all connection to neighbours Daemon initialised" << std::endl;
        p_backend.clock.setMockPrefix("./" + p_config.getName() + "_");
}


template<class _TBackend>
void Daemon<_TBackend>::processInputMessage(Message & message){
        //Check is the message needs a confirmation
        if(message.getIsConfirmationNeeded()){
                Message confirmation;
                //We do not need to change the time of the message because it has to be in the clock reference of the sender
                confirmation.setSendTime(message.getSendTime());
                confirmation.setId(message.getId());
                confirmation.setSender(p_config.getName());
                confirmation.setType(MessageType::MESSAGE_CONFIRMATION);
                confirmation.getVecRecver().push_back(message.getSender());
                sendMessage(confirmation);
        }
        
        //Let's check is we receive a stop message
        if(message.getType() == MessageType::STOP){stop();}
        else if(message.getType() == MessageType::MESSAGE_CONFIRMATION){
                processConfirmedMessage(message.getId(), p_backend.clock.now());
        }else if(message.getType() == MessageType::RESULT_DATA){
                AbstractDataFunction* function = getDataFunction(message.getData());
                if(function == NULL){
                        return;
                }
                time_t functionStartTime = p_backend.clock.now();
                if(!function->call(p_log, message.getData())){
                        getLog().getLogError() << "Daemon<_TBackend>::processInputMessage : cannot process data of type '"<<message.getData().getType()<<"' from Daemon '"<<message.getSender()<<"'" << std::endl;
                        return;
                }
                time_t functionEllapsedTime = p_backend.clock.now() - functionStartTime;
                getLog().getLogDebug() << "Daemon<_TBackend>::processInputMessage : function '"<<message.getData().getType()<<"' processed data from daemon '"<<message.getSender()<<"' in "<<functionEllapsedTime<<" nanoseconds" << std::endl;
                if(p_config.getStatDaemonName() != "" && p_config.getStatDaemonName() != p_config.getName()){
                        updateStatAccumulator(p_config.getDaemonStatAccumulator().getMapStatComputing().find(message.getData().getType())->second, functionEllapsedTime);
                }
        }
}


template<class _TBackend>
void Daemon<_TBackend>::sendStatToStatDaemon(time_t currentTime){
        //Let's fill the Stat
        Stat stat;
        fillDaemonStat(stat, p_startTimestamp, currentTime);
        stat.setName(p_config.getName());
 
        //Let's send it to the Stat Daemon
        if(p_config.getStatDaemonName() != "" && p_config.getStatDaemonName() != p_config.getName()){
                getLog().getLogDebug() << "Daemon::sendStatToStatDaemon() : sending statistics to Stat Daemon '"<<p_config.getStatDaemonName()<<"'" << std::endl;
                sendValue(p_config.getStatDaemonName(), stat, true);
        }
 
        clearStat();
        p_startTimestamp = currentTime;
}
 
#endif