RTScheduler.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

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

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

#include <cmw-log/Logger.h>


namespace
{

CMW::Log::Logger& logger = CMW::Log::LoggerFactory::getLogger("FESA.FWK.fesa-core.RealTime.RTScheduler");
const std::string diagTopic = fesa::DiagnosticUtils::eventTrackingStr;

} // namespace


namespace fesa
{

RTScheduler::RTScheduler(const std::string& name, int32_t eventQueueSize) :
    name_(name),
    queueSize_(eventQueueSize),
    lostEventCount_(0)
{
    setDetachState( PTHREAD_CREATE_DETACHED);
    {
        AbstractEquipment* eqp = AbstractEquipment::getInstance();
        std::string instanceFile = eqp->getDeviceDataFileName();
        XMLParser xmlParser(instanceFile, false);
        ThreadPriorityConfigurationFromFile conf(xmlParser, eqp->getProcessConfiguration());
        int32_t prio =  conf.getPrioRTScheduler(name_);
        setPriority(prio);
    }

    postEventCount_.last_ = 0;
    postEventCount_.current_ = 0;
    postEventCount_.max_ = 0;
    scheduledEventCount_.last_ = 0;
    scheduledEventCount_.current_ = 0;
    scheduledEventCount_.max_ = 0;
    accumulatedEventCount_.last_ = 0;
    accumulatedEventCount_.current_ = 0;
    accumulatedEventCount_.max_ = 0;

    std::ostringstream message;
    message << " Creation of RTScheduler: " << name;
    LOG_TRACE_IF(logger, message.str());
}

RTScheduler::~RTScheduler()
{
    std::map<const std::string, std::vector<AbstractRTAction*> >::iterator schedullingUnit;
    std::vector<AbstractRTAction*>::iterator actions;

    for (schedullingUnit = schedulingMap_.begin(); schedullingUnit != schedulingMap_.end(); schedullingUnit++)
        for (actions = schedullingUnit->second.begin(); actions != schedullingUnit->second.end(); actions++)
            if (*actions != NULL)
            {
                delete *actions;
                *actions = NULL;
            }
}

void RTScheduler::post(boost::shared_ptr<RTEvent>& pEv)
{
    (postEventCount_.current_)++;
    //Save maximum number of event accumulation
    accumulatedEventCount_.current_ = static_cast<uint32_t>(eventQueue_.size());
    if (accumulatedEventCount_.current_ > accumulatedEventCount_.max_)
    {
        accumulatedEventCount_.max_ = accumulatedEventCount_.current_;
    }
    Lock lock(mutexQueue_);
    if ((queueSize_ < 0) || // no limit defined
        (eventQueue_.size() < (uint32_t) queueSize_))
    { // check against the limit
        // No event jam, push-back the event
        eventQueue_.push_back(pEv);
        // Lower alarm Field in case it was raised before
        //pGlobalDevice->rtEventLost.lower();
        // Wake up the the scheduler and returns
        conditionalVariableQueue_.signal();
        if (logger.isLoggable(CMW::Log::Level::LL_TRACE))
        {
            std::ostringstream message;
            message << "Signaling " << name_;
            LOG_TRACE_IF(logger, message.str());
        }
    }
    else
    {
        std::ostringstream errorStrStream;
        errorStrStream << "Could not post event: " << pEv->getName() << "; RTScheduler queue is completely full." << std::endl;
        LOG_ERROR_IF(logger, errorStrStream.str());
        FesaException exception(__FILE__, __LINE__, errorStrStream.str());
        AbstractRTEquipment::getInstance()->handleSchedulerError(this,exception);
    }
}

void RTScheduler::run()
{
    if (logger.isLoggable(CMW::Log::Level::LL_DEBUG))
    {
        std::ostringstream message;
        message << "Running scheduler " << name_;
        LOG_DEBUG_IF(logger, message.str());
    }
    while (isRunning_)
    {
        // a blocking-call returning one-by-one the events hold in the Scheduler's event-queue
        boost::shared_ptr<RTEvent> pShEvt;
        {
            Lock lock(mutexQueue_);
            if (eventQueue_.size() == 0)
            {
                conditionalVariableQueue_.wait(mutexQueue_);
                if (logger.isLoggable(CMW::Log::Level::LL_TRACE))
                {
                    std::ostringstream message;
                    message << "Post received for " << name_;
                    LOG_TRACE_IF(logger, message.str());
                }
            }
            // dequeue events
            pShEvt = eventQueue_.front();
            // delete the entry in the queue (shared_ptr reference-counter decremented by one)
            eventQueue_.pop_front();
        }
        // retrieve from the shared_pointer the raw object.
        RTEvent* pEvent = pShEvt.get();
        if (isRunning_ && pEvent != NULL)
        {
            // Schedule the Event
            schedule(pEvent);
        }
        else
        {
            continue;
        }
    }
    hasFinished_ = true;
} // end run

void RTScheduler::logDiagnostics(RTEvent*& pEvent)
{
    const boost::shared_ptr<Diagnostics>& diagnostics = AbstractEquipment::getInstance()->getDiagnostics();
    DiagnosticUtils::DiagnosticMessage diagMsg;
    diagMsg.side = DiagnosticUtils::framework;
    diagMsg.source = DiagnosticUtils::rt;
    std::ostringstream traceStrStream;
    traceStrStream << "RTEvent '" << pEvent->getName() << "'";
    diagMsg.msg = traceStrStream.str();
    diagnostics->log(diagTopic, diagMsg);
}

void RTScheduler::schedule(RTEvent* pEvent)
{
    logDiagnostics(pEvent);
    std::map<const std::string, std::vector<AbstractRTAction*> >::iterator eventIterator = schedulingMap_.find(
                    pEvent->getName());
    if (eventIterator != schedulingMap_.end())
    {
        (scheduledEventCount_.current_)++;
        std::vector<AbstractRTAction*>::iterator actionsIterator;
        for (actionsIterator = eventIterator->second.begin(); actionsIterator != eventIterator->second.end();
                        actionsIterator++)
        {
            try
            {
                if (logger.isLoggable(CMW::Log::Level::LL_TRACE))
                {
                    std::ostringstream message;
                    message << "Executing action " << (*actionsIterator)->getName();
                    LOG_TRACE_IF(logger, message.str());
                }
                (*actionsIterator)->executeAction(pEvent);
            }
            catch (FesaException& ex)
            {
                std::ostringstream errorStrStream;
                errorStrStream << "RTScheduler::schedule() caught FesaException during " << name_
                                << " RTAction execution: " << ex.getMessage();
                LOG_ERROR_IF(logger, errorStrStream.str());
            }
            catch (std::exception& ex)
            {
                std::ostringstream errorStrStream;
                errorStrStream << "RTScheduler::schedule() caught std::exception during " << name_
                                << " RTAction execution: " << ex.what();
                LOG_ERROR_IF(logger, errorStrStream.str());
            }

            catch (...)
            {
                std::ostringstream errorStrStream;
                errorStrStream << "RTScheduler::schedule() caught unknown exception during " << name_
                                << " RTAction execution";
                LOG_ERROR_IF(logger, errorStrStream.str());
            }
        }
    }
    else
    {
        std::ostringstream errorStrStream;
        errorStrStream << "RTScheduler::schedule()  " << name_ << " RTAction execution: "
                        << "scheduler not found for event: " << pEvent->getName();
        LOG_ERROR_IF(logger, errorStrStream.str());
    }
}

void RTScheduler::printConfig(FesaStream* configStream)
{
    std::map<std::string, std::vector<AbstractRTAction*> >::iterator itr = schedulingMap_.begin();
    std::map<std::string, std::vector<AbstractRTAction*> >::iterator end = schedulingMap_.end();
    *configStream << "\t\t\t<concurrent-layer name=\"" << name_ << "\" priority=\"" << getPriority() << "\">"
                    << std::endl;
    for (; itr != end; ++itr)
    {
        *configStream << "\t\t\t\t<event name=\"" << itr->first << "\">" << std::endl;
        std::vector<AbstractRTAction*>::iterator itrAct = itr->second.begin();
        std::vector<AbstractRTAction*>::iterator endAct = itr->second.end();
        for (; itrAct != endAct; ++itrAct)
        {
            (*itrAct)->printConfig(configStream);
        }
        *configStream << "\t\t\t\t</event>" << std::endl;
    }

    *configStream << "\t\t\t</concurrent-layer>" << std::endl;
}

void RTScheduler::printState(FesaStream* fesaStream, double elapsedTime)
{
    float freq_postEventCount;
    float freq_scheduledEventCount;
    *fesaStream << "\nSCHEDULING --------------------------------------------------------------" << std::endl;
    freq_postEventCount = (float) (((postEventCount_.current_ - postEventCount_.last_) / elapsedTime));
    freq_scheduledEventCount = (float) (((scheduledEventCount_.current_ - scheduledEventCount_.last_) / elapsedTime));
    *fesaStream << "\n    concurrent-layer thread-name = \"" << name_ << "\"" << std::endl;
    *fesaStream << "        thread-id:\t" << getID() << std::endl;
    //*fesaStream << "        policy:\t\t" << policyStr << endl;
    *fesaStream << "        priority:\t\t" << getPriority() << std::endl;
    *fesaStream << "        event-queue-size:\t" << queueSize_ << std::endl;
    *fesaStream << "        post-event-count (last/current: freq):\t\t" << postEventCount_.last_ << "/"
                << postEventCount_.current_ << ": " << freq_postEventCount << " Hz" << std::endl;
    *fesaStream << "        scheduled-event-count (last/current: freq):\t" << scheduledEventCount_.last_ << "/"
                << scheduledEventCount_.current_ << ": " << freq_scheduledEventCount << " Hz" << std::endl;
    *fesaStream << "        current-accumulated-event-count:\t" << accumulatedEventCount_.current_ << std::endl;
    *fesaStream << "        max-accumulated-event-count:\t" << accumulatedEventCount_.max_ << std::endl;
    *fesaStream << "        lost-event-count:\t\t" << lostEventCount_ << std::endl;
    postEventCount_.last_ = postEventCount_.current_;
    scheduledEventCount_.last_ = scheduledEventCount_.current_;

    std::map<std::string, std::vector<AbstractRTAction*> >::iterator itrEvt = schedulingMap_.begin();
    std::map<std::string, std::vector<AbstractRTAction*> >::iterator endEvt = schedulingMap_.end();
    for (; itrEvt != endEvt; ++itrEvt)
    {
        *fesaStream << "\n        event name = \"" << itrEvt->first << "\"" << std::endl;
        std::vector<AbstractRTAction*>::iterator itrAct = itrEvt->second.begin();
        std::vector<AbstractRTAction*>::iterator endAct = itrEvt->second.end();
        for (; itrAct != endAct; ++itrAct)
            (*itrAct)->printState(fesaStream, elapsedTime);

    }

}

void RTScheduler::cancel()
{
    isRunning_ = false;
    int32_t attempts = MAX_ATTEMPTS_STOP_THREAD;

    //wait for the thread to stop
    while (!hasFinished_ && attempts > 0)
    {
        attempts--;
        conditionalVariableQueue_.signal();
        sleep(1);

    }
    if (attempts == 0)
    {
        stop();
        std::ostringstream errorStrStream;
        errorStrStream << "During shut down procedure RTScheduler: " << name_ << " was forced to exit ";
        LOG_ERROR_IF(logger, errorStrStream.str());
    }
}

void RTScheduler::addRTAction(const std::string& eventName, AbstractRTAction* rtAction)
{
    std::map<const std::string, std::vector<AbstractRTAction*> >::iterator rtActionList = schedulingMap_.find(
        eventName);

    if (rtActionList != schedulingMap_.end())
    {
        rtActionList->second.push_back(rtAction);
    }
    else
    {
        std::vector<AbstractRTAction*> actions;
        actions.push_back(rtAction);
        schedulingMap_.insert(std::make_pair(eventName, actions));
    }
}

} // fesa