DataSources

The development of DataSource components is discussed in the DataSource development section.

Developing DataSource components requires a good understanding of the MARTe2 framework (in particular the concept of Brokers – see also BrokerI), as well as solid knowledge of the C++ programming language and real-time programming concepts.

The most important methods in DataSource development are:

  • Initialise: called once at the beginning of the application execution; used to load static parameters.

  • SetConfiguredDatabase: called after the application is set up, when the characteristics of all input signals are known. It can be used to validate signal characteristics and allocate any private memory that depends on them.

  • Synchronise: called by synchronising Broker components (e.g. MemoryMapSynchronisedInputBroker) to update the memory before (or after) a copy.

  • AllocateMemory: responsible for allocating memory for input and output signals.

  • GetSignalMemoryBuffer: returns the pointer to the memory buffer of a given signal.

  • GetBrokerName: returns the name of the Broker(s) that can be used to interact with the DataSource.

Static parameters definition.
1     * Name of the network interface card to monitor
2     */
3    MARTe::StreamString nicName;
4    ///AUTO-GENERATED: END OF PARAMETERS. DO NOT EDIT!
DataSource Initialise method.
1    ///AUTO-GENERATED: INITIALISE PARAMETERS. DO NOT EDIT!
2
3    if (ok) {
4        ok = helper.Read("NICName", nicName);
5    }
6    ///AUTO-GENERATED: END OF INITIALISE PARAMETERS. DO NOT EDIT!

As with GAMs, DataSources are divided into two broad categories: generic and application-specific. Additionally, they are often platform-specific (hardware and operating system dependent).

For application-specific DataSource components, using the marte2_manager to generate the boilerplate code is recommended, as it allows developers to quickly create a new DataSource and focus on implementing its functionality.

Warning

Great care must be taken, as this approach significantly limits design flexibility. For example, the type of Broker is imposed (e.g. MemoryMapSynchronisedInputBroker or MemoryMapSynchronisedOutputBroker), and the implementation runs directly in the context of the RealTimeThread, whereas DataSource logic is typically expected to run in a separate thread.

To create a new DataSource using the marte2_manager, run:

python -m marte2_manager.cli --project_name MARTe2-training-proj --project_path $MARTE_TRAINING_PARENT_DIR -l DEBUG add --cpt_type datasources --cpt_name MyDataSource --cpt_namespace Tutorial

This creates a new DataSource component named MyDataSource in the Tutorial namespace, along with the required Makefiles and tests.

To simplify boilerplate generation, a JSON template can be used to define characteristics such as signals, data types, and static parameters. An example is shown below:

JSON definition for boilerplate generation.
 1{
 2    "parameters": {
 3        "NICName": {
 4            "type": "StreamString",
 5            "description": "Name of the network interface card to monitor",
 6            "number_of_elements": 1,
 7            "required": true
 8        }
 9    },
10    "signals": {
11        "RXPackets": {
12            "type": "uint64",
13            "description": "Number of RX packets",
14            "number_of_elements": 1,
15            "number_of_elements_fixed": true
16        },
17        "TXPackets": {
18            "type": "uint64",
19            "description": "Number of TX packets",
20            "number_of_elements": 1,
21            "number_of_elements_fixed": true
22        },
23        "RXDropped": {
24            "type": "uint64",
25            "description": "Number of dropped RX packets",
26            "number_of_elements": 1,
27            "number_of_elements_fixed": true
28        },
29        "TXDropped": {
30            "type": "uint64",
31            "description": "Number of dropped TX packets",
32            "number_of_elements": 1,
33            "number_of_elements_fixed": true
34        }
35    },
36    "direction": "input"
37}

In this example, a DataSource named SystemMonitor is defined, monitoring statistics by reading from /sys/class/net/lo/statistics/ and /proc/stat.

To update the boilerplate code based on the template:

python -m marte2_manager.cli --project_name MARTe2-training-proj --project_path $MARTE_TRAINING_PARENT_DIR -l DEBUG modify --cpt_type datasources --cpt_name SystemMonitor --cpt_template $MARTE_TRAINING_PARENT_DIR/MARTe2-training-proj/Resources/system_monitor_datasource_template.json

The code can then be compiled and tested using the generated Makefiles:

export TARGET=x86-linux
make -C $MARTE_TRAINING_PARENT_DIR/MARTe2-training-proj/ -f Makefile.gcc core

Note that the DataSource path is automatically added to the MARTeApp.sh script.

Running the application

The DataSource is already included in the configuration file ../Configurations/MassSpring/RTApp-MassSpring-59.cfg, so the application can be run directly after compilation.

./MARTeApp.sh -f ../Configurations/MassSpring/RTApp-MassSpring-59.cfg -l RealTimeLoader -m StateMachine::START

Once running, check the output to verify correct behaviour. Example log entries:

$ [Information - LoggerBroker.cpp:152]: TimeNet [0:0]:1000000
$ [Information - LoggerBroker.cpp:152]: RXPackets [0:0]:5207657751
$ [Information - LoggerBroker.cpp:152]: TXPackets [0:0]:5207657751
$ [Information - LoggerBroker.cpp:152]: RXDropped [0:0]:0
$ [Information - LoggerBroker.cpp:152]: TXDropped [0:0]:0
...

Exercise

The goal of this exercise is to extend the DataSource to compute CPU load by reading and parsing /proc/stat.

Edit Resources/system_monitor_datasource_template.json and add a signal named CPULoad with:

  • number_of_elements=8

  • number_of_elements_fixed=false

This allows dynamic sizing rather than enforcing a fixed number of elements.

  1. Update the boilerplate code:

python -m marte2_manager.cli --project_name tutorial --project_path ~/Projects/MARTe2/Docs/User/source/_static/ modify --cpt_type datasources --cpt_name SystemMonitor --cpt_template ~/Projects/MARTe2/Docs/User/source/_static/tutorial/Resources/system_monitor_datasource_template.json

  1. Modify the generated code to read and parse /proc/stat.

  2. The CPU load is computed from the cpu line:

\[\text{CPULoad} = 1 - \frac{\text{idle} + \text{iowait}}{\text{user} + \text{nice} + \text{system} + \text{iowait} + \text{irq} + \text{softirq} + \text{idle}}\]

  1. Use the MARTe2 stream API to read and parse the file.

  2. Compile the DataSource:

export TARGET=x86-linux
make -C $MARTE_TRAINING_PARENT_DIR/MARTe2-training-proj/ -f Makefile.gcc core

  1. Run the application:

make -C ../Configurations/MassSpring/ -f Makefile.cfg #The configuration file needs to be regenerated to update with the actual number of elements of the CPULoad signal.
./MARTeApp.sh -f ../Configurations/MassSpring/RTApp-MassSpring-60_Gen.cfg -l RealTimeLoader -m StateMachine::START

  1. Verify the output:

...
$ [Warning - Threads.cpp:185]: Failed to change the thread priority (likely due to insufficient permissions)
$ [Information - StateMachine.cpp:340]: In state (INITIAL) triggered message (StartNextStateExecutionRTApp)
$ [Information - LoggerBroker.cpp:152]: TimeMon [0:0]:500000
$ [Information - LoggerBroker.cpp:152]: RXPackets [0:0]:5208123795
$ [Information - LoggerBroker.cpp:152]: TXPackets [0:0]:5208123795
$ [Information - LoggerBroker.cpp:152]: RXDropped [0:0]:0
$ [Information - LoggerBroker.cpp:152]: TXDropped [0:0]:0
$ [Information - LoggerBroker.cpp:152]: CPULoad [0:19]:{ 0.025482 0.041496 0 0 0 0.033175 0.023800 0 0.024148 0.024399 0 0 0 0 0.031807 0.031398 0 0.030271 0.032084 0.031658 }
Solution

Add the signal to the template:

Updated JSON file.
 1{
 2    "parameters": {
 3        "NICName": {
 4            "type": "StreamString",
 5            "description": "Name of the network interface card to monitor",
 6            "number_of_elements": 1,
 7            "required": true
 8        }
 9    },
10    "signals": {
11        "RXPackets": {
12            "type": "uint64",
13            "description": "Number of RX packets",
14            "number_of_elements": 1,
15            "number_of_elements_fixed": true
16        },
17        "TXPackets": {
18            "type": "uint64",
19            "description": "Number of TX packets",
20            "number_of_elements": 1,
21            "number_of_elements_fixed": true
22        },
23        "RXDropped": {
24            "type": "uint64",
25            "description": "Number of dropped RX packets",
26            "number_of_elements": 1,
27            "number_of_elements_fixed": true
28        },
29        "TXDropped": {
30            "type": "uint64",
31            "description": "Number of dropped TX packets",
32            "number_of_elements": 1,
33            "number_of_elements_fixed": true
34        },
35        "CPULoad": {
36            "type": "float64",
37            "description": "CPU load percentage",
38            "number_of_elements": 8,
39            "number_of_elements_fixed": false
40        }
41    },
42    "direction": "input"
43}

Then implement the CPU load computation:

CPU load computation.
 1bool SystemMonitorSol::Synchronise() {
 2    //Skip the first line
 3    if (ok) {
 4        StreamString line;
 5        ok = statFile.GetLine(line);
 6    }
 7    for (uint32 c=0; (c<nOfElementsCPULoad) && (ok); c++) {
 8        StreamString line;
 9        ok = statFile.GetLine(line);
10        if (ok) {
11            ok = line.Seek(0LLU);
12        }
13        if (ok) {
14            cpuLoad[c] = ParseCPULoad(line, c);
15        }
16    }
17    if (ok) {
18        ok = statFile.BasicFile::Seek(0LLU);
19    }
20    return ok;
21}
22
23MARTe::float64 SystemMonitorSol::ParseCPULoad(MARTe::StreamString &line, MARTe::uint32 cpuIdx) {
24    using namespace MARTe;
25    //cpu0 11967760 10418 3650720 720076644 1657 1953761 669881 0 0 0
26    char8 term;
27    StreamString token;
28    (void) line.GetToken(token, " ", term);//skip the cpu id
29    uint64 user = ParseNextCPUToken(line);
30    uint64 nice = ParseNextCPUToken(line); 
31    uint64 system = ParseNextCPUToken(line); 
32    uint64 idle = ParseNextCPUToken(line); 
33    uint64 iowait = ParseNextCPUToken(line); 
34    uint64 irq = ParseNextCPUToken(line); 
35    uint64 softirq = ParseNextCPUToken(line);
36
37    uint64 totalIdle = idle + iowait;
38    uint64 total = user + nice + system + idle + iowait + irq + softirq;
39
40    float64 deltaTotalIdle = static_cast<float64>(totalIdle - lastTotalIdle[cpuIdx]);
41    float64 deltaTotal = static_cast<float64>(total - lastTotal[cpuIdx]);
42    
43    float64 cpuUsage = 0.0f;
44    if (deltaTotal > 0) {
45        cpuUsage = 1.0 - (deltaTotalIdle / deltaTotal);
46    }
47
48    lastTotalIdle[cpuIdx] = totalIdle;
49    lastTotal[cpuIdx] = total;
50
51    return cpuUsage;
52}
53
54
55MARTe::uint64 SystemMonitorSol::ParseNextCPUToken(MARTe::StreamString &line) {
56    using namespace MARTe;
57    uint64 tokenU64 = 0LLU;
58    StreamString token;
59    char8 term;
60    (void) line.GetToken(token, " ", term);
61    (void) TypeConvert(tokenU64, token.Buffer());
62    return tokenU64;
63}