Horizon » Horizon Simulation Framework

//=========================================================================

//  CSCHEDULER.CC - part of

//

//                  OMNeT++/OMNEST

//           Discrete System Simulation in C++

//

//  Author: Andras Varga, 2003

//

//=========================================================================

/*--------------------------------------------------------------*

  Copyright (C) 2003-2008 Andras Varga

  Copyright (C) 2006-2008 OpenSim Ltd.

  Monash University, Dept. of Electrical and Computer Systems Eng.

  Melbourne, Australia

  This file is distributed WITHOUT ANY WARRANTY. See the file

  `license' for details on this and other legal matters.

*--------------------------------------------------------------*/

#include "cscheduler.h"

#include "cmessage.h"

#include "csimulation.h"

#include "cmessageheap.h"

#include "globals.h"

#include "cenvir.h"

#include "cconfiguration.h"

#include "cconfigoption.h"

#include "platmisc.h" // usleep

#include "cbarriermessage.h"

USING_NAMESPACE

Register_GlobalConfigOption(CFGID_REALTIMESCHEDULER_SCALING, "realtimescheduler-scaling", CFG_DOUBLE, NULL, "When cRealTimeScheduler is selected as scheduler class: ratio of simulation time to real time. For example, scaling=2 will cause simulation time to progress twice as fast as runtime.");

cScheduler::cScheduler()

{

    sim = NULL;

}

cScheduler::~cScheduler()

{

}

void cScheduler::setSimulation(cSimulation *_sim)

{

    sim = _sim;

}

//-----

Register_Class(cSequentialScheduler);

cMessage *cSequentialScheduler::getNextEvent()

{

#ifdef NOBARRIER

    // Do we have to wait for a barrier?

    if (sim->threadPool) sim->threadPool->waitAtBarrier(&(sim->msgQueue));

    //

    // If we retrieve a valid msg from the queue, we return it:

    //

    cMessage *msg = sim->msgQueue.removeFirst();

    if (msg) {

        /*

         * Set Duration

         */

        cSimpleModule* mod = (cSimpleModule*) sim->getModule(msg->getArrivalModuleId());

        if (mod->isAsyncModule()) {

            msg->setEventDuration(((cAsyncModule*) mod)->getProcessingDelay(msg));

        }

        /*

         * return message

         */

        return msg;

    }

    //

    // if there is no event left and we don't use the threadpool, end the sim

    //

    if (!sim->threadPool)

        throw cTerminationException(eENDEDOK);

    //

    // If we did not get a valid msg from the queue, but there are still

    // barrier messages left, we wait:

    //

    while (sim->msgQueue.empty() && !sim->threadPool->barrierEmpty())

    {

        __asm__ ("pause");

    }

    msg = sim->msgQueue.removeFirst();

    //

    // If there is a msg now, we return it:

    //

    if (msg) {

        /*

         * Set Duration

         */

        cSimpleModule* mod = (cSimpleModule*) sim->getModule(msg->getArrivalModuleId());

        if (mod->isAsyncModule()) {

            msg->setEventDuration(((cAsyncModule*) mod)->getProcessingDelay(msg));

        }

        /*

         * return message

         */

        return msg;

    }

    //

    // If there is still no message in the queue, there are

    // also no barriers left (else we would have waited), and we quit:

    //

    else

    throw cTerminationException(eENDEDOK);

#else

    cMessage *msg = NULL;

    cBarrierMessage* barrier = NULL;

    while (!msg)

    {

        msg = sim->msgQueue.removeFirst();

        if (!msg)

            throw cTerminationException(eENDEDOK);

        /*

         * If we have a Barriermsg, we wait

         */

        barrier = dynamic_cast<cBarrierMessage*> (msg);

        if (barrier != NULL)

        {

            // wait for the task to complete

            barrier->wait();

            delete barrier;

            msg = NULL;

        }

    }

    cSimpleModule* mod = (cSimpleModule*) sim->getModule(msg->getArrivalModuleId());

    if (mod->isAsyncModule())

    {

        cAsyncModule* aMod = (cAsyncModule*) mod;

        simtime_t now = msg->getArrivalTime();

        simtime_t duration = aMod->getProcessingDelay(msg);

        if (aMod->mayParallelize(msg, duration))

        {

            msg->setEventDuration(duration);

            // create a new barrier and schedule it

            cBarrierMessage* barrier = new cBarrierMessage();

            barrier->setArrival(aMod, -1, now + duration);

            msg->setBarrier(barrier);

            // insert user supplied message in task queue.

            sim->msgQueue.insert(barrier);

        }

    }

    return msg;

#endif

}

//-----

Register_Class(cEEFScheduler)

;

cEEFScheduler::cEEFScheduler()

{

}

cEEFScheduler::~cEEFScheduler()

{

}

cMessage *cEEFScheduler::getNextEvent()

{

    //TODO: check if barriermessage is still vaild before waiting (may speed up things, as IES may not run empty)

    cMessage *msg = NULL;

    cSimpleModule* mod = NULL;

    cAsyncModule* aMod = NULL;

#ifdef NOBARRIER

    while (!(sim->msgQueue.empty()))

       {

        msg = sim->msgQueue.removeFirst();

        mod = (cSimpleModule*) sim->getModule(msg->getArrivalModuleId());

        if (!(mod->isAsyncModule())) {

            break;

        }

        aMod = (cAsyncModule*) mod;

        simtime_t duration = aMod->getProcessingDelay(msg);

        msg->setEventDuration(duration);

        bool mayPar = aMod->mayParallelize(msg, duration);

        if (!mayPar) {

            break;

        }

        /*

         * If there is an active barrier in the threadpool before msg, break

         */

        if (sim->threadPool->isBeforeBarrier(msg)) {

            break;

        }

        /*

         * If an event in the IES would cause a barrier before msg, break

         */

        if (!independentEventsHeap.empty() && msg->getArrivalTime() >= independentEventsHeap.peekFirst()->getTend()) {

            break;

        }

        printf(

                "adding to IEH: %s, tstart=%f, tend= %f, now First in IEH: ",

                ((cSimpleModule*) sim->getModule(msg->getArrivalModuleId()))->getName(),SIMTIME_DBL(msg->getArrivalTime()),

                SIMTIME_DBL(msg->getTend()));

        independentEventsHeap.insert(msg);

        printf(

                "%s, length=%i\n",

                ((cSimpleModule*) sim->getModule(

                        independentEventsHeap.peekFirst()->getArrivalModuleId()))->getName(),

                independentEventsHeap.length());

        msg = NULL;

    }

    if (msg) sim->msgQueue.insert(msg);

    if(independentEventsHeap.empty()) {

        if (!sim->msgQueue.empty()) {

            // Do we have to wait for a barrier?

            if (sim->threadPool) sim->threadPool->waitAtBarrier(&(sim->msgQueue));

            return sim->msgQueue.removeFirst();

        }

    } else {

        return independentEventsHeap.getFirst();

    }

    // At this point, both IES and FES are empty

    //

    // if there is no event left and we don't use the threadpool, end the sim

    //

    if (!sim->threadPool)

    throw cTerminationException(eENDEDOK);

    //

    // If we did not get a valid msg from the queue, but there are still

    // barrier messages left, we wait:

    //

    while (sim->msgQueue.empty() && !sim->threadPool->barrierEmpty())

    {

        __asm__ ("pause");

    }

    msg = sim->msgQueue.removeFirst();

    //

    // If there is a msg now, we return it:

    //

    if (msg)

    return msg;

    //

    // If there is still no message in the queue, there are

    // also no barriers left (else we would have waited), and we quit:

    //

    else

    throw cTerminationException(eENDEDOK);

#else

    cBarrierMessage* barrier = NULL;

    /*

     * Fill up independent event heap

     */

    while (!(sim->msgQueue.empty()))

    {

        msg = sim->msgQueue.removeFirst();

        barrier = dynamic_cast<cBarrierMessage*> (msg);

        if (barrier != NULL)

        {

            /*

             * if we hit a barrier, we are done and return the first independent msg

             * or we have wait at the barrier if no independent event exists

             */

            if (independentEventsHeap.empty())

            {

                barrier->wait();

                delete barrier;

                continue;

            }

            else

            {

                sim->msgQueue.insert(msg);

                return independentEventsHeap.getFirst();

            }

        }

        mod = (cSimpleModule*) sim->getModule(msg->getArrivalModuleId());

        aMod = NULL;

        if (mod->isAsyncModule())

        {

            aMod = (cAsyncModule*) mod;

            simtime_t now = msg->getArrivalTime();

            simtime_t duration = aMod->getProcessingDelay(msg);

            msg->setEventDuration(duration);

            if (!aMod->mayParallelize(msg, duration))

            {

                if (independentEventsHeap.empty())

                {

                    return msg;

                }

                else

                {

                    sim->msgQueue.insert(msg);

                    return independentEventsHeap.getFirst();

                }

            }

            // create a new barrier and schedule it

            cBarrierMessage* barrier = new cBarrierMessage();

            barrier->setArrival(aMod, -1, now + duration);

            msg->setBarrier(barrier);

            sim->msgQueue.insert(barrier);

            printf(

                     "adding to IEH: %s, tstart=%f, tend= %f, now First in IEH: ", SIMTIME_DBL(msg->getArrivalTime()),

                     ((cSimpleModule*) sim->getModule(msg->getArrivalModuleId()))->getName(),

                     SIMTIME_DBL(msg->getTend()));

            independentEventsHeap.insert(msg);

            printf(

                    "%s, length=%i\n",

                    ((cSimpleModule*) sim->getModule(

                            independentEventsHeap.peekFirst()->getArrivalModuleId()))->getName(),

                    independentEventsHeap.length());

        }

        else //Not a AsyncModule

        {

            if (independentEventsHeap.empty())

            {

                return msg;

            }

            else

            {

                sim->msgQueue.insert(msg);

                return independentEventsHeap.getFirst();

            }

        }

    } // while (!(sim->msgQueue.empty()))

/*

 * the FES is empty

 * check if the independent event set is also empty

 */

if (independentEventsHeap.empty())

{

    throw cTerminationException(eENDEDOK);

}

return independentEventsHeap.getFirst();

#endif

}

//-----

Register_Class(cRealTimeScheduler);

void cRealTimeScheduler::startRun()

{

    factor = ev.getConfig()->getAsDouble(CFGID_REALTIMESCHEDULER_SCALING);

    if (factor!=0)

        factor = 1/factor;

    doScaling = (factor!=0);

    gettimeofday(&baseTime, NULL);

}

void cRealTimeScheduler::endRun()

{

}

void cRealTimeScheduler::executionResumed()

{

    gettimeofday(&baseTime, NULL);

    baseTime = timeval_substract(baseTime, SIMTIME_DBL(doScaling ? factor*sim->getSimTime() : sim->getSimTime()));

}

bool cRealTimeScheduler::waitUntil(const timeval& targetTime)

{

    // if there's more than 200ms to wait, wait in 100ms chunks

    // in order to keep UI responsiveness by invoking ev.idle()

    timeval curTime;

    gettimeofday(&curTime, NULL);

    while (targetTime.tv_sec-curTime.tv_sec >=2 ||

           timeval_diff_usec(targetTime, curTime) >= 200000)

    {

        usleep(100000); // 100ms

        if (ev.idle())

            return false;

        gettimeofday(&curTime, NULL);

    }

    // difference is now at most 100ms, do it at once

    long usec = timeval_diff_usec(targetTime, curTime);

    if (usec>0)

        usleep(usec);

    return true;

}

cMessage *cRealTimeScheduler::getNextEvent()

{

#ifdef NOBARRIER

    //

    // If we retrieve a valid msg from the queue, we return it:

    //

    cMessage *msg = sim->msgQueue.peekFirst();

    if (msg)

        return msg;

    //

    // if there is no event left and we don't use the threadpool, end the sim

    //

    if (!sim->threadPool)

        throw cTerminationException(eENDEDOK);

    //

    // If we did not get a valid msg from the queue, but there are still

    // barrier messages left, we wait:

    //

    while (sim->msgQueue.empty() && !sim->threadPool->barrierEmpty())

    {

        __asm__ ("pause");

    }

    msg = sim->msgQueue.peekFirst();

    //

    // If there is a msg now, we return it:

    //

    if (msg)

        return msg;

    //

    // If there is still no message in the queue, there are

    // also no barriers left (else we would have waited), and we quit:

    //

    else

        throw cTerminationException(eENDEDOK);

#else

    cMessage *msg = sim->msgQueue.peekFirst();

    if (!msg)

        throw cTerminationException(eENDEDOK);

    // calculate target time

    simtime_t eventSimtime = msg->getArrivalTime();

    timeval targetTime = timeval_add(baseTime, SIMTIME_DBL(doScaling ? factor*eventSimtime : eventSimtime));

    // if needed, wait until that time arrives

    timeval curTime;

    gettimeofday(&curTime, NULL);

    if (timeval_greater(targetTime, curTime))

    {

        if (!waitUntil(targetTime))

            return NULL; // user break

    }

    else

    {

        // we're behind -- customized versions of this class may alert

        // if we're too much behind, or modify basetime to accept the skew

    }

    // ok, return the message

    return msg;

#endif

}