martino/MARTe-Integrated-Components
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixed multicast updated tests

Browse Source
This commit is contained in:
Martino Ferrari
2026-07-25 16:46:25 +02:00
parent 3e0a481c13
commit 915a192b16
16 changed files with 542 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Docs/E2E-Suite.md
+282
View File
@@ -0,0 +1,282 @@
# E2E Test Suite
The streaming-chain end-to-end suite (`Test/E2E/suite/`) validates the full data path from
MARTe2 real-time application through the UDPS wire protocol to StreamHub and client consumers.
It also covers the debug/trace path (DebugService, TCPLogger) and the direct
UDPStreamer-to-UDPStreamerClient round-trip.
## Overview
The suite is driven by a single orchestrator script:
```bash
source env.sh
./Test/E2E/suite/run_e2e.sh [flags]
```
For each scenario defined in `scenarios.py`, the orchestrator:
1. **Generates input data** (`gen_data.py`) — deterministic typed/shaped binary in MARTe2
FileReader format, plus a ground-truth dict for the validator.
2. **Generates configs** (`gen_cfg.py`) — MARTe2 app config (LinuxTimer + FileReader + IOGAM +
UDPStreamer) and StreamHub config, per scenario.
3. **Launches the server stack** — MARTe2 app + StreamHub (for chain/recorder scenarios) or
MARTe2 app alone (for direct/debug scenarios).
4. **Drives mock clients** — the Go `chain-client` (chain scenarios) or `debugclient`
(debug/tcplogger scenarios) connects, records data, and runs behavioural checks.
5. **Validates** (`validate_waveform.py`) — compares the recorded stream against the analytic
ground truth and/or the fed-reference tap file.
6. **Renders plots** (`plots.py`) — waveform, trigger, and zoom overlay PNGs per scenario.
7. **Runs unit tests + coverage** (`collect.py`) — C++ GTest, Go, and Python suites with
optional lcov C++ line coverage.
8. **Runs stress matrix** (`stress_run.py` / `stress.py`) — capacity sweeps (signal size,
count, fan-out, zoom rate) with survival/liveness/RSS/latency gates.
9. **Builds the report** (`report_build.py`) — consolidates everything into
`report_data.json` with regression tracking against the previous run, trend plots, and a
Typst PDF (`E2E_Report.typ`).
---
## Flags
| Flag | Effect |
| -------------------- | -------------------------------------------------------- |
| `--skip-build` | Skip C++ component rebuild |
| `--only <id>` | Run a single scenario by ID |
| `--pdf-only` | Just compile the Typst PDF report (no tests) |
| `--cpp-coverage` | Instrumented gcov rebuild + lcov capture (on by default) |
| `--skip-coverage` | Disable the coverage pass |
| `--skip-stress` | Skip the stress matrix |
| `--skip-datasources` | Skip `direct` scenarios |
| `--skip-recorder` | Skip `recorder` scenarios |
| `--skip-debug` | Skip `debug` and `debug_pause_resume` scenarios |
| `--skip-tcplogger` | Skip `tcplogger` scenarios |
---
## Scenario Kinds
### chain
Full streaming pipeline: MARTe2 (FileReader -> IOGAM -> UDPStreamer) -> StreamHub -> Go
`chain-client`. The client records the live binary stream and runs behavioural checks
(live, zoom, window, trigger). The validator compares the recording against the analytic
ground truth (fidelity, sine shape fit, continuity) and optionally a fed-reference tap.
### direct
MARTe2 FileReader -> UDPStreamer -> UDPStreamerClient -> FileWriter round-trip. Validates that
the written binary matches the input binary (bit-exact for each signal type).
### recorder
MARTe2 -> UDPStreamer -> StreamHub with BinaryRecorder enabled. Validates the `.bin` file
written to disk by the recorder against the original input.
### debug / debug_pause_resume
DebugService scenarios exercising FORCE, TRACE, and BREAK commands over TCP (port 8080) with
trace telemetry on UDP (port 8081). The Go `debugclient` scripts a fixed command sequence and
verifies real acknowledgements. The `debug_pause_resume` variant additionally verifies that
PAUSE halts the RT loop and RESUME restarts it via live VALUE polling.
### tcplogger
TCPLogger delivery: verifies that a triggered DebugService event produces a log line on the
TCPLogger TCP port (8082/9090).
---
## Validation Oracles
Each chain scenario specifies an `oracle` mode:
- **analytic** — ground truth is reconstructed from `gen_data.py`'s deterministic formulas
(sine, ramp, counter, time_us, time_ns). No reference file needed.
- **fed** — a second IOGAM branch in the MARTe config taps the same signals into a FileWriter
("tap file"). The validator compares recordings against this tap.
- **both** — both oracles are applied.
Per-signal checks (`validate_waveform.py`):
| Check | Description |
| ----------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Fidelity** | Every received value within tolerance of some ground-truth value. Tolerance is 0 for raw integers, float epsilon for raw floats, `quant_step/2 + 1e-6*range` for quantised floats. |
| **Shape** | Sine signals (>= 8 points): least-squares fit of `a*sin(wt)+b*cos(wt)+c`. Requires correlation >= 0.99 and low normalised RMSE (relaxed by quant step). |
| **Fed reference** | When `--tap` is given, each received value must also match the tap. |
| **Continuity** | Flags inter-sample gaps > 10x median spacing. Fails when summed gap duration exceeds 5% of capture span. |
---
## Client Checks
The Go `chain-client` (`Test/E2E/suite/client/`) performs behavioural checks specified per
scenario in `client_checks`:
| Check | What it verifies |
| --------- | ----------------------------------------------------------------------------------------------- |
| `live` | WebSocket connection succeeds and live binary pushes arrive with monotonic timestamps. |
| `zoom` | A `zoom` WS command returns a valid binary response covering the requested time range. |
| `window` | A `window` WS command returns data within the specified time bounds. |
| `trigger` | A `trigger` WS command on the specified signal fires and returns data around the trigger point. |
---
## Stress Matrix
The stress module (`stress.py` + `stress_run.py`) exercises capacity by sweeping one load axis
at a time:
| Axis | What is scaled |
| ------------------ | ------------------------------------------------------------ |
| Signal size | Bytes per packet (array element count) |
| Signal count | Number of signals per source |
| Subscriber fan-out | Number of StreamHub instances subscribing to one UDPStreamer |
| WS client count | Parallel WebSocket clients on one StreamHub |
| Zoom request rate | Concurrent zoom queries per second per client |
Gates:
- **Survival** (hard) — neither server crashed or hung.
- **Liveness** (hard) — every client received monotonic, timestamped pushes.
- **Peak RSS** (soft) — MARTe and StreamHub memory stayed under case ceilings.
- **Zoom p95 latency** (soft) — round-trip zoom query latency under load.
Results are written to `stress_results.json` with axis/level for scaling-curve plots.
---
## Artifacts
| Path | Content |
| -------------------------------------------- | ------------------------------------------------------------------- |
| `Build/x86-linux/E2E/chain/results.json` | Per-scenario status (PASS/FAIL/SKIP/XFAIL/XPASS) + waveform metrics |
| `Build/x86-linux/E2E/chain/report_data.json` | Full report data including regression diffs |
| `Build/x86-linux/E2E/chain/history.jsonl` | One-line-per-run headline metrics for trend tracking |
| `Build/x86-linux/E2E/chain/trend_*.png` | Pass-rate / coverage / fidelity / memory trend plots |
| `Build/x86-linux/E2E/chain/E2E_Report.pdf` | Compiled Typst PDF report |
| `Build/x86-linux/E2E/chain/unit_tests.json` | Per-suite test results (GTest, Go, Python) |
| `Build/x86-linux/E2E/chain/coverage.json` | Per-language coverage percentages |
| `Build/x86-linux/E2E/chain/stress/` | Stress matrix results |
| `Build/x86-linux/E2E/chain/hub_<id>.log` | StreamHub stdout/stderr per scenario |
| `Build/x86-linux/E2E/chain/marte_<id>.log` | MARTe2 app stdout/stderr per scenario |
| `Build/x86-linux/E2E/chain/client_<id>.log` | Client stdout/stderr per scenario |
| `/tmp/chain_e2e/` | Scratch: input binaries, configs, recordings, metrics, plots |
---
## XFAIL / XPASS Handling
Scenarios may carry a `known_issue` marker (a human-readable string describing a documented,
not-yet-fixed chain gap). When present:
- A raw **FAIL** is reclassified as **XFAIL** (expected failure) — does not break the green
baseline.
- A raw **PASS** becomes **XPASS** (unexpectedly fixed) — surfaced as a failure to prompt
removal of the stale marker.
Overall status is PASS when there are no hard FAILs and no XPASSes.
---
## Framework Files
| File | Role |
| ---------------------- | --------------------------------------------------------------- |
| `run_e2e.sh` | Top-level orchestrator (build, run scenarios, coverage, report) |
| `scenarios.py` | Declarative scenario matrix + validation |
| `gen_data.py` | Deterministic input binary generator |
| `gen_cfg.py` | MARTe2 + StreamHub config generator |
| `validate_waveform.py` | Waveform comparison (fidelity, shape, continuity) |
| `plots.py` | Per-scenario PNG figure renderer |
| `collect.py` | Unit test runner + coverage collector (GTest, Go, Python, lcov) |
| `report_build.py` | Report data consolidator + trend plots + history |
| `stress.py` | Declarative stress case matrix |
| `stress_run.py` | Stress matrix orchestrator |
| `proc_perf.py` | Live-process CPU/RSS snapshot from `/proc` |
| `E2E_Report.typ` | Typst template for the PDF report |
| `tests_py.py` | Python framework unit tests (`python3 -m unittest tests_py`) |
| `client/main.go` | Go chain-client (live record + zoom/window/trigger checks) |
| `debugclient/main.go` | Go debug/tcplogger client (command scripting + verification) |
---
## Scenario Matrix
| ID | Kind | Description |
| ----------------------------- | ------------------ | ---------------------------------------------------------------------------------------- |
| `s01_scalar_uint32` | chain | Single uint32 scalar counter, Strict unicast (type fidelity) |
| `s02_array_float32_fullarray` | chain | 100-elem float32 array, FullArray time mode, uint64 ns time array |
| `s03_quant_uint16` | chain | float32 scalar quantised to uint16 over [-5,5], Strict unicast |
| `s04_int8_scalar` | chain | int8 scalar counter, type fidelity |
| `s05_uint8_scalar` | chain | uint8 scalar counter, type fidelity |
| `s06_int16_scalar` | chain | int16 scalar ramp, type fidelity |
| `s07_uint16_scalar` | chain | uint16 scalar ramp, type fidelity |
| `s08_int32_scalar` | chain | int32 scalar counter, type fidelity |
| `s09_int64_scalar` | chain | int64 scalar counter, type fidelity |
| `s10_uint64_scalar` | chain | uint64 scalar counter, type fidelity |
| `s11_float64_scalar` | chain | float64 scalar sine 5 Hz (double-precision path) |
| `s12_f32_arr8` | chain | float32 8-elem array sine 5 Hz |
| `s13_f32_arr32` | chain | float32 32-elem array sine 10 Hz |
| `s14_f64_arr64` | chain | float64 64-elem array ramp |
| `s15_i16_arr16` | chain | int16 16-elem array counter |
| `s16_f32_arr256` | chain | float32 256-elem array sine 5 Hz (large frame) |
| `s17_lastsample` | chain | float32 8-elem LastSample, uint64 ns scalar anchor |
| `s18_firstsample` | chain | float32 8-elem FirstSample, uint32 us scalar anchor |
| `s19_fullarray_f64` | chain | float64 50-elem FullArray sine 5 Hz, uint64 ns time |
| `s20_quant_uint8` | chain | float32 scalar quant uint8 [-1,1] sine 5 Hz |
| `s21_quant_int8` | chain | float32 scalar quant int8 [-10,10] sine 5 Hz |
| `s22_quant_int16` | chain | float32 scalar quant int16 [-100,100] ramp |
| `s23_quant_f64_arr` | chain | float64 16-elem quant uint16 [-2,2] sine 5 Hz |
| `s24_accumulate` | chain | float32 scalar sine 5 Hz, Accumulate @50 Hz refresh |
| `s25_decimate4` | chain | float32 scalar sine 5 Hz, Decimate ratio 4 |
| `s26_decimate10_arr` | chain | float32 8-elem counter, Decimate ratio 10 |
| `s27_frag_f64_128` | chain | float64 128-elem ramp, MaxPayload 512 (fragmented) |
| `s28_frag_f32_100` | chain | float32 100-elem sine 5 Hz, MaxPayload 256 (fragmented) |
| `s29_mcast_scalar` | chain | multicast float32 scalar sine 5 Hz |
| `s30_mcast_arr_fullarray` | chain | multicast float32 32-elem FullArray sine 5 Hz |
| `s31_two_src` | chain | two unicast sources: float32 sine + uint32 counter |
| `s32_three_src` | chain | three unicast sources: int16 ramp / float64 sine / uint8 counter |
| `s33_dec_arr_quant` | chain | Decimate 2 + 16-elem quant uint16 sine 5 Hz |
| `s34_acc_fullarray` | chain | Accumulate @100 Hz: accumulated scalar + 32-elem FullArray sine passenger |
| `s35_mcast_decimate` | chain | multicast + Decimate ratio 5, float32 scalar sine 5 Hz |
| `s36_big_frag_dec` | chain | float64 64-elem ramp, MaxPayload 256 + Decimate 4 |
| `s37_trig_ramp_i32` | chain | trigger on int32 ramp scalar |
| `s38_trig_f64_sine` | chain | trigger on float64 sine 5 Hz scalar |
| `s39_uint8_arr32` | chain | uint8 32-elem array counter (wrap fidelity) |
| `s40_int8_arr16` | chain | int8 16-elem array counter (wrap fidelity) |
| `s41_f32_unit` | chain | float32 scalar ramp with Unit=V |
| `s42_f64_counter` | chain | float64 scalar counter (large integer values) |
| `s43_fullarray_quant` | chain | float32 16-elem FullArray quant uint16 sine 5 Hz |
| `s44_window_check` | chain | float32 sine 5 Hz scalar, window time-range check |
| `s45_decimate_multisig` | chain | Decimate ratio 2 over a 2-signal source |
| `s46_accumulate_arr` | chain | Accumulate @200 Hz: accumulated scalar sine + 16-elem array passenger |
| `s47_mcast_multisrc` | chain | multicast, two sources (scalar each) |
| `s48_f64_arr_big_payload` | chain | float64 100-elem ramp, MaxPayload 65490 (single frame) |
| `s49_mixed_quant_raw` | chain | one source: quant uint8 sine + raw float32 sine |
| `s50_trig_quant` | chain | trigger on quantised uint16 sine 10 Hz |
| `s51_8x1msps_100hz` | chain | 8x float32 10k-elem arrays @1 MSps, FirstSample, 100 Hz packets (~32 MB/s) |
| `s52_direct_unicast` | direct | Direct UDPStreamer->UDPStreamerClient round-trip, unicast |
| `s53_direct_multicast` | direct | Direct UDPStreamer->UDPStreamerClient round-trip, multicast |
| `s54_recorder` | recorder | StreamHub BinaryRecorder disk-output round-trip |
| `s55_debug_force_trace_break` | debug | DebugService FORCE/TRACE/BREAK over real TCP 8080 + UDP 8081 |
| `s56_tcplogger_delivery` | tcplogger | TCPLogger delivers a log line for a triggered DebugService event |
| `s57_debug_pause_resume` | debug_pause_resume | DebugService PAUSE/RESUME halts and resumes the RT loop, verified via live VALUE polling |
---
## Coverage Goals
The chain scenario matrix is a curated covering set: every configurable UDPStreamer option
value appears in at least one scenario:
- **All 10 MARTe2 types**: int8, uint8, int16, uint16, int32, uint32, int64, uint64, float32, float64
- **Scalar and array shapes**: elements 1, 8, 16, 32, 50, 64, 100, 128, 256, 1000, 10000
- **All four TimeModes**: PacketTime, FullArray, FirstSample, LastSample
- **All five QuantizedTypes**: none, uint8, int8, uint16, int16
- **All three PublishingModes**: Strict, Accumulate, Decimate
- **Both network modes**: unicast and multicast
- **Fragmentation**: small MaxPayloadSize forcing multi-fragment datagrams
- **Multi-source**: 1, 2, and 3 independent UDPStreamer feeds into one StreamHub
- **High-risk interactions**: decimate+quant+array, accumulate+fullarray, multicast+decimate,
fragmentation+decimate, mixed quant+raw signals
Docs/UDPStreamer.md
+123 -9
View File
@@ -8,7 +8,9 @@ thread.
## Key Features
- **Zero-copy RT path** — `Synchronise()` only locks, copies signal memory, and posts a semaphore.
- **Single-client model** — one client at a time; a new CONNECT replaces the previous session.
- **Unicast and multicast** — unicast (default): single client at a time, new CONNECT replaces
the previous session. Multicast: multiple clients receive data simultaneously by joining
a multicast group; control traffic uses a TCP listener.
- **Packet fragmentation** — large payloads are split into ≤ `MaxPayloadSize`-byte datagrams,
each with a header carrying fragment index and total count so the client can reassemble them.
- **Signal quantization** — `float32`/`float64` signals can be linearly quantized to
@@ -16,6 +18,8 @@ thread.
- **Temporal arrays** — signals with `NumberOfElements > 1` can carry per-sample time
metadata via `TimeMode` and `TimeSignal`, enabling high-frequency burst transmission
(e.g. 1 000 samples per RT cycle at 1 MSps).
- **Publishing modes** — `Strict` (one packet per RT cycle), `Accumulate` (batch N snapshots
then flush on size or time limit), `Decimate` (send every Nth cycle).
---
@@ -26,10 +30,22 @@ thread.
Class = UDPStreamer
// Network
Port = 44500 // UDP port the server listens on (default: 44500)
Port = 44500 // UDP port (unicast) or TCP control port (multicast)
MaxPayloadSize = 1400 // Maximum bytes per UDP datagram (default: 1400)
// Must be > 17 (header size). Tune for MTU.
// Multicast (optional — omit for unicast mode)
MulticastGroup = "239.0.0.1" // IPv4 multicast address (224.0.0.0/4)
Interface = "eth0" // Multicast-bound interface (mandatory when MulticastGroup is set)
DataPort = 44501 // UDP port for multicast DATA (default: Port+1)
// Publishing mode (optional)
PublishingMode = "Strict" // Strict | Accumulate | Decimate
// For Accumulate mode:
MinRefreshRate = 120.0 // Flush frequency in Hz (required for Accumulate)
// For Decimate mode:
Ratio = 10 // Send 1 packet every N RT cycles (required for Decimate)
// Background thread (optional)
CPUMask = 0x2 // CPU affinity mask for the network thread
StackSize = 1048576 // Stack size in bytes (default: 1 MiB)
@@ -66,12 +82,18 @@ thread.
### Top-level Parameters
| Parameter | Type | Default | Description |
| ---------------- | ------ | --------- | ------------------------------------------- |
| `Port` | uint16 | 44500 | UDP server port |
| `MaxPayloadSize` | uint32 | 1400 | Max payload bytes per UDP datagram (min 18) |
| `CPUMask` | uint32 | 0 (any) | Background thread CPU affinity |
| `StackSize` | uint32 | 1 048 576 | Background thread stack size in bytes |
| Parameter | Type | Default | Description |
| ---------------- | ------ | ---------------- | --------------------------------------------------------------------------- |
| `Port` | uint16 | 44500 | UDP server port (unicast) or TCP control port (multicast). Values ≤ 1024 produce a warning. |
| `MaxPayloadSize` | uint32 | 1400 | Max payload bytes per UDP datagram (min 18) |
| `MulticastGroup` | string | *(absent)* | IPv4 multicast address (e.g. `"239.0.0.1"`). Must be in 224.0.0.0/4. Absent or empty = unicast mode. |
| `Interface` | string | *(absent)* | Network interface for multicast binding (e.g. `"eth0"`). **Mandatory** when `MulticastGroup` is set. |
| `DataPort` | uint16 | Port+1 | UDP port for multicast DATA datagrams. Ignored in unicast mode. |
| `PublishingMode` | string | Strict | `Strict`: send every RT cycle. `Accumulate`: batch until size/time limit. `Decimate`: send every Nth cycle. |
| `MinRefreshRate` | float64| — | Flush frequency in Hz. **Required** when `PublishingMode` = `Accumulate`. |
| `Ratio` | uint32 | — | Send 1 packet every `Ratio` RT cycles. **Required** when `PublishingMode` = `Decimate`. |
| `CPUMask` | uint32 | 0xFFFFFFFF (any) | Background thread CPU affinity bitmask |
| `StackSize` | uint32 | MARTe2 default | Background thread stack size in bytes |
### Per-signal Parameters
@@ -113,6 +135,62 @@ wire_value = (uint16)(normalized × 65535)
---
## Network Modes
### Unicast (default)
The server opens a single UDP socket on `Port`. The client initiates the session by sending a
CONNECT packet to that port. The server replies with a CONFIG packet on the same socket and
subsequently sends DATA packets directly to the client's address. One client at a time; a new
CONNECT evicts the previous client.
### Multicast
Enabled by setting `MulticastGroup` to a valid IPv4 multicast address (224.0.0.0/4).
The `Interface` parameter is **mandatory** and specifies the network interface to bind.
The server opens a TCP listener on `Port` for control traffic and a UDP socket aimed at
`MulticastGroup:DataPort` for data traffic. The client:
1. Connects to `Port` via TCP and sends a CONNECT packet.
2. Receives the CONFIG packet over TCP.
3. Joins the multicast group (`MulticastGroup:DataPort`) to receive DATA packets.
Multiple clients may receive data simultaneously by joining the same group.
---
## Publishing Modes
### Strict (default)
Sends one DATA packet for every `Synchronise()` call (every RT cycle). Simplest and lowest
latency.
### Accumulate
Batches multiple RT-cycle snapshots into a single DATA packet. All signals (scalars and arrays)
are accumulated: one full snapshot per RT cycle. The batch is flushed when either:
- **Size condition**: adding one more sample would exceed `MaxPayloadSize`.
- **Time condition**: `1/MinRefreshRate` seconds have elapsed since the last flush.
The maximum batch count is computed automatically from `MaxPayloadSize` and the total wire size
of all signals. Scalar signals with `Unit="us"` or `"ns"` are auto-promoted as the per-sample
FullArray time reference for all other scalars.
Requires `MinRefreshRate` (Hz) to be set.
### Decimate
Sends one DATA packet every `Ratio` RT cycles, dropping intermediate cycles. Only the most
recent snapshot at the Nth cycle is sent.
Requires `Ratio` (≥ 1) to be set. `Ratio = 1` is equivalent to `Strict` mode (a warning is
logged).
---
## Broker
UDPStreamer uses `MemoryMapSynchronisedOutputBroker` for output signals. This broker is
@@ -151,7 +229,7 @@ PrepareNextState() ← opens UDP server socket, starts background threa
---
## Example: minimal scalar streaming
## Example: minimal scalar streaming (unicast)
```
+Data = {
@@ -168,6 +246,26 @@ PrepareNextState() ← opens UDP server socket, starts background threa
}
```
## Example: multicast with accumulation
```
+Streamer = {
Class = UDPStreamer
Port = 44500 // TCP control port
MulticastGroup = "239.0.0.1" // Enables multicast mode
Interface = "eth0" // Mandatory for multicast
DataPort = 44501 // UDP data port (default: Port+1)
MaxPayloadSize = 1400
PublishingMode = "Accumulate"
MinRefreshRate = 60.0 // Flush at least 60 times/s
Signals = {
Time = { Type = uint32; Unit = "us" }
Voltage = { Type = float32; Unit = "V"; RangeMin = -10.0; RangeMax = 10.0; QuantizedType = uint16 }
}
}
```
## Example: high-frequency burst
```
@@ -198,3 +296,19 @@ With `MaxPayloadSize = 1400`, a single 1000-element float32 signal produces:
payload = 8 B (HRT timestamp) + 4 B (T0/uint32) + 4000 B (float32×1000) = 4012 B
fragments = ceil(4012 / 1383) = 3
```
## Example: decimated output
```
+Streamer = {
Class = UDPStreamer
Port = 44500
PublishingMode = "Decimate"
Ratio = 10 // Send 1 packet every 10 RT cycles
Signals = {
Time = { Type = uint32; Unit = "us" }
Position = { Type = float64; Unit = "mm" }
}
}
```