Statistics
| Branch: | Revision:

root / src / sim / cscheduler.cc @ fbe00e73

History | View | Annotate | Download (11.2 KB)

1
//=========================================================================
2
//  CSCHEDULER.CC - part of
3
//
4
//                  OMNeT++/OMNEST
5
//           Discrete System Simulation in C++
6
//
7
//  Author: Andras Varga, 2003
8
//
9
//=========================================================================
10

    
11
/*--------------------------------------------------------------*
12
  Copyright (C) 2003-2008 Andras Varga
13
  Copyright (C) 2006-2008 OpenSim Ltd.
14
  Monash University, Dept. of Electrical and Computer Systems Eng.
15
  Melbourne, Australia
16

17
  This file is distributed WITHOUT ANY WARRANTY. See the file
18
  `license' for details on this and other legal matters.
19
*--------------------------------------------------------------*/
20

    
21
#include "cscheduler.h"
22
#include "cmessage.h"
23
#include "csimulation.h"
24
#include "cmessageheap.h"
25
#include "globals.h"
26
#include "cenvir.h"
27
#include "cconfiguration.h"
28
#include "cconfigoption.h"
29
#include "platmisc.h" // usleep
30
#include "cbarriermessage.h"
31
#include "csimplemodule.h"
32

    
33
USING_NAMESPACE
34

    
35
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.");
36

    
37
cScheduler::cScheduler()
38
{
39
    sim = NULL;
40
}
41

    
42
cScheduler::~cScheduler()
43
{
44
}
45

    
46
void cScheduler::setSimulation(cSimulation *_sim)
47
{
48
    sim = _sim;
49
}
50
//-----
51

    
52
Register_Class(cSequentialScheduler);
53

    
54
cMessage *cSequentialScheduler::getNextEvent()
55
{
56
#ifdef NOBARRIER
57
    // Do we have to wait for a barrier?
58
    if (sim->threadPool)
59
        sim->threadPool->waitAtBarrier(&(sim->msgQueue));
60

    
61
    //
62
    // If we retrieve a valid msg from the queue, we return it:
63
    //
64
    cMessage *msg = sim->msgQueue.removeFirst();
65

    
66
    if (msg)
67
        return msg;
68

    
69

    
70
    //
71
    // if there is no event left and we don't use the threadpool, end the sim
72
    //
73
    if (!sim->threadPool)
74
        throw cTerminationException(eENDEDOK);
75

    
76
    //
77
    // If we did not get a valid msg from the queue, but there are still
78
    // barrier messages left, we wait:
79
    //
80
    while (sim->msgQueue.empty() && !sim->threadPool->barrierEmpty())
81
    {
82
        __asm__ ("pause");
83
    }
84
    msg = sim->msgQueue.removeFirst();
85

    
86
    //
87
    // If there is a msg now, we return it:
88
    //
89
    if (msg)
90
        return msg;
91

    
92
    //
93
    // If there is still no message in the queue, there are
94
    // also no barriers left (else we would have waited), and we quit:
95
    //
96
    else
97
        throw cTerminationException(eENDEDOK);
98

    
99
#else
100
    cMessage *msg = NULL;
101
    cBarrierMessage* barrier = NULL;
102

    
103
    while (!msg)
104
    {
105
        msg = sim->msgQueue.removeFirst();
106
        if (!msg)
107
            throw cTerminationException(eENDEDOK);
108

    
109
        //
110
        // If we have a Barriermsg, we wait
111
        //
112
        barrier = dynamic_cast<cBarrierMessage*> (msg);
113
        if (barrier != NULL)
114
        {
115
            //
116
            // wait for the task to complete
117
            //
118
            barrier->wait();
119
            delete barrier;
120
            msg = NULL;
121
        }
122
    }
123

    
124
    cSimpleModule* mod = (cSimpleModule*) sim->getModule(msg->getArrivalModuleId());
125
        simtime_t duration = msg->getEventDuration();
126
        if (duration != SimTime::simTimeSequentialExecution && mod->mayParallelize(msg, duration) && sim->threadPool)
127
        {
128
                //
129
                // create a new barrier and schedule it
130
                //
131
                cBarrierMessage* barrier = new cBarrierMessage();
132
                barrier->setArrival(mod, -1, msg->getArrivalTime() + duration);
133
                msg->setBarrier(barrier);
134

    
135
                // If the event is a null duration event,
136
                // we have to set additional data in the barrier
137
                // to make sure the barrier gets placed correctly
138
                barrier->setSchedulingPriority(msg->getSchedulingPriority());
139
                barrier->setParentStartTime(msg->getArrivalTime());
140
                barrier->setParentExecutionOrderId(sim->getNextExecutionOrderId(msg));
141
                // Set scheduling order to 0 (in front of all children)
142
                barrier->setSchedulingOrderId(0);
143

    
144
                sim->msgQueue.insert(barrier);
145
        }
146
    return msg;
147
#endif
148
}
149

    
150
//-----
151
Register_Class(cEEFScheduler)
152
;
153

    
154
cEEFScheduler::cEEFScheduler()
155
{
156
}
157

    
158
cEEFScheduler::~cEEFScheduler()
159
{
160
}
161

    
162
cMessage *cEEFScheduler::getNextEvent()
163
{
164
    if(!(sim->threadPool)) {
165
        throw cRuntimeError("EEFScheduler is not supported in sequential mode. Activate Threadpool or use cSequentialScheduler.");
166
    }
167

    
168
    updateIES();
169
    return getFirstEvent();
170
}
171

    
172
void cEEFScheduler::updateIES() {
173
    cMessage *msg = NULL;
174
    cSimpleModule* mod = NULL;
175
#ifdef NOBARRIER
176

    
177
    while (!(sim->msgQueue.empty()))
178
    {
179
        msg = sim->msgQueue.removeFirst();
180

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

    
183
        simtime_t duration = msg->getEventDuration();
184

    
185
        if (!mod->mayParallelize(msg, duration)) {
186
            break;
187
        }
188

    
189
        //
190
        // If there is an active barrier in the threadpool before msg, break
191
        //
192
        if (sim->threadPool->isBeforeBarrier(msg)) {
193
            break;
194
        }
195

    
196
        //
197
        // If an event in the IES would cause a barrier before msg, break
198
        //
199
        if (!independentEventsHeap.empty() && msg->getArrivalTime() >= independentEventsHeap.peekFirst()->getTend()) {
200
            break;
201
        }
202
/*        printf(
203
                "adding to IEH: %s, tstart=%f, tend= %f, now First in IEH: ",
204
                ((cSimpleModule*) sim->getModule(msg->getArrivalModuleId()))->getName(),SIMTIME_DBL(msg->getArrivalTime()),
205
                SIMTIME_DBL(msg->getTend()));*/
206

    
207
        independentEventsHeap.insert(msg);
208

    
209
/*        printf(
210
                "%s, length=%i\n",
211
                ((cSimpleModule*) sim->getModule(
212
                        independentEventsHeap.peekFirst()->getArrivalModuleId()))->getName(),
213
                independentEventsHeap.length());*/
214
        msg = NULL;
215
    }
216

    
217
    if (msg)
218
        sim->msgQueue.insert(msg);
219
#else
220
    cBarrierMessage* barrier = NULL;
221
    /*
222
     * Fill up independent event heap
223
     */
224
    while (!(sim->msgQueue.empty()))
225
    {
226
        msg = sim->msgQueue.removeFirst();
227
        barrier = dynamic_cast<cBarrierMessage*> (msg);
228
        if (barrier != NULL)
229
        {
230
            /*
231
             * If the barrier has already been signaled, we can just delete it,
232
             * as we would not have to wait here
233
            */
234
            if(!barrier->isValid()) {
235
                delete barrier;
236
                continue;
237
            }
238
            /*
239
             * if we hit a barrier, we are done and return the first independent msg
240
             * or we have wait at the barrier if no independent event exists
241
             */
242
            if (independentEventsHeap.empty())
243
            {
244
                barrier->wait();
245
                delete barrier;
246
                continue;
247
            }
248
            else
249
            {
250
                sim->msgQueue.insert(msg);
251
                return;
252
            }
253
        }
254

    
255
        mod = (cSimpleModule*) sim->getModule(msg->getArrivalModuleId());
256
                simtime_t duration = msg->getEventDuration();
257

    
258
                if (!mod->mayParallelize(msg, duration))
259
                {
260
                        sim->msgQueue.insert(msg);
261
                        return;
262
                }
263
                // create a new barrier and schedule it
264
                cBarrierMessage* barrier = new cBarrierMessage();
265
                barrier->setArrival(mod, -1, msg->getArrivalTime() + duration);
266
                msg->setBarrier(barrier);
267
                sim->msgQueue.insert(barrier);
268

    
269
                independentEventsHeap.insert(msg);
270

    
271
    } // while (!(sim->msgQueue.empty()))
272
#endif
273
}
274

    
275
cMessage* cEEFScheduler::getFirstEvent() {
276
#ifdef NOBARRIER
277
    cMessage *msg = NULL;
278
    if(independentEventsHeap.empty()) {
279
            if (!sim->msgQueue.empty()) {
280
                // Do we have to wait for a barrier?
281
                if (sim->threadPool) sim->threadPool->waitAtBarrier(&(sim->msgQueue));
282
                return sim->msgQueue.removeFirst();
283
            }
284
        } else {
285
            return independentEventsHeap.getFirst();
286
        }
287
        // At this point, both IES and FES are empty
288
        //
289
        // if there is no event left and we don't use the threadpool, end the sim
290
        //
291
        if (!sim->threadPool)
292
            throw cTerminationException(eENDEDOK);
293

    
294
        //
295
        // If we did not get a valid msg from the queue, but there are still
296
        // barrier messages left, we wait:
297
        //
298
        while (sim->msgQueue.empty() && !sim->threadPool->barrierEmpty())
299
        {
300
            __asm__ ("pause");
301
        }
302
        msg = sim->msgQueue.removeFirst();
303

    
304
        if (msg) return msg;
305

    
306
        //
307
        // If there is still no message in the queue, there are
308
        // also no barriers left (else we would have waited), and we quit:
309
        //
310

    
311
        else
312
            throw cTerminationException(eENDEDOK);
313
#else
314
        if (independentEventsHeap.empty())
315
        {
316
            if (sim->msgQueue.empty()) {
317
                /*
318
                 * Both Queues are empty, we are done
319
                 */
320
                throw cTerminationException(eENDEDOK);
321
            } else {
322
                /*
323
                 * In this case, we have sequential execution
324
                 */
325
                return sim->msgQueue.removeFirst();
326
            }
327
        }
328
        return independentEventsHeap.getFirst();
329
#endif
330
}
331

    
332
//-----
333
Register_Class(cRealTimeScheduler);
334

    
335
void cRealTimeScheduler::startRun()
336
{
337
    factor = ev.getConfig()->getAsDouble(CFGID_REALTIMESCHEDULER_SCALING);
338
    if (factor!=0)
339
        factor = 1/factor;
340
    doScaling = (factor!=0);
341

    
342
    gettimeofday(&baseTime, NULL);
343
}
344

    
345
void cRealTimeScheduler::endRun()
346
{
347
}
348

    
349
void cRealTimeScheduler::executionResumed()
350
{
351
    gettimeofday(&baseTime, NULL);
352
    baseTime = timeval_substract(baseTime, SIMTIME_DBL(doScaling ? factor*sim->getSimTime() : sim->getSimTime()));
353
}
354

    
355
bool cRealTimeScheduler::waitUntil(const timeval& targetTime)
356
{
357
    // if there's more than 200ms to wait, wait in 100ms chunks
358
    // in order to keep UI responsiveness by invoking ev.idle()
359
    timeval curTime;
360
    gettimeofday(&curTime, NULL);
361
    while (targetTime.tv_sec-curTime.tv_sec >=2 ||
362
           timeval_diff_usec(targetTime, curTime) >= 200000)
363
    {
364
        usleep(100000); // 100ms
365
        if (ev.idle())
366
            return false;
367
        gettimeofday(&curTime, NULL);
368
    }
369

    
370
    // difference is now at most 100ms, do it at once
371
    long usec = timeval_diff_usec(targetTime, curTime);
372
    if (usec>0)
373
        usleep(usec);
374
    return true;
375
}
376

    
377
cMessage *cRealTimeScheduler::getNextEvent()
378
{
379
#ifdef NOBARRIER
380
    throw cRuntimeError("RealTimeScheduler is not supported with NOBARRIER.");
381
#endif
382
    if(sim->threadPool) {
383
        //TODO: Handle barriermsg
384
        throw cRuntimeError("RealTimeScheduler is not supported with Threadpool.");
385
    }
386
    cMessage *msg = sim->msgQueue.removeFirst();
387
    if (!msg)
388
        throw cTerminationException(eENDEDOK);
389

    
390
    // calculate target time
391
    simtime_t eventSimtime = msg->getArrivalTime();
392
    timeval targetTime = timeval_add(baseTime, SIMTIME_DBL(doScaling ? factor*eventSimtime : eventSimtime));
393

    
394
    // if needed, wait until that time arrives
395
    timeval curTime;
396
    gettimeofday(&curTime, NULL);
397
    if (timeval_greater(targetTime, curTime))
398
    {
399
        if (!waitUntil(targetTime))
400
            return NULL; // user break
401
    }
402
    else
403
    {
404
        // we're behind -- customized versions of this class may alert
405
        // if we're too much behind, or modify basetime to accept the skew
406
    }
407

    
408
    // ok, return the message
409
    return msg;
410
}