RTController.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

#include <fesa-core/RealTime/RTController.h>
#include <fesa-core/RealTime/AbstractRTActionFactory.h>
#include <fesa-core/RealTime/AbstractEventSourceFactory.h>
#include <fesa-core/DataStore/AbstractDeviceFactory.h>
#include <fesa-core/Core/AbstractEquipment.h>
#include <fesa-core/Core/NotificationIDManager.h>
#include <fesa-core/Core/AbstractDeviceClass.h>
#include <fesa-core/Sorting/HomogeneousDevCol.h>
#include <fesa-core/Exception/FesaExceptionDef.h>
#include <fesa-core/RealTime/AbstractEventSourceFactory.h>
#include <fesa-core/RealTime/EventsMappingLoader.h>
#include <fesa-core/RealTime/AbstractRTEquipment.h>
#include <fesa-core/RealTime/RTScheduler.h>
#include <fesa-core/Diagnostic/FesaStream.h>
#include <fesa-core/Utilities/ParserElements/EventElement.h>
#include <fesa-core/Sorting/SortingInterpreter.h>
#include <fesa-core/RealTime/AbstractRTAction.h>

#include <iostream>
#include <sys/stat.h>
#include <vector>
#include <sstream>

#include <cmw-log/Logger.h>

namespace
{
CMW::Log::Logger& logger = CMW::Log::LoggerFactory::getLogger("FESA.FWK.fesa-core.RealTime");
}

namespace fesa
{

    RTController::RTController()
    {
    }

    // remove the resources
    // CRFESA-764 to be review in the shutdown procedure
    RTController::~RTController()
    {
        LOG_TRACE_IF(logger, "Removing controller");
        std::map<const std::string, RTScheduler*>::iterator schedulers;
        for (schedulers = schedulersCol_.begin(); schedulers != schedulersCol_.end(); schedulers++)
        {
            if (schedulers->second != NULL)
            {
                schedulers->second->stop();
            }
        }
        AbstractEventSourceFactory::stopEventSources();
        for (schedulers = schedulersCol_.begin(); schedulers != schedulersCol_.end(); schedulers++)
        {
            if (schedulers->second != NULL)
            {
                delete schedulers->second;
                schedulers->second = NULL;

            }
        }
        AbstractEventSourceFactory::deleteEventSources();
        LOG_TRACE_IF(logger, "Controller removed");
    }

    void RTController::initialize()
    {
        // Get scheduler (layer) information from the RTEquipment definition
        AbstractRTEquipment::RTLayerInfoCol layerInfoCol;
        AbstractRTEquipment::getInstance()->fillRTSchedulerInfo(layerInfoCol);
        // Create RT infrastructure needed by each RTDeviceClass involved in the same DeployUnit
        const std::vector<AbstractDeviceClass*>& pRTDeviceClassCol =
                        AbstractRTEquipment::getInstance()->getDeviceClassCol();
        for (std::vector<AbstractDeviceClass*>::const_iterator it = pRTDeviceClassCol.begin(); it
                        != pRTDeviceClassCol.end(); ++it)
        {
            AbstractRTDeviceClass* pRTDeviceClass = static_cast<AbstractRTDeviceClass*> (*it);
            struct stat stFileInfo;
            const std::string instanceFile = AbstractEquipment::getInstance()->getDeviceDataFileName();
            int32_t exists = stat(instanceFile.c_str(), &stFileInfo); // returns 0 if the file exists
            bool isOptional = AbstractEquipment::getInstance()->isClassOptional(pRTDeviceClass->getName());
            if (exists != 0 && !isOptional) // file does not exist and not optional => exception
            {
                LOG_ERROR_IF(logger, "Instantiation file does not exist and not optional");
                throw FesaFileException(__FILE__, __LINE__, FesaErrorOpenXMLFile.c_str(), instanceFile.c_str());

            }
            else if (exists == 0)
            {
                std::ostringstream msg;
                msg << "Configuring the RTController with the info coming from RTDeviceClass " << pRTDeviceClass->getName();
                LOG_DEBUG_IF(logger, msg.str());
                createSchedulers(layerInfoCol, pRTDeviceClass, instanceFile);
            }

        }
        //check that all scheduling units have been used. Delete the layers
        for (unsigned i = 0; i < layerInfoCol.size(); ++i)
        {
            assertSchedulingUnitsConfig(layerInfoCol[i]->schedulingUnitRefCol_);
            delete (layerInfoCol[i]);
        }
        //all NotificationID's have been created now by the RT actions ... lets save them
        NotificationIDManager::getInstance()->save();
        if (schedulersCol_.empty())//No schedulers? --> we dont need a RealTime side at all!
        {
            LOG_ERROR_IF(logger, "No scheduler found => RT part unnecessary.");
            throw FesaException(__FILE__, __LINE__, FesaErrorNothingToDoForRTSide.c_str());
        }
    }

void RTController::assertSchedulingUnitsConfig(const std::vector<RTSchedulingUnitRef> schedulingUnits)
{
    for (std::vector<RTSchedulingUnitRef>::const_iterator it = schedulingUnits.begin(); it != schedulingUnits.end();
                    ++it)
    {
        const RTSchedulingUnitRef& unit = *it;
        if (unit.found_)
        {
            continue;
        }
        bool schedFound = false;
        // check if the scheduling unit not found was not found because the device collection was empty
        for (std::vector<std::string>::iterator itr = schedulingUnitWithEmptyDevCol_.begin();
                        itr != schedulingUnitWithEmptyDevCol_.end(); ++itr)
        {
            if (unit.name_ == *itr)
            {
                schedFound = true;
                break;
            }
        }
        if (schedFound == false)
        {
            std::ostringstream msg;
            msg << "Scheduling unit " << unit.name_ << " for class " << unit.className_ << " not found.";
            LOG_ERROR_IF(logger, msg.str());
            throw FesaException(__FILE__, __LINE__, FesaErrorSchedulingUnitNotFound.c_str(), unit.name_.c_str(),
                            unit.className_.c_str());
        }
    }
}

    void RTController::createSchedulers(AbstractRTEquipment::RTLayerInfoCol& layerInfoCol,
                    AbstractRTDeviceClass* pRTDeviceClass, const std::string instanceFile)
    {
        // fill the collections
        std::vector<RTSchedulingUnitInfo*> schedulingUnitsInfoCol;
        AbstractRTDeviceClass::RTActionInfoCol actionInfoCol;
        pRTDeviceClass->fillRTSchedulingUnitsInfo(schedulingUnitsInfoCol);
        pRTDeviceClass->addDiagnosticSchedulingUnitInfo(schedulingUnitsInfoCol);
        pRTDeviceClass->fillRTActionInfo(actionInfoCol);
        pRTDeviceClass->addDiagnosticRTActionInfo(actionInfoCol);

        XMLParser xmlParser(instanceFile, false);
        boost::shared_ptr<ClassElement> pClassElement(new ClassElement(pRTDeviceClass)); // FIXME: Do we need a pointer here because of its implementation? See shared_from_this in loadLogicalEventElements
        EventsMappingLoader eventsManager(pClassElement,&xmlParser);

        // for each scheduling-unit  
        for (std::vector<RTSchedulingUnitInfo*>::iterator schedulingUnit = schedulingUnitsInfoCol.begin(); schedulingUnit
                        != schedulingUnitsInfoCol.end(); ++schedulingUnit)
        {
            if (logger.isLoggable(CMW::Log::Level::LL_TRACE))
            {
                std::ostringstream msg;
                msg << "Creating RT action config for scheduling unit " << (*schedulingUnit)->name_;
                LOG_TRACE_IF(logger, msg.str());
            }
            // create RT action configuration
            createRTActionConfig(layerInfoCol, pRTDeviceClass, *schedulingUnit, actionInfoCol, eventsManager);
            delete (*schedulingUnit); // this one is processed and was allocated by the pRTDeviceClass->fillRTSchedulingUnitsInfo => delete it
        }
        // delete resources
        for (AbstractRTDeviceClass::RTActionInfoCol::iterator actions = actionInfoCol.begin(); actions
                        != actionInfoCol.end(); actions++)
        {
            delete (actions->second);
        }
        actionInfoCol.clear();
    } // end createSchedulers

void RTController::createRTActionConfig(AbstractRTEquipment::RTLayerInfoCol& layerInfoCol, AbstractRTDeviceClass* pRTDeviceClass,
                RTSchedulingUnitInfo* schedulingUnit, AbstractRTDeviceClass::RTActionInfoCol& actionInfoCol, EventsMappingLoader& eventsManager)
{
    AbstractRTDeviceClass::RTActionInfoCol::iterator itActionInfo = actionInfoCol.find(schedulingUnit->actionName_);
    if (itActionInfo == actionInfoCol.end())
    {
        throw FesaException(__FILE__, __LINE__, FesaErrorRTActionNameNotFound.c_str(),
                        schedulingUnit->actionName_.c_str());
    }
    RTActionConfig actionConfig;
    actionConfig.actionName_ = schedulingUnit->actionName_;
    actionConfig.automaticallyNotifiedPropertiesCol_ = itActionInfo->second->automaticallyNotifiedPropertiesCol_;
    actionConfig.manuallyNotifiedPropertiesCol_ = itActionInfo->second->manuallyNotifiedPropertiesCol_;
    actionConfig.isGlobalPropertyToBeNotified_ = itActionInfo->second->isGlobalPropertyToBeNotified_;
    actionConfig.selectionCriterion_ = schedulingUnit->selectionCriterion_;
    actionConfig.className_ = schedulingUnit->className_;
    actionConfig.globalDevice_ = pRTDeviceClass->getDeviceFactory()->getGlobalDevice();
    // Looking the concrete-name for this logical event name and scheduling
    std::vector<EventElementPointer> eventConfigurationCollection = eventsManager.getEventElementCol(
                    schedulingUnit->eventName_);
    for (std::vector<EventElementPointer>::iterator itEvt = eventConfigurationCollection.begin();
                    itEvt != eventConfigurationCollection.end(); ++itEvt)
    {
        if (logger.isLoggable(CMW::Log::Level::LL_TRACE))
        {
            std::ostringstream msg;
            msg << "Processing event " << (*itEvt)->getConcreteName();
            LOG_TRACE_IF(logger, msg.str());
        }
        actionConfig.eventSourceName_ = (*itEvt)->getEventSourceName();
        const std::set<AbstractDevice*>& devCol = (*itEvt)->getDeviceCollection();
        actionConfig.eventName_ = (*itEvt)->getConcreteName(); // Concrete event name
        // split devices depending on selection-criterion
        std::vector<AbstractDevice *> unsortedDevicelCol;
        std::copy(devCol.begin(), devCol.end(), std::back_inserter(unsortedDevicelCol)); // TODO Why's that if it's just for the SortingInterp, can we change it?
        SortingInterpreter sortingInterpreter;
        std::set<HomogeneousDevCol *> collectionOfDevCol = sortingInterpreter.interpret(
                        actionConfig.selectionCriterion_, actionConfig.className_, &(unsortedDevicelCol));
        // register scheduling-unit with empty device collections
        if (collectionOfDevCol.empty())
        {
            schedulingUnitWithEmptyDevCol_.push_back(schedulingUnit->name_);
            std::ostringstream msg;
            msg << "Scheduling unit " << schedulingUnit->name_ << " doesn't have any device => ignore it.";
            LOG_TRACE_IF(logger, msg.str());
        }
        else
        {
            for (std::set<HomogeneousDevCol*>::iterator itDevCol = collectionOfDevCol.begin();
                            itDevCol != collectionOfDevCol.end(); ++itDevCol)
            {
                if (logger.isLoggable(CMW::Log::Level::LL_TRACE))
                {
                    std::ostringstream msg;
                    msg << "Processing device collection of " << (*itDevCol)->getSize() << " devices";
                    LOG_TRACE_IF(logger, msg.str());
                }
                std::set<AbstractDevice*>* deviceCollection = (*itDevCol)->getDevCol();
                RTScheduler* scheduler = createAction(layerInfoCol, pRTDeviceClass, schedulingUnit, actionConfig,
                                deviceCollection);
                if (scheduler != NULL)
                { // create a new event-source (if needed)
                    createEventSource(pRTDeviceClass->getName(), *itEvt, scheduler,
                                    pRTDeviceClass->getEventSourceFactory());
                }
            }
        }
    }
}

RTScheduler* RTController::createAction(AbstractRTEquipment::RTLayerInfoCol& layerInfoCol, AbstractRTDeviceClass* pRTDeviceClass,
                RTSchedulingUnitInfo* schedulingUnit, RTActionConfig& actionConfig, std::set<AbstractDevice*>* deviceCollection)
{
    actionConfig.deviceCol_.clear();
    std::copy(deviceCollection->begin(), deviceCollection->end(), std::back_inserter(actionConfig.deviceCol_)); // TODO Why's that if it's just for the SortingInterp, can we change it?
    RTSchedulingUnitRef schedulingUnitRef;
    RTLayerInfo* layerInfo;
    uint32_t k;
    bool schedulingFound = false;
    for (k = 0; k < layerInfoCol.size(); ++k)
    {
        for (uint32_t j = 0; j < layerInfoCol[k]->schedulingUnitRefCol_.size(); ++j)
        {
            if (((((layerInfoCol[k]->schedulingUnitRefCol_[j].name_) == schedulingUnit->name_))
                            && ((schedulingUnit)->name_ == DIAGNOSTIC_LAYER))
                            || (((layerInfoCol[k]->schedulingUnitRefCol_[j].name_) == schedulingUnit->name_)
                                            && (layerInfoCol[k]->schedulingUnitRefCol_[j].className_
                                                            == schedulingUnit->className_)))
            {
                layerInfo = layerInfoCol[k];
                schedulingUnitRef = layerInfoCol[k]->schedulingUnitRefCol_[j];
                layerInfoCol[k]->schedulingUnitRefCol_[j].found_ = true;
                schedulingFound = true;
                break;
            }
        }
        if (schedulingFound == true)
        {
            break;
        }
    }
    if (k == layerInfoCol.size())
    {
        std::ostringstream msg;
        msg << "scheduling unit" << schedulingUnit->className_ << ":" << schedulingUnit->name_.c_str()
                        << " is not used in the current deploy unit";
        LOG_WARNING_IF(logger, msg.str());
        return (RTScheduler*)NULL;
    }
    // compute scheduler name
    std::stringstream layerName;
    layerName << layerInfo->layerName_;
    // relying on selection criterion if required
    if (schedulingUnitRef.perDeviceCollection_)
    {
        layerName << actionConfig.eventName_;
    }
    RTScheduler* currentScheduler = findScheduler(layerName.str(), layerInfo->queueSize_);
    // create the RT action
    AbstractRTActionFactory* pRTActionFactory = pRTDeviceClass->getRTActionFactory();
    if (pRTActionFactory == NULL)
    {
        throw FesaException(__FILE__, __LINE__, FesaErrorRTFactoryNotFound.c_str(), schedulingUnit->className_.c_str());
    }
    AbstractRTAction* action = pRTActionFactory->createRTAction(actionConfig.actionName_, actionConfig);
    if (action == NULL)
    {
        throw FesaException(__FILE__, __LINE__, FesaErrorRTActionNameNotFound.c_str(),
                        actionConfig.actionName_.c_str());
    }
    // at this point currentScheduler contains the scheduler for this action
    currentScheduler->addRTAction(actionConfig.eventName_, action);
    if (logger.isLoggable(CMW::Log::Level::LL_TRACE))
    {
        std::ostringstream msg;
        msg << "Action " << action->getName() << " created and added to scheduler " << currentScheduler->getName();
        LOG_TRACE_IF(logger, msg.str());
    }
    return currentScheduler;
}

RTScheduler* RTController::findScheduler(const std::string& schedulerName, uint32_t queueSize)
{
    RTScheduler* scheduler;
    std::map<const std::string, RTScheduler*>::iterator schedIter = schedulersCol_.find(schedulerName);
    if (schedIter != schedulersCol_.end())
    {
        scheduler = (*schedIter).second;
    }
    else // create a new scheduler
    {
        // Create a log with the appropriate className in oder to follow the
        // custom topics which are class dependant
        scheduler = new RTScheduler(schedulerName, queueSize);
        schedulersCol_.insert(std::make_pair(schedulerName, scheduler));
    }
    return scheduler;
}

    void RTController::start()
    {

        //start the schedulers
        std::map<std::string, RTScheduler*>::iterator rtTmpScheduler;

        for (rtTmpScheduler = schedulersCol_.begin(); rtTmpScheduler != schedulersCol_.end(); rtTmpScheduler++)
        {
            // run all the schedulers but the diagnostic one
            const std::string name = rtTmpScheduler->second->getName();
            if (name != "diagnosticLayer")
            {
                rtTmpScheduler->second->start(false, name);
            }
        }

        // start the all event sources (Fwk and Custom ones)
        AbstractEventSourceFactory::startEventSources();

    rtTmpScheduler = schedulersCol_.find("diagnosticLayer");
        // run diagnostic layer in a separate thread without blocking call
        if (rtTmpScheduler != schedulersCol_.end())
        {
        const std::string name = rtTmpScheduler->second->getName();
        rtTmpScheduler->second->start(false, name);
        }
        else
        {
            //No Diagnostic Layer? --> Something is wrong on the RT side, throw an exception!
            throw FesaException(__FILE__, __LINE__, FesaErrorNoDiagnosticLayerFound.c_str());
        }
    }// end run


    void RTController::createEventSource(const std::string& className, boost::shared_ptr<fesa::EventElement>& eventElement,
                    RTScheduler* currentScheduler, AbstractEventSourceFactory* eventSourceFactory)
    {
        AbstractEventSource* source;
        if(eventSourceFactory->eventSourceExists(className,eventElement->getEventSourceName()))
        {
                source = eventSourceFactory->getEventSource(className,eventElement->getEventSourceName());
        }
        else
        {
                source = eventSourceFactory->createEventSource(className,eventElement->getEventSourceName(),eventElement->getEventSourceType());
        }

        source->addRTScheduler(eventElement, currentScheduler);
    }

    void RTController::printConfigAll(FesaStream* fesaStream)
    {

        AbstractEventSourceFactory::printConfigAll(fesaStream);

        *fesaStream << "\t\t<scheduling>" << std::endl;
        std::map<const std::string, RTScheduler*>::iterator itr = schedulersCol_.begin();
        std::map<const std::string, RTScheduler*>::iterator end = schedulersCol_.end();
        for (; itr != end; ++itr)
        {
            itr->second->printConfig(fesaStream);
        }
        *fesaStream << "\t\t</scheduling>" << std::endl;
    }

    void RTController::printStateAll(FesaStream* fesaStream, double elapsedTime)
    {

        AbstractEventSourceFactory::printStateAll(fesaStream, elapsedTime);

        *fesaStream << "\t\t<scheduling>" << std::endl;
        std::map<const std::string, RTScheduler*>::iterator itr = schedulersCol_.begin();
        std::map<const std::string, RTScheduler*>::iterator end = schedulersCol_.end();
        for (; itr != end; ++itr)
        {
            itr->second->printState(fesaStream, elapsedTime);
        }
        *fesaStream << "\t\t</scheduling>" << std::endl;

    }

}// end class