AbstractEventSource.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

#include <fesa-core/RealTime/AbstractEventSource.h>

#include <fesa-core/RealTime/AbstractRTEquipment.h>
#include <fesa-core/RealTime/RTEvent.h>
#include <fesa-core/RealTime/RTScheduler.h>
#include <fesa-core/Diagnostic/FesaStream.h>
#include <fesa-core/Core/AbstractEquipment.h>
#include <fesa-core/Core/ThreadPriorityConfiguration.h>
#include <fesa-core/Utilities/XMLParser.h>
#include <fesa-core/Utilities/Lock.h>
#include <fesa-core/Utilities/ParserElements/EventElement.h>

#include <cmw-log/Logger.h>


namespace
{

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

} // namespace


namespace fesa
{

AbstractEventSource::AbstractEventSource(const std::string eventSourceName, EventSourceType type) :
    sourceIsEnabled_(true),//event-source is enabled by default
    eventSourceName_(eventSourceName),
    type_(type)
{
    std::ostringstream message;
    message << "createEventSource, name: " << eventSourceName_;
    LOG_TRACE_IF(logger, message.str());
    Thread::registerThreadIdName(getID(), eventSourceName);

    //Number of event fired by this event-source.
    firedEventCount_.last_ = 0;
    firedEventCount_.current_ = 0;
    firedEventCount_.max_ = 0;
    {
        AbstractEquipment* eqp = AbstractEquipment::getInstance();
        std::string instanceFile = eqp->getDeviceDataFileName();
        XMLParser xmlParser(instanceFile, false);
        ThreadPriorityConfigurationFromFile conf(xmlParser,eqp->getProcessConfiguration());
        int32_t prio =  conf.getPrioEventSource(eventSourceName_);
        setPriority(prio);
    }
}

AbstractEventSource::~AbstractEventSource()
{
    std::ostringstream message;
    message << "Removing event source: " << eventSourceName_;
    LOG_TRACE_IF(logger, message.str());
}

void AbstractEventSource::run()
{
    isRunning_ = true;
    AbstractRTEquipment* rtEquipment = AbstractRTEquipment::getInstance();
    RTEvent* pEvt;
    while (isRunning_)
    {
        try
        {
            waitTillSourceEnabled();
            pEvt = wait();
            if (pEvt != NULL)
            {
                boost::shared_ptr<RTEvent> sh_pEvt(pEvt);
                postEventToSchedulers(sh_pEvt);
                (firedEventCount_.current_)++;
            }
        }
        catch (FesaException& ex)
        {
            std::ostringstream errorStrStream;
            errorStrStream << ex.getMessage();
            LOG_ERROR_IF(logger, errorStrStream.str());
            rtEquipment->handleEventSourceError(this,ex);
            continue;//start at beginning of while loop
        }
        catch (...)
        {
            LOG_ERROR_IF(logger, "Unknown exception caught");
            FesaException ex (__FILE__, __LINE__, "Unknown exception caught");
            rtEquipment->handleEventSourceError(this, ex);
            continue; //start at beginning of while loop
        }
    }// end while loop

    std::ostringstream message;
    message << "Terminating " << eventSourceName_;
    LOG_TRACE_IF(logger, message.str());
    hasFinished_ = true;

} // end run

void AbstractEventSource::postEventToSchedulers(boost::shared_ptr<RTEvent>& event)
{
    ConcreteEventMap::const_iterator eventIter = eventMap_.find(event->getName());
    if (eventIter == eventMap_.end())//not found
    {
        std::ostringstream message;
        message << "No scheduler available to scheduler the concrete-event:" << event->getName() << " in event-source: " << getName() ;
        LOG_TRACE_IF(logger, message.str());
        return;
    }

    std::vector<SubscribedScheduler>::const_iterator sched_iter;
    for (sched_iter = eventIter->second.begin(); sched_iter != eventIter->second.end(); ++sched_iter)
    {
        if(sched_iter->isEnabled_)
        {
            if(logger.isLoggable(CMW::Log::Level::LL_TRACE))
            {
                std::ostringstream message;
                message << "Posting logical-event: " << eventIter->first << " to scheduler: " << sched_iter->scheduler_->getName();
                LOG_TRACE_IF(logger, message.str());
            }
            sched_iter->scheduler_->post(event);
        }
        else
        {
            std::ostringstream message;
            message << "The logical event:" << eventIter->first << " is disabled and will not be fired.";
            LOG_TRACE_IF(logger, message.str());
        }
    }
}

void AbstractEventSource::addRTScheduler(boost::shared_ptr<fesa::EventElement>& eventElement, RTScheduler* scheduler)
{
    SubscribedScheduler schedulingInfo;
    schedulingInfo.scheduler_ = scheduler;
    schedulingInfo.logicalEventName_ = eventElement->getLogicalName();
    schedulingInfo.isEnabled_ = true;

    ConcreteEventMap::iterator eventIter = eventMap_.find(eventElement->getConcreteName());
    if (eventIter == eventMap_.end())//not found
    {
        std::vector< SubscribedScheduler > schedulers;
        schedulers.push_back(schedulingInfo);
        eventMap_.insert(make_pair(eventElement->getConcreteName(),schedulers));

        connect(eventElement);
    }
    else
    {
        std::vector< SubscribedScheduler >::iterator schedIter;
        for (schedIter= eventIter->second.begin(); schedIter != eventIter->second.end(); ++schedIter)
        {
            if (schedIter->scheduler_ == scheduler && schedIter->logicalEventName_ == eventElement->getLogicalName() )
            {
                std::ostringstream message;
                message << "RTScheduler: " << scheduler->getName() << " already registered for event: " << eventElement->getLogicalName();
                LOG_TRACE_IF(logger, message.str());
                return;
            }
        }
        eventIter->second.push_back(schedulingInfo);
    }
    std::ostringstream message;
    message << "Added scheduler: " << scheduler->getName() << " for logical-event: " << eventElement->getLogicalName();
    LOG_TRACE_IF(logger, message.str());
}

void AbstractEventSource::waitTillSourceEnabled()
{
    while(!sourceIsEnabled_)
    {
        Lock lock(mutex_waitIfDisabled_);
        waitIfDisabled_.wait(mutex_waitIfDisabled_);
    }
}

void AbstractEventSource::printConfig(FesaStream* configStream) const
{
    *configStream << "\t\t\t<event-source name=\"" << eventSourceName_ << "\">" << std::endl;
    ConcreteEventMap::const_iterator itr;
    for ( itr = eventMap_.begin(); itr != eventMap_.end(); ++itr)
    {
        *configStream << "\t\t\t\t<event concreteName=\"" << itr->first << "\"/>" << std::endl;
    }
    *configStream << "\t\t\t</event-source>" << std::endl;
}

void AbstractEventSource::printState(FesaStream* fesaStream, double elapsedTime)
{
    float freq_firedEventCount = (float) ((firedEventCount_.current_ - firedEventCount_.last_) / elapsedTime);
    *fesaStream << "\n    event-source thread-name = \"" << eventSourceName_ << "\"" << std::endl;
    *fesaStream << "        thread-id:\t" << getID() << std::endl;
    *fesaStream << "        priority:\t\t" << getEvtSrcPriority() << std::endl;
    *fesaStream << "        fired-event-count (last/current: freq):\t\t" << firedEventCount_.last_ << "/" << firedEventCount_.current_ << ": " << freq_firedEventCount << " Hz" << std::endl;
    firedEventCount_.last_ = firedEventCount_.current_;

}
const std::string& AbstractEventSource::getName() const
{
    return eventSourceName_;
}

EventSourceType AbstractEventSource::getType() const
{
    return type_;
}

uint32_t AbstractEventSource::getEvtSrcPriority() const
{
    return (getPriority());
}

uint32_t AbstractEventSource::getEvtSrcID() const
{
    return ((uint32_t) getID());
}

void AbstractEventSource::enable()
{
    sourceIsEnabled_ = true;
    {
        Lock lock(mutex_waitIfDisabled_);
        waitIfDisabled_.signal();
    }
    std::ostringstream message;
    message << "Events source: " << eventSourceName_ << " got enabled.";
    LOG_TRACE_IF(logger, message.str());
}

void AbstractEventSource::disable()
{
    sourceIsEnabled_ = false;
    std::ostringstream message;
    message << "Events source: " << eventSourceName_ << " got disabled.";
    LOG_TRACE_IF(logger, message.str());
}

void AbstractEventSource::enableEvent(const std::string& logicalEventName)
{
    setLogicalEventEnableState(true, logicalEventName);
    std::ostringstream message;
    message << "Event: " << logicalEventName << " of events-source: " << eventSourceName_ << " got enabled.";
    LOG_TRACE_IF(logger, message.str());
}

void AbstractEventSource::disableEvent(const std::string& logicalEventName)
{
    setLogicalEventEnableState(false, logicalEventName);
    std::ostringstream message;
    message << "Event: " << logicalEventName << " of events-source: " << eventSourceName_ << " got disabled.";
    LOG_TRACE_IF(logger, message.str());
}

bool AbstractEventSource::isEventEnabled(const std::string& logicalEventName) const
{
    ConcreteEventMap::const_iterator eventIter;
    std::vector< SubscribedScheduler >::const_iterator schedulerIter;
    for(eventIter = eventMap_.begin();eventIter!=eventMap_.end();++eventIter)
    {
        for(schedulerIter = eventIter->second.begin();schedulerIter!=eventIter->second.end();++schedulerIter)
        {
            if(schedulerIter->logicalEventName_ == logicalEventName)
            {
                //all logical events of that name share the same enable-state
                //so we can directly return the first state we can find
                return schedulerIter->isEnabled_;
            }
        }
    }
    //if not found
    throw FesaException(__FILE__, __LINE__, FesaErrorEventNotFound.c_str(), logicalEventName.c_str());
}

void AbstractEventSource::setLogicalEventEnableState(bool newState,const std::string& logicalEventName)
{
    bool found = false;
    ConcreteEventMap::iterator eventIter;
    std::vector< SubscribedScheduler >::iterator schedulerIter;
    for(eventIter = eventMap_.begin();eventIter!=eventMap_.end();++eventIter)
    {
        for(schedulerIter = eventIter->second.begin();schedulerIter!=eventIter->second.end();++schedulerIter)
        {
            if(schedulerIter->logicalEventName_ == logicalEventName)
            {
                schedulerIter->isEnabled_ = newState;
                found = true;
            }
        }
    }
    if( !found )
    {
        throw FesaException(__FILE__, __LINE__, FesaErrorEventNotFound.c_str(), logicalEventName.c_str());
    }
}

} // fesa