PersistencyManager.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

#include <fesa-core/Persistency/PersistencyManager.h>

#include <fesa-core/Persistency/PersistencyUnit.h>
#include <fesa-core/Core/AbstractEquipment.h>
#include <fesa-core/Core/ThreadPriorityConfiguration.h>
#include <fesa-core/DataStore/GlobalDevice.h>
#include <fesa-core/Exception/FesaException.h>
#include <fesa-core/Utilities/Lock.h>
#include <fesa-core/Utilities/ProcessConfiguration.h>
#include <fesa-core/Utilities/XMLParser.h>

#include <cmw-log/Logger.h>

#include <iostream>


namespace
{

CMW::Log::Logger& logger = CMW::Log::LoggerFactory::getLogger("FESA.FWK.fesa-core.Persistency.PersistencyManager");

}


namespace fesa
{

PersistencyManager* PersistencyManager::theInstance_ = NULL;

void PersistencyManager::registerPersistencyUnit(PersistencyUnit * pPersistencyUnit)
{

    const std::string& name = pPersistencyUnit->getGlobalDevice().className.getAsString();
    persistencyUnitsCol_.insert(std::pair<const std::string, PersistencyUnit *>(name, pPersistencyUnit));
}

PersistencyManager::PersistencyManager() :
    isStarted_(false), counterOfTriggers_(0)
{
    persistencyMaxDelay_ = AbstractEquipment::getInstance()->getProcessConfiguration()->getIntValue(PropertyTag::MAX_PERSISTENCY_DELAY);
}

PersistencyManager::~PersistencyManager()
{

    for (std::map<const std::string, PersistencyUnit*>::iterator p = persistencyUnitsCol_.begin(); p != persistencyUnitsCol_.end(); ++p)
    {
        delete p->second;
    }
}

PersistencyManager* PersistencyManager::getInstance()
{
    if (!theInstance_)
    {
        theInstance_ = new PersistencyManager();
    }
    return theInstance_;
}

void PersistencyManager::start()
{
    {
        AbstractEquipment* eqp = AbstractEquipment::getInstance();
        std::string instanceFile = eqp->getDeviceDataFileName();
        XMLParser xmlParser(instanceFile, false);
        ThreadPriorityConfigurationFromFile conf(xmlParser, eqp->getProcessConfiguration());
        int32_t prio =  conf.getPrioPersistence();
        setPriority(prio);
    }

    if (!isStarted_)
    {
        const std::string threadName("PersistencyManager");
        Thread::start(false, threadName);
        isStarted_ = true;
    }
}

void PersistencyManager::trigger(const std::string& className)
{
    std::map<const std::string, PersistencyUnit*>::iterator pos = persistencyUnitsCol_.find(className);
    if (pos == persistencyUnitsCol_.end())
    {
        return;
    }
    {
        Lock lock(accessMutex_);
        persistencySet_.insert(persistencyUnitsCol_[className]);
        counterOfTriggers_++;
    }
    triggerCondVar_.signal();
}

void PersistencyManager::run()
{
    bool persistencyTimeout = false;
    uint32_t persistencyDelay = 0;
    uint32_t avalancheTimeout = 30;

    struct timespec delay, time_left_before_wakeup;
    delay.tv_sec = avalancheTimeout;
    delay.tv_nsec = 0;

    while (isRunning_)
    {
        {
            Lock lock(accessMutex_);
            while (persistencySet_.size() == 0)
            {
                triggerCondVar_.wait(accessMutex_);
            }
            // timeout in order to protect against an avalanche of changes
        }
        uint32_t lastCounterValue = 0;
        do
        {
            lastCounterValue = counterOfTriggers_;
            // wait avalance timeout before triggering the persistency
            nanosleep(&delay, &time_left_before_wakeup);
            persistencyDelay += avalancheTimeout;
            /* In case of periodic storeFieldssettings, we could never exit from this avalanche filtering loop.
             * To prevent such a situation (no storing any more), we define a maximum delay (~5mn).
             */
            if (persistencyDelay > persistencyMaxDelay_)
            {
                persistencyDelay = 0;
                persistencyTimeout = true;
            }
        }
        while ((lastCounterValue != counterOfTriggers_) && (persistencyTimeout == false));
        {
            Lock lock(accessMutex_);
            persistencyTimeout = false;
            counterOfTriggers_ = 0;
            while(persistencySet_.size() > 0)
            {
                // extract the first element
                std::set<PersistencyUnit*>::iterator puItr = persistencySet_.begin();
                PersistencyUnit* pu = (*puItr);
                persistencySet_.erase(puItr);
                try
                {
                    storeManager_.store(*pu);
                }
                catch (FesaException& e)
                {
                    LOG_ERROR_IF(logger, e.getMessage());
                }
            }
        }
    }
    hasFinished_ = true;
}

} // fesa