Statistics
| Branch: | Revision:

root / src / sim / casyncmodule.cc @ 342aee1b

History | View | Annotate | Download (14.1 KB)

1
//=========================================================================
2
//  CASYNCMODULE.CC - part of
3
//
4
//                  OMNeT++/OMNEST
5
//           Discrete System Simulation in C++
6
//
7
//
8
//   Member functions of
9
//    cAsyncModule : derived class of cSimpleModule for asynchronous event handling
10
//
11
//  Author: Georg Kunz
12
//
13
//=========================================================================
14

    
15
/*--------------------------------------------------------------*
16
  Copyright (C) 2009 Georg Kunz
17

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

    
22
#include <cstring>
23
#include <stdlib.h>
24

    
25
#include "casyncmodule.h"
26
#include "cthreadpool.h"
27
#include "cconfiguration.h"
28
#include "cconfigoption.h"
29

    
30

    
31
Register_PerRunConfigOption(CFGID_ASYNCMODULE_PARZERODUR, "parallelize-zero-duration-events", CFG_BOOL, "false", "Parallelize Events with zero duration");
32

    
33

    
34
cAsyncModule::cAsyncModule(const char *name, cModule *parent, unsigned stacksize) :
35
                t_end(0.0)
36
{
37
    //
38
        // how many local random number generators do we need? This should become a
39
        // special parameter of every asyncmodule. Not yet implemented.
40
        //
41
        unsigned nrNumGens = 1;
42
        /*
43
         cPar *p = (cPar*)paramv.get("number-local-rngs");
44
         if (p)
45
         nrNumGens = *p;
46

47
         std::cout << "number-local-rngs in " << fullName() << ": " << nrNumGens << std::endl;
48
         */
49
        numGen = new cNumberGenerator(nrNumGens);
50

    
51
        //
52
        // find out if events with 0 durations also be parallelized?
53
        //
54
        parZeroDur = ev.getConfig()->getAsBool(CFGID_ASYNCMODULE_PARZERODUR);
55

    
56
        //
57
        // init state flag to non-busy since no thread is active yet
58
        //
59
        AO_store(&busy, 0);
60

    
61
        //
62
        // needs to be initialized here for use during model initilization
63
        //
64
        scheduledMessageCount = 0;
65
}
66

    
67

    
68
cAsyncModule::~cAsyncModule()
69
{
70
  delete numGen;
71
}
72

    
73

    
74
void cAsyncModule::setMessageMetaData(cMessage* msg)
75
{
76
    msg->setEventDuration(-1);
77
    msg->setSchedulingOrderId(scheduledMessageCount++);
78
    msg->setParentExecutionOrderId(executionOrderId);
79
    msg->setParentStartTime(now);
80
}
81

    
82
int cAsyncModule::scheduleAt(simtime_t t, cMessage *msg)
83
{
84
    setMessageMetaData(msg);
85

    
86
    //
87
    // check current state
88
    //
89
    if (executionState == asynchronous)
90
    {
91
        if (msg==NULL)
92
          throw cRuntimeError("scheduleAt(): message pointer is NULL");
93
        if (t < t_end)
94
                throw cRuntimeError("scheduleAt(): event cannot be scheduled at "
95
                                "%s because this precedes the end of the processing duration "
96
                                "at %s of the scheduling event.",
97
                                t.str().c_str(), t_end.str().c_str());
98

    
99
        // insert this message in the FES. When the scheduler arrives at this
100
        // message it will trigger the processing of the associated task
101
        msg->setSentFrom(this, -1, t_end);
102
        msg->setArrival(this, -1, t);
103

    
104

    
105
        simulation.msgQueue.insert(msg);
106
        return 0;
107
    }
108
    else // synch execution
109
    {
110
        return cSimpleModule::scheduleAt(t, msg);
111
    }
112
}
113

    
114

    
115
void cAsyncModule::callHandleAsyncMessage(cMessage* msg)
116
{
117
    //
118
    // set execution state
119
    //
120
    executionState = asynchronous;
121

    
122
    //
123
    // set the simtime for this event
124
    //
125
    cThreadPool::setSimTime(msg->getArrivalTime());
126
    // corresponds to setContextModule(mod);
127
    cThreadPool::setContext(this);
128
    cThreadPool::setDefaultOwner(this);
129

    
130
    //
131
    // update meta data
132
    //
133
    prepareHandleMessage(msg);
134

    
135
    //
136
    // call the actual method
137
    //
138
    handleMessage(msg);
139
}
140

    
141

    
142
void cAsyncModule::callHandleMessage(cMessage* msg)
143
{
144
    //
145
    // set execution state
146
    //
147
    executionState = synchronous;
148

    
149
    //
150
    // update meta data
151
    //
152
    prepareHandleMessage(msg);
153

    
154
    //
155
    // call the actual method
156
    //
157
    handleMessage(msg);
158
}
159

    
160

    
161
void cAsyncModule::prepareHandleMessage(cMessage* msg)
162
{
163
    //
164
    // set some internal state
165
    //
166
    t_end = msg->getArrivalTime() + msg->getEventDuration();
167
    now = msg->getArrivalTime();
168

    
169
    //
170
    // take ownership of the message only after the test above
171
    //
172
    this->take(msg);
173

    
174
    //
175
    // reset message counter
176
    //
177
    scheduledMessageCount = 0;
178

    
179
    executionOrderId = msg->getExecutionOrderId();
180
}
181

    
182
int cAsyncModule::send(cMessage *msg, int gateid)
183
{
184
    return sendDelayed(msg, 0.0, gateid);
185
}
186

    
187

    
188
int cAsyncModule::send(cMessage *msg, const char *gatename, int sn)
189
{
190
    return sendDelayed(msg, 0.0, gatename, sn);
191
}
192

    
193

    
194
int cAsyncModule::send(cMessage *msg, cGate *outputgate)
195
{
196
    return sendDelayed(msg, 0.0, outputgate);
197
}
198

    
199

    
200
int cAsyncModule::sendDelayed(cMessage *msg, simtime_t delay, int gateid)
201
{
202
    cGate *outgate = gate(gateid);
203
    if (outgate==NULL)
204
        throw cRuntimeError("send()/sendDelayed(): module has no gate #%d", gateid);
205

    
206
    return sendDelayed(msg, delay, outgate);
207
}
208

    
209

    
210
int cAsyncModule::sendDelayed(cMessage *msg, simtime_t delay, const char *gatename, int sn)
211
{
212
    cGate *outgate = gate(gatename, sn);
213
    if (outgate==NULL)
214
       throw cRuntimeError(sn<0 ? "send()/sendDelayed(): module has no gate `%s'":
215
                               "send()/sendDelayed(): module has no gate `%s[%d]'",gatename,sn);
216

    
217
    return sendDelayed(msg, delay, outgate);
218
}
219

    
220

    
221
int cAsyncModule::sendDelayed(cMessage *msg, simtime_t delay, cGate *outputgate)
222
{
223
    setMessageMetaData(msg);
224

    
225
    //
226
    // check current state
227
    //
228
    if (executionState == asynchronous)
229
    {
230
        //
231
        // error checking, omit ownership checks if it breaks
232
        //
233
        if (outputgate==NULL)
234
            throw cRuntimeError("send()/sendDelayed(): gate pointer is NULL");
235
        if (outputgate->getType()=='I')
236
            throw cRuntimeError("send()/sendDelayed(): cannot send via an input gate (`%s')",outputgate->getName());
237
        if (!outputgate->getNextGate())  // NOTE: without this error check, msg would become self-message
238
            throw cRuntimeError("send()/sendDelayed(): gate `%s' not connected",outputgate->getFullName());
239
        if (msg==NULL)
240
            throw cRuntimeError("send()/sendDelayed(): message pointer is NULL");
241
        if (msg->getOwner()!=this)
242
        {
243
            if (this!=simulation.getContext())
244
                throw cRuntimeError("send()/sendDelayed() of module (%s)%s called in the context of "
245
                                        "module (%s)%s: method called from the latter module "
246
                                        "lacks Enter_Method() or Enter_Method_Silent()? "
247
                                        "Also, if message to be sent is passed from that module, "
248
                                        "you'll need to call take(msg) after Enter_Method() as well",
249
                                        getClassName(), getFullPath().c_str(),
250
                                        simulation.getContext()->getClassName(),
251
                                        simulation.getContext()->getFullPath().c_str());
252
            else if (msg->getOwner()==&simulation.msgQueue && msg->isSelfMessage() && msg->getArrivalModuleId()==getId())
253
                throw cRuntimeError("send()/sendDelayed(): cannot send message (%s)%s, it is "
254
                                        "currently scheduled as a self-message for this module",
255
                                        msg->getClassName(), msg->getName());
256
            else if (msg->getOwner()==&simulation.msgQueue && msg->isSelfMessage())
257
                throw cRuntimeError("send()/sendDelayed(): cannot send message (%s)%s, it is "
258
                                        "currently scheduled as a self-message for ANOTHER module",
259
                                        msg->getClassName(), msg->getName());
260
            else if (msg->getOwner()==&simulation.msgQueue)
261
                throw cRuntimeError("send()/sendDelayed(): cannot send message (%s)%s, it is "
262
                                        "currently in scheduled-events, being underway between two modules",
263
                                        msg->getClassName(), msg->getName());
264
            else
265
                throw cRuntimeError("send()/sendDelayed(): cannot send message (%s)%s, "
266
                                        "it is currently contained/owned by (%s)%s",
267
                                        msg->getClassName(), msg->getName(), msg->getOwner()->getClassName(),
268
                                        msg->getOwner()->getFullPath().c_str());
269
        }
270

    
271
        simtime_t t = simTime();
272
        if (delay<0.0)
273
            throw cRuntimeError("sendDelayed(): negative delay %s",delay.str().c_str());
274
        if (t+delay<t_end) // TODO: Consider channel delay!
275
            throw cRuntimeError("sendDelayed(): send delay shorter than processing delay %s", delay.str().c_str());
276

    
277
        //
278
        // set message parameters and send it
279
        //
280
        msg->setSentFrom(this, outputgate->getId(), t+delay);
281
        bool keepit = outputgate->deliver(msg, t+delay);
282
        if (!keepit)
283
            delete msg;
284

    
285
        return 0;
286
    }
287
    else // syncExecution
288
    {
289
        return cSimpleModule::sendDelayed(msg, delay, outputgate);
290
    }
291
}
292

    
293

    
294
int cAsyncModule::sendDirect(cMessage *msg, simtime_t delay, simtime_t duration, cModule *mod, int inputgateid)
295
{
296
    cGate *togate = mod->gate(inputgateid);
297
    if (togate==NULL)
298
        throw cRuntimeError("sendDirect(): module `%s' has no gate #%d",
299
                                mod->getFullPath().c_str(), inputgateid);
300

    
301
    return sendDirect(msg, delay, duration, togate);
302
}
303

    
304

    
305
int cAsyncModule::sendDirect(cMessage *msg, simtime_t delay, simtime_t duration, cModule *mod, const char *inputgatename, int sn)
306
{
307
    if (!mod)
308
        throw cRuntimeError("sendDirect(): module ptr is NULL");
309
    cGate *togate = mod->gate(inputgatename,sn);
310
    if (togate==NULL)
311
        throw cRuntimeError(sn<0 ? "sendDirect(): module `%s' has no gate `%s'":
312
                                "sendDirect(): module `%s' has no gate `%s[%d]'",
313
                                mod->getFullPath().c_str(), inputgatename, sn);
314
    return sendDirect(msg, delay, duration, togate);
315
}
316

    
317

    
318
int cAsyncModule::sendDirect(cMessage *msg, simtime_t delay, simtime_t duration, cGate *togate)
319
{
320
    setMessageMetaData(msg);
321

    
322
    //
323
    // check current state
324
    //
325
    if (executionState == asynchronous)
326
    {
327
        //
328
        // error checking
329
        //
330
        // Note: it is permitted to send to an output gate. It is especially useful
331
        // with several submodules sending to a single output gate of their parent module.
332
        if (togate==NULL)
333
                throw cRuntimeError("sendDirect(): destination gate pointer is NULL");
334
        if (togate->getPreviousGate())
335
                throw cRuntimeError("sendDirect(): module must have dedicated gate(s) for receiving via sendDirect()"
336
                                                                " (\"from\" side of dest. gate `%s' should NOT be connected)",togate->getFullPath().c_str());
337
        if (msg==NULL)
338
                throw cRuntimeError("sendDirect(): message pointer is NULL");
339
        if (msg->getOwner()!=this)
340
        {
341
                if (this!=simulation.getContext())
342
                        throw cRuntimeError("sendDirect() of module (%s)%s called in the context of "
343
                                                                        "module (%s)%s: method called from the latter module "
344
                                                                        "lacks Enter_Method() or Enter_Method_Silent()? "
345
                                                                        "Also, if message to be sent is passed from that module, "
346
                                                                        "you'll need to call take(msg) after Enter_Method() as well",
347
                                                                        getClassName(), getFullPath().c_str(),
348
                                                                        simulation.getContext()->getClassName(),
349
                                                                        simulation.getContext()->getFullPath().c_str());
350
                else if (msg->getOwner()==&simulation.msgQueue && msg->isSelfMessage() && msg->getArrivalModuleId()==getId())
351
                        throw cRuntimeError("sendDirect(): cannot send message (%s)%s, it is "
352
                                                                        "currently scheduled as a self-message for this module",
353
                                                                        msg->getClassName(), msg->getName());
354
                else if (msg->getOwner()==&simulation.msgQueue && msg->isSelfMessage())
355
                        throw cRuntimeError("sendDirect(): cannot send message (%s)%s, it is "
356
                                                                        "currently scheduled as a self-message for ANOTHER module",
357
                                                                        msg->getClassName(), msg->getName());
358
                else if (msg->getOwner()==&simulation.msgQueue)
359
                        throw cRuntimeError("sendDirect(): cannot send message (%s)%s, it is "
360
                                                                        "currently in scheduled-events, being underway between two modules",
361
                                                                        msg->getClassName(), msg->getName());
362
                else
363
                        throw cRuntimeError("sendDirect(): cannot send message (%s)%s, "
364
                                                                        "it is currently contained/owned by (%s)%s",
365
                                                                        msg->getClassName(), msg->getName(), msg->getOwner()->getClassName(),
366
                                                                        msg->getOwner()->getFullPath().c_str());
367
        }
368

    
369
        // set message parameters and send it
370
        simtime_t time = simTime();
371
        if (time + delay < t_end) // TODO: Consider channel delay!
372
            throw cRuntimeError("sendDelayed(): send delay shorter than processing delay %s",delay.str().c_str());
373

    
374
        msg->setSentFrom(this, -1, time);
375

    
376
        bool keepit = togate->deliver(msg, time + delay);
377
        if (!keepit)
378
                delete msg;
379
        return 0;
380
    }
381
    else  // sync execution
382
    {
383
            return cSimpleModule::sendDirect(msg, delay, duration, togate);
384
    }
385
}