diff --git a/CLAUDE.md b/CLAUDE.md index a5ef047..7dce449 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -39,7 +39,38 @@ cd Client/streamhub-qt && cmake -B build && cmake --build build End-to-end demo script (build + launch full stack, see header for ports/options): `./run_streamhub.sh`. -**Streaming-chain E2E suite** (`Test/E2E/suite/`): `./run_e2e.sh [--skip-build] [--only ] [--cpp-coverage]` drives the full chain per scenario (`scenarios.py`) — generates typed/shaped input + both cfgs, runs MARTe2+StreamHub, records via the Go `chain-client` (live/zoom/window/trigger), and validates the recorded waveform against an analytic/fed oracle (`validate_waveform.py`: fidelity gates correctness, sine shape-fit is a gross-sanity gate + tracked metric pending Phase-A timestamp calibration). It then runs the unit suites + coverage (`collect.py`: C++ GTest, Go, Python; `--cpp-coverage` does an instrumented `--coverage` rebuild, captures with lcov restricted to `Source/*` (the `Test/` harness itself is excluded — it executes every line by construction and would just inflate the number), then restores the clean build), consolidates everything into `report_data.json` with per-field progression/regression vs the previous run and trend plots (`report_build.py`, history in `Build/x86-linux/E2E/chain/history.jsonl`), and compiles a Typst PDF (`E2E_Report.typ`). Python framework unit tests: `python3 -m unittest tests_py` (in `Test/E2E/suite/`). +**Streaming-chain E2E suite** (`Test/E2E/suite/`): + +```bash +./Test/E2E/suite/run_e2e.sh [flags] +``` + +Flags: + +| Flag | Effect | +|---|---| +| `--skip-build` | Skip C++ component rebuild | +| `--only ` | Run a single scenario by ID | +| `--pdf-only` | Just compile the Typst PDF report | +| `--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 (defined in `scenarios.py`): + +- **chain** — full streaming pipeline: MARTe2 → UDPStreamer → StreamHub → Go `chain-client` (live/zoom/window/trigger). Validates recorded waveform against analytic/fed oracle (`validate_waveform.py`: fidelity gates correctness, sine shape-fit is a gross-sanity gate + tracked metric). +- **direct** — MARTe2 FileReader → FileWriter round-trip, validates binary output. +- **recorder** — MARTe2 → StreamHub with history recorder, validates recorded `.bin` file. +- **debug / debug_pause_resume** — DebugService scenarios via the Go `debugclient`. +- **tcplogger** — TcpLogger scenarios via the Go `debugclient`. + +After scenarios, the suite runs unit tests + coverage (`collect.py`: C++ GTest, Go, Python; coverage uses lcov restricted to `Source/*` — the `Test/` harness is excluded), consolidates everything into `report_data.json` with per-field progression/regression vs the previous run and trend plots (`report_build.py`, history in `Build/x86-linux/E2E/chain/history.jsonl`), and compiles a Typst PDF (`E2E_Report.typ`). Artifacts go to `Build/x86-linux/E2E/chain/` (report, logs, PDF) and `/tmp/chain_e2e/` (scratch). Results are aggregated into `results.json` with XFAIL/XPASS handling for known issues. + +Python framework unit tests: `python3 -m unittest tests_py` (in `Test/E2E/suite/`). Build output goes to `Build/x86-linux/` (shared libs per component, `.ex` executables). diff --git a/Docs/E2E-Suite.md b/Docs/E2E-Suite.md new file mode 100644 index 0000000..064c816 --- /dev/null +++ b/Docs/E2E-Suite.md @@ -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 ` | 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_.log` | StreamHub stdout/stderr per scenario | +| `Build/x86-linux/E2E/chain/marte_.log` | MARTe2 app stdout/stderr per scenario | +| `Build/x86-linux/E2E/chain/client_.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 diff --git a/Docs/UDPStreamer.md b/Docs/UDPStreamer.md index 2124a55..f809b3c 100644 --- a/Docs/UDPStreamer.md +++ b/Docs/UDPStreamer.md @@ -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" } + } +} +``` diff --git a/Source/Components/DataSources/UDPStreamer/depends.x86-linux b/Source/Components/DataSources/UDPStreamer/depends.x86-linux index 2e76c11..42db5ce 100644 --- a/Source/Components/DataSources/UDPStreamer/depends.x86-linux +++ b/Source/Components/DataSources/UDPStreamer/depends.x86-linux @@ -1,48 +1,48 @@ ../../../..//Build/x86-linux/Components/DataSources/UDPStreamer/UDPStreamer.o: UDPStreamer.cpp \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/AdvancedErrorManagement.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/ClassProperties.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/GeneralDefinitions.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/TypeCharacteristics.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/ErrorType.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/CompilerTypes.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/Architecture/x86_gcc/CompilerTypes.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/GeneralDefinitions.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/FractionalInteger.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/TypeCharacteristics.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitBoolean.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/TemplateParametersVerificator.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/CompilerTypes.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitRange.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/TypeCharacteristics.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/StreamString.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/GeneralDefinitions.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/TypeCharacteristics.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/GeneralDefinitions.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HeapManager.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HeapI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/ErrorManagement.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/ErrorInformation.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/ErrorType.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitBoolean.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitRange.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/GeneralDefinitions.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/StreamI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/TimeoutType.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HighResolutionTimer.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/TimeStamp.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Architecture/x86_gcc/HighResolutionTimerA.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/CompilerTypes.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/HighResolutionTimer.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/../../HighResolutionTimer.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/HighResolutionTimerCalibrator.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/../../GeneralDefinitions.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/TimeStamp.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/StreamMemoryReference.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/CharBuffer.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitBoolean.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/ErrorManagement.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/BufferedStreamI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/TimeoutType.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/AnyType.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/ClassProperties.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/FractionalInteger.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/ClassRegistryDatabase.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/GlobalObjectsDatabase.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/GlobalObjectI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/StandardHeap.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/../../HeapI.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/../../ErrorManagement.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/../../GeneralDefinitions.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/../Generic/StandardHeap_Generic.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/FastPollingMutexSem.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Atomic.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Architecture/x86_gcc/AtomicA.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/ErrorManagement.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Sleep.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/ClassRegistryItem.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/CString.h \ @@ -53,7 +53,6 @@ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BasicType.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/FractionalInteger.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitRange.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitBoolean.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/ZeroTerminatedArray.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/LinkedListable.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/LinkedListHolderT.h \ @@ -70,17 +69,18 @@ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/StaticListHolder.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/Matrix.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/Vector.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HeapManager.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HeapI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/MemoryOperationsHelper.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/FormatDescriptor.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/IOBuffer.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/CharBuffer.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/StreamI.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/StringHelper.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/StreamStringIOBuffer.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/AdvancedErrorManagement.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/ClassProperties.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/StreamMemoryReference.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L4Configuration/ConfigurationDatabase.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L4Configuration/AnyObject.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/Object.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/StringHelper.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/StructuredDataI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/AnyType.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/CLASSREGISTER.h \ @@ -104,8 +104,6 @@ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/StructuredDataI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L4Configuration/TypeConversion.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/Vector.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/StreamString.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/StreamStringIOBuffer.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/EmbeddedThreadI.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/EmbeddedServiceMethodBinderI.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/ExecutionInfo.h \ @@ -116,18 +114,12 @@ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitSet.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/EmbeddedServiceMethodBinderT.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HighResolutionTimer.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/MemoryMapSynchronisedOutputBroker.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/MemoryMapOutputBroker.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/MemoryMapBroker.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/BrokerI.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/DataSourceI.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/StatefulI.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/ExecutableI.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Sleep.h \ UDPStreamer.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/EmbeddedServiceMethodBinderI.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L1Portability/EventSem.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/MemoryDataSourceI.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/DataSourceI.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/StatefulI.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/SingleThreadService.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/EmbeddedServiceI.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/EmbeddedThreadI.h \ @@ -141,5 +133,6 @@ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HandleI.h \ /home/martino/workspace/MARTe2/Source/Core/FileSystem/L1Portability/Environment/Linux/SocketCore.h \ /home/martino/workspace/MARTe2/Source/Core/FileSystem/L1Portability/BasicUDPSocket.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Sleep.h \ /home/martino/workspace/MARTe2/Source/Core/FileSystem/L1Portability/InternetHost.h \ ../../../..//Common/UDP/UDPSProtocol.h diff --git a/Source/Components/DataSources/UDPStreamer/dependsRaw.x86-linux b/Source/Components/DataSources/UDPStreamer/dependsRaw.x86-linux index 723a38a..7fd2eb1 100644 --- a/Source/Components/DataSources/UDPStreamer/dependsRaw.x86-linux +++ b/Source/Components/DataSources/UDPStreamer/dependsRaw.x86-linux @@ -1,48 +1,48 @@ UDPStreamer.o: UDPStreamer.cpp \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/AdvancedErrorManagement.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/ClassProperties.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/GeneralDefinitions.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/TypeCharacteristics.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/ErrorType.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/CompilerTypes.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/Architecture/x86_gcc/CompilerTypes.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/GeneralDefinitions.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/FractionalInteger.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/TypeCharacteristics.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitBoolean.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/TemplateParametersVerificator.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/CompilerTypes.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitRange.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/TypeCharacteristics.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/StreamString.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/GeneralDefinitions.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/TypeCharacteristics.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/GeneralDefinitions.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HeapManager.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HeapI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/ErrorManagement.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/ErrorInformation.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/ErrorType.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitBoolean.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitRange.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/GeneralDefinitions.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/StreamI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/TimeoutType.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HighResolutionTimer.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/TimeStamp.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Architecture/x86_gcc/HighResolutionTimerA.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/CompilerTypes.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/HighResolutionTimer.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/../../HighResolutionTimer.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/HighResolutionTimerCalibrator.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/../../GeneralDefinitions.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/TimeStamp.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/StreamMemoryReference.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/CharBuffer.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitBoolean.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/ErrorManagement.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/BufferedStreamI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/TimeoutType.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/AnyType.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/ClassProperties.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/FractionalInteger.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/ClassRegistryDatabase.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/GlobalObjectsDatabase.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/GlobalObjectI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/StandardHeap.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/../../HeapI.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/../../ErrorManagement.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/../../GeneralDefinitions.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Environment/Linux/../Generic/StandardHeap_Generic.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/FastPollingMutexSem.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Atomic.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Architecture/x86_gcc/AtomicA.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/ErrorManagement.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Sleep.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/ClassRegistryItem.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/CString.h \ @@ -53,7 +53,6 @@ UDPStreamer.o: UDPStreamer.cpp \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BasicType.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/FractionalInteger.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitRange.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitBoolean.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/ZeroTerminatedArray.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/LinkedListable.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/LinkedListHolderT.h \ @@ -70,17 +69,18 @@ UDPStreamer.o: UDPStreamer.cpp \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/StaticListHolder.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/Matrix.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/Vector.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HeapManager.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HeapI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/MemoryOperationsHelper.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/FormatDescriptor.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/IOBuffer.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/CharBuffer.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/StreamI.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/StringHelper.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/StreamStringIOBuffer.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/AdvancedErrorManagement.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/ClassProperties.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/StreamMemoryReference.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L4Configuration/ConfigurationDatabase.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L4Configuration/AnyObject.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/Object.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/StringHelper.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/StructuredDataI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/AnyType.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/CLASSREGISTER.h \ @@ -104,8 +104,6 @@ UDPStreamer.o: UDPStreamer.cpp \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L2Objects/StructuredDataI.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L4Configuration/TypeConversion.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/Vector.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/StreamString.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L3Streams/StreamStringIOBuffer.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/EmbeddedThreadI.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/EmbeddedServiceMethodBinderI.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/ExecutionInfo.h \ @@ -116,18 +114,12 @@ UDPStreamer.o: UDPStreamer.cpp \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L0Types/BitSet.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/EmbeddedServiceMethodBinderT.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HighResolutionTimer.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/MemoryMapSynchronisedOutputBroker.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/MemoryMapOutputBroker.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/MemoryMapBroker.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/BrokerI.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/DataSourceI.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/StatefulI.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/ExecutableI.h \ - /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Sleep.h \ UDPStreamer.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/EmbeddedServiceMethodBinderI.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L1Portability/EventSem.h \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/MemoryDataSourceI.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/DataSourceI.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L5GAMs/StatefulI.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/SingleThreadService.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/EmbeddedServiceI.h \ /home/martino/workspace/MARTe2/Source/Core/Scheduler/L3Services/EmbeddedThreadI.h \ @@ -141,5 +133,6 @@ UDPStreamer.o: UDPStreamer.cpp \ /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/HandleI.h \ /home/martino/workspace/MARTe2/Source/Core/FileSystem/L1Portability/Environment/Linux/SocketCore.h \ /home/martino/workspace/MARTe2/Source/Core/FileSystem/L1Portability/BasicUDPSocket.h \ + /home/martino/workspace/MARTe2/Source/Core/BareMetal/L1Portability/Sleep.h \ /home/martino/workspace/MARTe2/Source/Core/FileSystem/L1Portability/InternetHost.h \ ../../../..//Common/UDP/UDPSProtocol.h diff --git a/Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.h b/Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.h index da1e7f1..8ab8962 100644 --- a/Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.h +++ b/Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.h @@ -176,9 +176,9 @@ private: uint32 maxPayloadSize; /**< Max payload bytes per datagram. */ uint32 cpuMask; /**< Background thread CPU affinity. */ uint32 stackSize; /**< Background thread stack size. */ - StreamString multicastGroup; /**< Multicast group IP; empty = unicast. */ - uint16 dataPort; /**< UDP port for DATA datagrams (multicast). */ - bool useMulticast; /**< True when MulticastGroup is set. */ + StreamString multicastGroup; /**< Multicast group IP; empty = unicast. */ + uint16 dataPort; /**< UDP port for DATA datagrams (multicast). */ + bool useMulticast; /**< True when MulticastGroup is set. */ /* Signal metadata */ uint32 numSigs; /**< Number of signals. */ diff --git a/Source/Components/Interfaces/UDPStream/UDPSClient.h b/Source/Components/Interfaces/UDPStream/UDPSClient.h index d9691e7..25078b4 100644 --- a/Source/Components/Interfaces/UDPStream/UDPSClient.h +++ b/Source/Components/Interfaces/UDPStream/UDPSClient.h @@ -111,6 +111,7 @@ public: * - ServerAddr (char*) Server IPv4 address. Required. * - Port (uint16) Server UDP port (unicast) or TCP listen port (multicast). Required. * - MulticastGroup (char*) IPv4 multicast address; presence enables multicast mode. + * - Interface (char*) Network interface for multicast join (e.g. "lo"). Required when MulticastGroup is set. * - DataPort (uint16) UDP multicast data port (defaults to Port+1). * - SilenceTimeout (uint32) Seconds of no data before reconnect. Default 5. * - ReconnectDelay (uint32) Seconds to wait between reconnect attempts. Default 2. diff --git a/Source/Components/Interfaces/UDPStream/UDPSServer.cpp b/Source/Components/Interfaces/UDPStream/UDPSServer.cpp index 0a0279e..a273e56 100644 --- a/Source/Components/Interfaces/UDPStream/UDPSServer.cpp +++ b/Source/Components/Interfaces/UDPStream/UDPSServer.cpp @@ -11,8 +11,10 @@ #include "MemoryOperationsHelper.h" #include "StreamString.h" +#include #include #include +#include namespace MARTe { @@ -125,9 +127,17 @@ bool UDPSServer::Start() { if (ok) { tcpListener.SetBlocking(false); } - // UDP data socket connected to multicast group + // UDP data socket connected to multicast group. + // Set IP_MULTICAST_IF so outgoing datagrams leave on the specified + // interface rather than whichever the kernel routing table picks. ok &= dataSocket.Open(); - ok &= dataSocket.Join(multicastGroup.Buffer(), interface.Buffer()); + if (ok) { + struct in_addr localIf; + localIf.s_addr = inet_addr(interface.Buffer()); + int fd = static_cast(dataSocket.GetWriteHandle()); + ok = (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_IF, + &localIf, static_cast(sizeof(localIf))) == 0); + } ok &= dataSocket.Connect(multicastGroup.Buffer(), dataPort); if (!ok) { REPORT_ERROR_STATIC( diff --git a/Test/Configurations/TestApp.cfg b/Test/Configurations/TestApp.cfg index 2768b6a..68d3a65 100644 --- a/Test/Configurations/TestApp.cfg +++ b/Test/Configurations/TestApp.cfg @@ -424,6 +424,7 @@ $TestApp = { Port = 44500 MulticastGroup = "239.0.0.1" DataPort = 44503 + Interface = "127.0.0.1" MaxPayloadSize = 1400 PublishingMode = "Accumulate" MinRefreshRate = 100 diff --git a/Test/Configurations/combined_test.cfg b/Test/Configurations/combined_test.cfg index 634ccae..07a76a2 100644 --- a/Test/Configurations/combined_test.cfg +++ b/Test/Configurations/combined_test.cfg @@ -279,6 +279,7 @@ $App = { Port = 44500 MulticastGroup = "239.0.0.1" DataPort = 44503 + Interface = "127.0.0.1" MaxPayloadSize = 1400 PublishingMode = "Accumulate" MinRefreshRate = 100 diff --git a/Test/Configurations/streamhub_demo.cfg b/Test/Configurations/streamhub_demo.cfg index 5ff4e60..33394ee 100644 --- a/Test/Configurations/streamhub_demo.cfg +++ b/Test/Configurations/streamhub_demo.cfg @@ -411,6 +411,7 @@ $App = { Port = 44500 MulticastGroup = "239.0.0.1" DataPort = 44503 + Interface = "127.0.0.1" MaxPayloadSize = 1400 PublishingMode = "Accumulate" MinRefreshRate = 100 diff --git a/Test/E2E/datasources/E2EMulticastTest.cfg b/Test/E2E/datasources/E2EMulticastTest.cfg index 6cbebfd..0a037e0 100644 --- a/Test/E2E/datasources/E2EMulticastTest.cfg +++ b/Test/E2E/datasources/E2EMulticastTest.cfg @@ -64,6 +64,7 @@ $E2EMulticastTest = { Port = 44600 MulticastGroup = "239.0.0.1" DataPort = 44610 + Interface = "127.0.0.1" MaxPayloadSize = 65507 PublishingMode = "Strict" Signals = { diff --git a/Test/E2E/suite/__pycache__/gen_cfg.cpython-314.pyc b/Test/E2E/suite/__pycache__/gen_cfg.cpython-314.pyc index fc43031..85ce53f 100644 Binary files a/Test/E2E/suite/__pycache__/gen_cfg.cpython-314.pyc and b/Test/E2E/suite/__pycache__/gen_cfg.cpython-314.pyc differ diff --git a/Test/E2E/suite/client/chain-client b/Test/E2E/suite/client/chain-client index d578e07..98f3c50 100755 Binary files a/Test/E2E/suite/client/chain-client and b/Test/E2E/suite/client/chain-client differ diff --git a/Test/E2E/suite/debugclient/debugclient b/Test/E2E/suite/debugclient/debugclient index cee52f8..24ee852 100755 Binary files a/Test/E2E/suite/debugclient/debugclient and b/Test/E2E/suite/debugclient/debugclient differ diff --git a/Test/E2E/suite/gen_cfg.py b/Test/E2E/suite/gen_cfg.py index faccfc9..b918377 100644 --- a/Test/E2E/suite/gen_cfg.py +++ b/Test/E2E/suite/gen_cfg.py @@ -17,6 +17,8 @@ expected, not an error. """ import argparse import os +import re +import subprocess import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) @@ -25,6 +27,41 @@ import scenarios as S # noqa: E402 PRODUCER_HZ = 1000 # LinuxTimer frequency (Hz) +def _iface_to_ip(name): + """Resolve a network interface name (e.g. ``"wlan0"``) to its first IPv4 + address via ``ip -4 addr show ``. Returns ``"127.0.0.1"`` on failure.""" + try: + out = subprocess.check_output( + ["ip", "-4", "addr", "show", name], + stderr=subprocess.DEVNULL, text=True) + m = re.search(r"inet\s+(\d+\.\d+\.\d+\.\d+)", out) + if m: + return m.group(1) + except (subprocess.CalledProcessError, FileNotFoundError, OSError): + pass + return "127.0.0.1" + + +def _mcast_interface_ip(group): + """Return the IPv4 address of the OS-selected interface for a multicast + group. MARTe2's ``BasicUDPSocket::Join`` expects an IP address (passed to + ``inet_addr``), not an interface name. + + Uses ``ip route get `` to find the outgoing device, then resolves + its IP. Falls back to ``"127.0.0.1"`` if the route or device lookup fails. + """ + try: + out = subprocess.check_output( + ["ip", "route", "get", group], + stderr=subprocess.DEVNULL, text=True) + m = re.search(r"dev\s+(\S+)", out) + if m: + return _iface_to_ip(m.group(1)) + except (subprocess.CalledProcessError, FileNotFoundError, OSError): + pass + return "127.0.0.1" + + def _ndims(elements): return 0 if elements == 1 else 1 @@ -78,6 +115,8 @@ def _streamer_block(src, scenario): if scenario["network"] == "multicast": parts.append(f'MulticastGroup = "{src["multicast_group"]}"') parts.append(f"DataPort = {src['data_port']}") + iface = src.get("interface") or _mcast_interface_ip(src["multicast_group"]) + parts.append(f'Interface = "{iface}"') sigs = " ".join(_streamer_sig(sig) for sig in s) return (f" +Streamer_{src['id']} = {{ Class = UDPStreamer " f"{' '.join(parts)} Signals = {{ {sigs} }} }}")