From e3bc1f72f306b9b00ce035861b11d144a0052505 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Mon, 2 Nov 2026 01:18:43 +0100 Subject: [PATCH] flatudp: vector rows as channels or as a waveform, a re-openable source, probes for both Co-Authored-By: Claude Opus 4.6 --- docs/claude/build-and-run.md | 17 +- docs/claude/probe-readings.md | 79 +++++--- docs/claude/verification.md | 36 ++-- src/source/flatudp/flatudp_signals.cpp | 65 +++++- src/source/flatudp/flatudp_signals.hpp | 18 +- src/source/flatudp/flatudp_source.cpp | 47 ++++- src/source/flatudp/flatudp_source.hpp | 3 +- src/ui_app.hpp | 5 +- src/ui_probe.cpp | 270 ++++++++++++++++++++++--- tools/mock_flatudp.py | 6 +- 10 files changed, 446 insertions(+), 100 deletions(-) diff --git a/docs/claude/build-and-run.md b/docs/claude/build-and-run.md index 8e819c2..b525c21 100644 --- a/docs/claude/build-and-run.md +++ b/docs/claude/build-and-run.md @@ -140,20 +140,27 @@ concatenated in declaration order, no header, no padding). The datagram carries settings are the whole contract: `port` 44488, `bind`, `multicast` and `iface` (a group to join and the local interface for it), `rate` (the sender's cycle rate in Hz; **required**, it sizes the ring and the ingest decimation and spaces the samples within a frame), `endian` `little|big`, and `signals` -`name:type[count],...` with types `uint8/16/32/64`, `int8/16/32/64`, `float32`, `float64` (a count above 1 -makes a vector signal: one channel whose `count` values per datagram are consecutive samples, as a UDPStreamer -block signal), e.g. `Trigger:uint8,Ia:float32,Wave:float32[16]`. A datagram of any other size than +`name:type[count][w],...` with types `uint8/16/32/64`, `int8/16/32/64`, `float32`, `float64`, e.g. +`Trigger:uint8,Ia:float32,Wave:float32[16]`. A count above 1 makes a vector signal, shown one of two ways: by +default as `count` **channels** `Wave[0]`..`Wave[15]` of one sample per cycle each (batched like any scalar +list); with a trailing `w` (`Wave:float32[16]w`) as ONE **waveform** channel receiving `count` consecutive +samples per cycle at `rate * count`, as a UDPStreamer block signal. A waveform row needs a count above 1, a +list with one holds only waveform rows of one count (no zero-padding of scalars), and a list may make at most +256 channels (a scalar 1, a `channels` row `count`, a waveform 1). A datagram of any other size than the list implies is dropped and counted as stale; each distinct wrong size is reported once on stderr as `[flatudp] ...`. Loss is never reported: the stream has no counter. `tools/mock_flatudp.py` is a sender for a signal spec (`--rate`, `--signals`, `--big-endian`, `--gap S,D`, `--bad-every S`, `--multicast G --mcast-if A`, `--log FILE`). ```sh -python3 tools/mock_flatudp.py --port 44588 --rate 1000 --signals Trigger:uint8,Ia:float32,Wave:int16[4] & +python3 tools/mock_flatudp.py --port 44588 --rate 1000 --signals 'Trigger:uint8,Ia:float32,Wave:int16[4]' & ./build/udpscope --source flatudp --source-opt port=44588 --source-opt rate=1000 \ - --source-opt signals=Trigger:uint8,Ia:float32,Wave:int16[4] + --source-opt 'signals=Trigger:uint8,Ia:float32,Wave:int16[4]' ``` +(Quote the `signals` value: zsh globs the `[4]`. The mock takes the same text with or without the trailing `w`; +the datagram is the same, `w` only says how the app shows it.) + `tools/mock_udpstreamer.py` has waveform-injection options used by the probes: - `--square-harmonics N` shapes `square`: `N` odd harmonics of the ideal step's Fourier diff --git a/docs/claude/probe-readings.md b/docs/claude/probe-readings.md index a862689..e2266a1 100644 --- a/docs/claude/probe-readings.md +++ b/docs/claude/probe-readings.md @@ -1913,46 +1913,61 @@ the one `[e1608] handshake with ...: Connection refused` line, `stream: ... sign ``` [flatudp] codec: ok=1 [flatudp] datagram of 22 bytes, expected 21, check the signal list -[flatudp] stream: connected=1 signals=4 names=Trigger,Ia,Cnt,Wave expect=Trigger,Ia,Cnt,Wave frames=3735 rate=1000.000 ok=1 -[flatudp] channels: count=4 expect=4 ok=1 -[flatudp] continuity: ch0_breaks=0 ch1_breaks=0 ok=1 -[flatudp] wrong size: count=3 message='datagram of 22 bytes, expected 21, check the signal list' ok=1 +[flatudp] stream: connected=1 signals=7 names=Trigger,Ia,Cnt,Wave[0],Wave[1],Wave[2],Wave[3] expect=Trigger,Ia,Cnt,Wave[0],Wave[1],Wave[2],Wave[3] frames=368 rate=1000.000 ok=1 +[flatudp] channels: count=7 expect=7 ok=1 +[flatudp] continuity: Trigger_breaks=0 Ia_breaks=0 Cnt_breaks=0 Wave[0]_breaks=0 Wave[1]-[0]_breaks=0 ok=1 +[flatudp] wrong size: count=4 message='datagram of 22 bytes, expected 21, check the signal list' ok=1 [flatudp] capture button: present=1 sensitive=1 -[flatudp] state: valid=3735 wrong=3 -[flatudp] silence: connected=0 model_connected=0 ok=1 -[flatudp] resumed: connected=1 frames_before=4990 frames_after=7147 ok=1 +[flatudp] state: valid=3683 wrong=4 +[flatudp] silence: connected=0 model_connected=0 after_ms=4300 ok=1 +[flatudp] resumed: connected=1 frames_before=401 frames_after=410 ok=1 ``` -`UDPSCOPE_PROBE_FLATUDP=2500` (the first lines; counts vary) and `UDPSCOPE_PROBE_FLATUDP_GAP=8000` (the last two) -run as `--source flatudp` against `tools/mock_flatudp.py` with the options in `verification.md`. The second line is -the source's own stderr message, printed once per distinct wrong size, not a probe line. The `continuity` line -above is the scalar-list run (`Trigger,Ia,Cnt`, big endian); the mixed-list run with `Wave:int16[4]` reads -`continuity: skipped (...)`. With no sender the stream lines read `connected=0 ... frames=0 ... ok=0`, `continuity` -is skipped and the capture button reads `sensitive=0`: a reading, not a failure. A run without `--bad-every` reads -`wrong size: count=0 ... ok=0` for the same reason. +and, for the waveform-only list `Wave:int16[4]w`: -- `codec: ok=1`: the `signals` text round-trips, a hand-built 24-byte datagram decodes to the hand-computed - values in both byte orders (`uint8` 254, `int16` -2, `float32` 1.5, `uint64[2]` 1 and 4294967298, `int8` -1) - and the validity rules refuse what they should. It needs no source, so `--source probe` shows it too. - `ok=0` is a codec defect: a swapped byte order, a wrong width or a rule that does not fire. -- `stream: ... ok=1`: the model's signals are the list's rows in order, at the declared rate. `names` out of - order is a decode that mixed rows; `rate` of 1000 against a different declared rate is a `sampling_rate` - that did not reach the model (the ring and the decimation are then sized for the wrong cadence). -- `channels: count=4`: one channel per signal. A vector signal is **not** one channel per element: the model - reads its `count` values as consecutive samples of one channel, as for a UDPStreamer block signal. -- `continuity: ch0_breaks=0 ch1_breaks=0`: consecutive samples of the mock's first two signals step by 1 and - by 2 modulo 100. A break is a lost or reordered datagram, a batch flushed with the wrong number of samples, - a wrong byte order or ingest decimation that is not 1:1. It is skipped for a list with a vector row: the - model spans each frame by its largest element block, so the scalar channels of a mixed list are padded - with zeros (`Frame::value` reads 0 out of range) and a continuity check on them would read nearly every - step as a break. +``` +[flatudp] stream: connected=1 signals=1 names=Wave expect=Wave frames=3737 rate=4000.000 ok=1 +[flatudp] channels: count=1 expect=1 ok=1 +[flatudp] continuity: wave_breaks=0 ok=1 +``` + +`UDPSCOPE_PROBE_FLATUDP=2500` (the first lines; counts vary) and `UDPSCOPE_PROBE_FLATUDP_GAP=1500` (the last two) +run as `--source flatudp` against `tools/mock_flatudp.py` with the options in `verification.md`. The second line is +the source's own stderr message, printed once per distinct wrong size, not a probe line. With no sender the stream +lines read `connected=0 ... frames=0 ... ok=0`, `continuity` is skipped and the capture button reads +`sensitive=0`: a reading, not a failure. A run without `--bad-every` reads `wrong size: count=0 ... ok=0` for the +same reason. + +- `codec: ok=1`: the `signals` text round-trips (including a `w` row and a bracketed count of 1 on a waveform), a + hand-built 24-byte datagram decodes to the hand-computed values in both byte orders (`uint8` 254, `int16` -2, + `float32` 1.5, `uint64[2]` 1 and 4294967298, `int8` -1), so do a `float64`, `uint16`, `int32`, `uint32` and + `int64`; `decode_channels` puts a `channels` vector in `count` columns and a waveform in one; and the validity + rules refuse what they should (a waveform of count 1, a waveform mixed with another row or of another length, + 257 channels). It needs no source, so `--source probe` shows it too. `ok=0` is a codec defect: a swapped byte + order, a wrong width or a rule that does not fire. +- `stream: ... ok=1`: the model's signals are the list's channels in order, each at its declared rate: `rate` + for a scalar or a `channels` element (named `Name[k]`), `rate * count` for a waveform. `names` out of order is a + decode that mixed rows; a rate that is not the expected one is a `sampling_rate` that did not reach the model + (the ring and the decimation are then sized for the wrong cadence; a waveform at `rate` instead of `rate * 4` + has a time axis 4 times too long). +- `channels: count=7`: one channel per `SignalInfo`, which is `channel_count(rows)`: a scalar 1, a `channels` + vector `count`, a waveform 1 (its `count` values are consecutive samples of that one channel). +- `continuity: _breaks=0 ...`: the mock's row k carries `(i * (k + 1) + e) % 100`, so consecutive samples + of a channel from row k step by k + 1 modulo 100 (`Trigger` 1, `Ia` 2, `Cnt` 3, `Wave[0]` 4), and `Wave[1]-[0]` + is the same-sample difference of two elements, which must be 1. A break is a lost or reordered datagram, a + batch flushed with the wrong number of samples, an element mapped to the wrong channel, a wrong byte order + or ingest decimation that is not 1:1. A waveform list reads `wave_breaks`: samples step by 1 inside a + datagram and by `2 - count` into the next, at a phase the probe finds from the first such step and then holds; + a non-zero count is a block of the wrong length or one that dropped samples. (Before the `channels` form, a + mixed list showed ~400 breaks here: a scalar next to a vector was zero-padded to the vector's length.) - `wrong size: ok=1`: wrong-size datagrams were counted and the message names the received and expected size. The message must also be the only stderr line of its kind: one per distinct size, however many arrive. - `capture button: present=1 sensitive=1`: the kind reports datagrams, so the capture UI is live (contrast `[source-kinds]`, where the probe kind leaves it disabled). With no stream it reads `sensitive=0`. -- `silence: connected=0 model_connected=0`: 2 s without a valid datagram raised `Disconnected`. `connected=1` - is the timeout not firing, wrong-size datagrams refreshing the silence clock (they must not), or the mock's - gap not covering the probe (see the timing note in `verification.md`). +- `silence: connected=0 model_connected=0 after_ms=...`: 2 s without a valid datagram raised `Disconnected` + (`after_ms` is the wait from the stream coming up to the drop; it only reads the gap's start in the app's + clock). `connected=1` is the timeout not firing or wrong-size datagrams refreshing the silence clock (they + must not); `never connected` is a stream that never came up before the probe gave up after 20 s. - `resumed: ok=1`: `Connected` was raised again and frames resumed. `connected=0` here means the source never recovered after the sender came back. diff --git a/docs/claude/verification.md b/docs/claude/verification.md index 57cf967..3aafbea 100644 --- a/docs/claude/verification.md +++ b/docs/claude/verification.md @@ -443,22 +443,28 @@ precedent, not an exception. later checks the frame count moved on. The reading's own entry is in `probe-readings.md`. - `UDPSCOPE_PROBE_FLATUDP=2500` with `UDPSCOPE_EXIT_AFTER_MS`, run as `--source flatudp` against `tools/mock_flatudp.py` -- the flat UDP source end to end: a hand-computed codec and decode check (it - needs no source), the stream's signal names and declared rate, the channel count (one per signal), value - continuity across the first two signals (all-scalar lists only), the wrong-size accounting and the capture - button. It enables the signals itself, so do **not** add `UDPSCOPE_ENABLE_ALL_MS`. Runs, each short and - serial, with the mock killed afterwards (quote the `signals` value: zsh globs the `[4]`): - - **mixed list, gap and bad datagrams:** mock `--rate 1000 --signals Trigger:uint8,Ia:float32,Cnt:uint64,Wave:int16[4] - --gap 9,5 --bad-every 1 --log /tmp/flatudp-mock.log` started 1 s before the app, the same `signals` as a - `--source-opt`, with `UDPSCOPE_PROBE_FLATUDP=2500 UDPSCOPE_PROBE_FLATUDP_GAP=8000 UDPSCOPE_EXIT_AFTER_MS=14500`; - stderr carries exactly one `datagram of 22 bytes, expected 21` line. The mock's clock starts about 4 s - before the app's probe clock (`xvfb-run` and the app's own startup), so a gap of 9 s to 14 s on the mock's - clock covers the app's 8 s probe (silence) and ends before its 12 s one (resumed); a mock gap measured in - the app's own seconds would miss both; - - **scalar list, big endian (batching, continuity):** mock `--signals Trigger:uint8,Ia:float32,Cnt:uint64 --big-endian`, - app `--source-opt endian=big`, `UDPSCOPE_PROBE_FLATUDP=2500 UDPSCOPE_EXIT_AFTER_MS=5000`; + needs no source; it covers the `w` suffix, the waveform and 256-channel rules, the channel decode and the + float64/uint16/int32/uint32/int64 types in both byte orders), the stream's channel names (`Wave[0]`.. + for a `channels` vector) and declared rates (`rate * count` for a waveform), the channel count, value + continuity per row (and the offset between a vector's elements, or a waveform's block phase), the + wrong-size accounting and the capture button. It enables the signals itself, so do **not** add + `UDPSCOPE_ENABLE_ALL_MS`. Runs, each short and serial, with the mock killed afterwards (quote the `signals` + value: zsh globs the `[4]`): + - **mixed list, gap and bad datagrams:** mock `--rate 1000 --signals 'Trigger:uint8,Ia:float32,Cnt:uint64,Wave:int16[4]' + --gap 8,4 --bad-every 1` started 1 s before the app, the same `signals` as a `--source-opt`, with + `UDPSCOPE_PROBE_FLATUDP=2500 UDPSCOPE_PROBE_FLATUDP_GAP=1500 UDPSCOPE_EXIT_AFTER_MS=14500`; stderr carries + exactly one `datagram of 22 bytes, expected 21` line. The silence probe is a 100 ms poll that waits for the + stream, then for the 2 s of silence to take the source down, then for it to come back, so it does not + depend on when `xvfb-run` finished starting; the only requirement is that the mock's gap begins after + the app connected (the mock's clock starts about 4 s before the app's) and ends in time to read the return + (`--gap 8,4` with a 14.5 s run does both with several seconds of margin); + - **waveform-only list:** mock `--signals 'Wave:int16[4]w'`, app `--source-opt 'signals=Wave:int16[4]w'`, + `UDPSCOPE_PROBE_FLATUDP=2500 UDPSCOPE_EXIT_AFTER_MS=5000`: one channel at `rate=4000`; + - **big endian (decode of every width through the stream):** mock `--big-endian` on the mixed list, app + `--source-opt endian=big`, `UDPSCOPE_PROBE_FLATUDP=2500 UDPSCOPE_EXIT_AFTER_MS=5000`; - **no sender:** `connected=0` throughout, a reading and not a failure. -- `UDPSCOPE_PROBE_FLATUDP_GAP=8000` -- reads the source inside the mock's silence (2 s after its last datagram - the source is disconnected) and 4 s later checks it reconnected and the stream moved on. +- `UDPSCOPE_PROBE_FLATUDP_GAP=1500` -- starts the silence poll described above (2 s after the source's last + datagram it must be disconnected; then it must reconnect and the frame count move on). - `UDPSCOPE_PROBE_SEARCH_MS` — runs a buffer search for the armed condition, cross-checks the live trigger's own `t0` against the nearest hit (`agree=1` is the load-bearing reading), and jumps to hit 0. Reports the resulting `mode=`, `axis=` and `marks placed=`. diff --git a/src/source/flatudp/flatudp_signals.cpp b/src/source/flatudp/flatudp_signals.cpp index fb67e90..0631d16 100644 --- a/src/source/flatudp/flatudp_signals.cpp +++ b/src/source/flatudp/flatudp_signals.cpp @@ -87,12 +87,23 @@ bool all_scalar(const std::vector& rows) { return std::all_of(rows.begin(), rows.end(), [](const Row& r) { return r.count == 1; }); } +bool any_wave(const std::vector& rows) { + return std::any_of(rows.begin(), rows.end(), [](const Row& r) { return r.wave; }); +} + +int channel_count(const std::vector& rows) { + long n = 0; + for (const Row& r : rows) n += r.wave ? 1 : static_cast(r.count); + return static_cast(std::min(n, 1L << 30)); +} + std::string format(const std::vector& rows) { std::string out; for (const Row& r : rows) { if (!out.empty()) out += ','; out += r.name + ':' + type_name(r.type); - if (r.count != 1) out += '[' + std::to_string(r.count) + ']'; + if (r.count != 1 || r.wave) out += '[' + std::to_string(r.count) + ']'; + if (r.wave) out += 'w'; } return out; } @@ -118,11 +129,19 @@ std::string parse(const std::string& text, std::vector& rows) { std::string ty = trim(tok.substr(colon + 1)); const std::size_t br = ty.find('['); if (br != std::string::npos) { - if (ty.back() != ']') { + const std::size_t cl = ty.rfind(']'); + if (cl == std::string::npos || cl < br) { rows.clear(); return where + "missing ']' after the count"; } - const std::string n = trim(ty.substr(br + 1, ty.size() - br - 2)); + const std::string tail = trim(ty.substr(cl + 1)); + if (tail == "w") { + r.wave = true; + } else if (!tail.empty()) { + rows.clear(); + return where + "unexpected '" + tail + "' after the count (only w, for a waveform)"; + } + const std::string n = trim(ty.substr(br + 1, cl - br - 1)); const bool digits = !n.empty() && n.size() <= 6 && std::all_of(n.begin(), n.end(), [](char c) { return c >= '0' && c <= '9'; }); if (!digits) { @@ -159,6 +178,30 @@ std::string problem(const std::vector& rows) { if (n > kMaxDatagram) return std::to_string(n) + " bytes per datagram exceed the " + std::to_string(kMaxDatagram) + " a UDP datagram can carry"; + + // A waveform is one channel of `count` samples per datagram; it cannot share a frame with a + // channel of another length, so a list of waveforms is all waveforms of one length. + const Row* first_wave = nullptr; + for (const Row& r : rows) { + if (!r.wave) continue; + if (r.count < 2) return "signal '" + r.name + "' is a waveform and needs a count above 1"; + if (first_wave == nullptr) first_wave = &r; + } + if (first_wave != nullptr) { + for (const Row& r : rows) { + if (!r.wave) + return "signal '" + r.name + "' is not a waveform but '" + first_wave->name + + "' is; every row must be a waveform of the same length"; + if (r.count != first_wave->count) + return "signal '" + r.name + "' is a waveform of " + std::to_string(r.count) + + " values but '" + first_wave->name + "' has " + std::to_string(first_wave->count) + + "; every row must be a waveform of the same length"; + } + } + const int channels = channel_count(rows); + if (channels > kMaxChannels) + return std::to_string(channels) + " channels exceed the " + std::to_string(kMaxChannels) + + " one list may make (a vector row makes one per element unless it is a waveform)"; return {}; } @@ -176,4 +219,20 @@ void decode_datagram(const std::vector& rows, const std::uint8_t* data, boo } } +void decode_channels(const std::vector& rows, const std::uint8_t* data, bool big_endian, + std::vector>& cols) { + const std::size_t channels = static_cast(channel_count(rows)); + if (cols.size() < channels) cols.resize(channels); + const bool swap = big_endian != (std::endian::native == std::endian::big); + std::size_t off = 0, ch = 0; + for (const Row& r : rows) { + const std::size_t w = type_size(r.type); + for (std::uint32_t k = 0; k < r.count; ++k) { + cols[r.wave ? ch : ch + k].push_back(load_typed(r.type, data + off, swap)); + off += w; + } + ch += r.wave ? 1 : r.count; + } +} + } // namespace udpscope::flatudp diff --git a/src/source/flatudp/flatudp_signals.hpp b/src/source/flatudp/flatudp_signals.hpp index a27ccb5..ca4f027 100644 --- a/src/source/flatudp/flatudp_signals.hpp +++ b/src/source/flatudp/flatudp_signals.hpp @@ -15,12 +15,16 @@ enum class Type : std::uint8_t { U8 = 0, I8, U16, I16, U32, I32, U64, I64, F32, constexpr std::uint32_t kMaxCount = 4096; constexpr std::size_t kMaxDatagram = 65507; ///< the largest UDP payload over IPv4 +constexpr int kMaxChannels = 256; ///< the most channels one list may make /// One signal of the sender's datagram: `count` consecutive values of `type`. struct Row { std::string name; Type type = Type::F32; std::uint32_t count = 1; + /// A vector row (count > 1) only: true is ONE channel of `count` consecutive samples per + /// datagram (a waveform); false is `count` channels `name[0]`.. of one sample each. + bool wave = false; bool operator==(const Row&) const = default; }; @@ -33,7 +37,13 @@ std::size_t datagram_size(const std::vector& rows); /// True when no row is a vector. Such a list can be batched, several datagrams to a frame. bool all_scalar(const std::vector& rows); -/// `name:type[,name:type[count]...]`; a count of 1 is written without brackets. +/// True when some row is a waveform (`wave`). +bool any_wave(const std::vector& rows); +/// The channels the list makes: a scalar 1, a non-wave vector `count`, a waveform 1. +int channel_count(const std::vector& rows); + +/// `name:type[,name:type[count][w]...]`; a count of 1 is written without brackets, and the `w` +/// of a waveform row follows its count. std::string format(const std::vector& rows); /// The inverse of `format`. Whitespace around names, types and counts is ignored. An empty text /// is an empty list (and not an error: `problem` reports it). Returns an error in words, empty @@ -43,6 +53,12 @@ std::string parse(const std::string& text, std::vector& rows); /// The first reason this list cannot describe a datagram, in words for the user; empty when valid. std::string problem(const std::vector& rows); +/// Decodes one datagram of `datagram_size(rows)` bytes to one column per *channel* (see +/// `channel_count`): a non-wave vector row feeds `count` columns of one value each, a waveform row +/// one column of `count` values. Appends; the caller clears the columns between uses. +void decode_channels(const std::vector& rows, const std::uint8_t* data, bool big_endian, + std::vector>& cols); + /// Decodes one datagram of `datagram_size(rows)` bytes to doubles, row by row, byte-swapping when /// `big_endian` differs from the host. Appends to `cols[s]`; the caller clears them between uses. void decode_datagram(const std::vector& rows, const std::uint8_t* data, bool big_endian, diff --git a/src/source/flatudp/flatudp_source.cpp b/src/source/flatudp/flatudp_source.cpp index 93b8f58..b4f3ebc 100644 --- a/src/source/flatudp/flatudp_source.cpp +++ b/src/source/flatudp/flatudp_source.cpp @@ -90,7 +90,7 @@ std::string FlatUdpSettings::identity() const { } std::size_t FlatUdpSettings::block() const { - if (!all_scalar(rows) || !(rate > 0.0) || !std::isfinite(rate)) return 1; + if (any_wave(rows) || !(rate > 0.0) || !std::isfinite(rate)) return 1; const long b = std::lround(rate / 100.0); return static_cast(std::min(1024L, std::max(1L, b))); } @@ -127,20 +127,44 @@ void FlatUdpSource::open(SourceSink& sink) { if (!p.empty()) throw std::runtime_error(p); sink_ = &sink; + // A fresh session: Stop then Go live opens the same object again, and the model needs its CONFIG. + configured_ = false; + connected_.store(false, std::memory_order_relaxed); + ever_connected_ = false; + recv_error_noted_ = false; + reported_sizes_.clear(); + frame_counter_ = 0; + frames_ = 0; + reconnects_ = 0; + valid_.store(0, std::memory_order_relaxed); + wrong_.store(0, std::memory_order_relaxed); + + // One SignalInfo per channel: a scalar, one element of a `channels` vector (`Name[k]`), or a whole + // waveform (rows = count consecutive samples at rate * count). signals_.clear(); - for (const Row& r : cfg_.rows) { + const auto add = [&](const std::string& name, const Row& r, std::uint32_t rows, double rate) { SignalInfo s; - s.name = r.name; + s.name = name; s.type_code = static_cast(r.type); - s.rows = r.count; + s.rows = rows; s.cols = 1; s.time_mode = kTimePacket; - s.sampling_rate = cfg_.rate; + s.sampling_rate = rate; signals_.push_back(s); + }; + for (const Row& r : cfg_.rows) { + if (r.wave) { + add(r.name, r, r.count, cfg_.rate * static_cast(r.count)); + } else if (r.count == 1) { + add(r.name, r, 1, cfg_.rate); + } else { + for (std::uint32_t k = 0; k < r.count; ++k) + add(r.name + "[" + std::to_string(k) + "]", r, 1, cfg_.rate); + } } expected_ = datagram_size(cfg_.rows); block_ = cfg_.block(); - pending_.assign(cfg_.rows.size(), {}); + pending_.assign(signals_.size(), {}); pending_n_ = 0; rx_.assign(65536, 0); @@ -223,14 +247,19 @@ void FlatUdpSource::poll(int timeout_ms) { void FlatUdpSource::drain_socket() { for (int i = 0; i < 8192; ++i) { // bounded, so a flood cannot starve the timers - const ssize_t n = ::recv(fd_, rx_.data(), rx_.size(), 0); + // MSG_TRUNC: the return value is the datagram's real length even if it did not fit `rx_`. + const ssize_t n = ::recv(fd_, rx_.data(), rx_.size(), MSG_TRUNC); if (n < 0) { if (errno == EINTR) continue; + if (errno != EAGAIN && errno != EWOULDBLOCK && !recv_error_noted_) { + recv_error_noted_ = true; + note(std::string("receive failed: ") + std::strerror(errno)); + } break; // EAGAIN: drained } const double t = epoch_seconds(); const std::size_t len = static_cast(n); - if (tap_) tap_(rx_.data(), len, t); + if (tap_) tap_(rx_.data(), std::min(len, rx_.size()), t); if (len == expected_) { accept(rx_.data()); } else { @@ -248,7 +277,7 @@ void FlatUdpSource::accept(const std::uint8_t* data) { connected_.store(true, std::memory_order_relaxed); sink_->on_event(SourceEvent::Connected, ""); } - decode_datagram(cfg_.rows, data, cfg_.big_endian, pending_); + decode_channels(cfg_.rows, data, cfg_.big_endian, pending_); if (pending_n_ == 0) batch_start_ = last_valid_; if (++pending_n_ >= block_) flush(); } diff --git a/src/source/flatudp/flatudp_source.hpp b/src/source/flatudp/flatudp_source.hpp index 412565f..b1bbd67 100644 --- a/src/source/flatudp/flatudp_source.hpp +++ b/src/source/flatudp/flatudp_source.hpp @@ -40,7 +40,7 @@ struct FlatUdpSettings { std::string problem() const; /// `[multicast:]port`; names the session file. std::string identity() const; - /// Datagrams per frame: 1 for a list with a vector, else `clamp(round(rate / 100), 1, 1024)`. + /// Datagrams per frame: 1 for a list with a waveform row, else `clamp(round(rate / 100), 1, 1024)`. std::size_t block() const; }; @@ -78,6 +78,7 @@ class FlatUdpSource final : public DataSource { int fd_ = -1; bool configured_ = false; bool ever_connected_ = false; + bool recv_error_noted_ = false; ///< a failing `recv` is logged once per open std::size_t expected_ = 0; ///< bytes of a valid datagram std::size_t block_ = 1; diff --git a/src/ui_app.hpp b/src/ui_app.hpp index c0229ed..76497c8 100644 --- a/src/ui_app.hpp +++ b/src/ui_app.hpp @@ -1885,8 +1885,9 @@ gboolean e1608_probe_reconnect(gpointer user); gboolean e1608_probe_resumed(gpointer user); /// The flat UDP (MARTe2 UDPSender) source: the signal-list codec and a hand-computed decode, then, /// against `tools/mock_flatudp.py`, the stream, the value continuity, the wrong-size accounting and -/// the capture button (`_report`, a second later). `_gap` reads the source inside the mock's silence; -/// `_resumed` shows the stream came back. +/// the capture button (`_report`, a second later). `_gap` starts a 100 ms poll that waits for the +/// stream, then for the mock's silence to drop the source (whenever it falls), then (`_resumed`) for +/// the stream to come back. gboolean flatudp_probe(gpointer user); gboolean flatudp_probe_report(gpointer user); gboolean flatudp_probe_gap(gpointer user); diff --git a/src/ui_probe.cpp b/src/ui_probe.cpp index b74acfd..2c24f19 100644 --- a/src/ui_probe.cpp +++ b/src/ui_probe.cpp @@ -2103,11 +2103,90 @@ bool flatudp_codec_check() { if (valid({{"a,b", Type::U8, 1}})) return false; if (valid({{"A", Type::U8, 0}})) return false; if (valid({{"A", Type::U8, 4097}})) return false; - if (!valid({{"A", Type::U8, 4096}})) return false; - // 2 x 32768 = 65536 bytes is over the datagram limit; 32768 + 16384 is not. - if (prob({{"A", Type::F64, 4096}, {"B", Type::F64, 4096}}).find("65536") == std::string::npos) + // 4096 is the count limit; as `channels` it would be 4096 channels, so it is a waveform here. + if (!valid({{"A", Type::U8, 4096, true}})) return false; + if (valid({{"A", Type::U8, 4096, false}})) return false; + // 2 x 32768 = 65536 bytes is over the datagram limit; 32768 + 16384 is not (waveforms of one length). + if (prob({{"A", Type::F64, 4096, true}, {"B", Type::F64, 4096, true}}).find("65536") == std::string::npos) return false; - if (!valid({{"A", Type::F64, 4096}, {"B", Type::F32, 4096}})) return false; + if (!valid({{"A", Type::F64, 4096, true}, {"B", Type::F32, 4096, true}})) return false; + + // The `w` suffix: a waveform row. It round-trips, `format` writes it only for a waveform, and a + // count of 1 is still bracketed when the row is one. + const std::vector mixed = {{"S", Type::U8, 1, false}, {"V", Type::F32, 3, false}, + {"W", Type::I16, 16, true}, {"X", Type::U8, 1, true}}; + const std::string text = format(mixed); + if (text != "S:uint8,V:float32[3],W:int16[16]w,X:uint8[1]w") return false; + std::vector back; + if (!parse(text, back).empty() || back != mixed) return false; + if (!parse("W : int16 [ 4 ] w", back).empty() || back.size() != 1 || !back[0].wave || + back[0].count != 4 || format(back) != "W:int16[4]w") + return false; + if (parse("W:int16[4]x", back).empty() || !back.empty()) return false; + if (parse("W:int16w", back).empty()) return false; + if (parse("W:int16[4", back).empty()) return false; + if (!any_wave(mixed) || any_wave(want)) return false; + + // The waveform rules: count above 1; all rows waveforms, of one length. + if (valid({{"W", Type::I16, 1, true}})) return false; + if (prob({{"W", Type::I16, 1, true}}).find("'W'") == std::string::npos) return false; + if (!valid({{"W", Type::I16, 4, true}})) return false; + if (!valid({{"W", Type::I16, 4, true}, {"X", Type::F32, 4, true}})) return false; + if (valid({{"W", Type::I16, 4, true}, {"X", Type::F32, 8, true}})) return false; + if (valid({{"W", Type::I16, 4, true}, {"S", Type::F32, 1, false}})) return false; + if (valid({{"S", Type::F32, 1, false}, {"W", Type::I16, 4, true}})) return false; + if (valid({{"W", Type::I16, 4, true}, {"V", Type::F32, 4, false}})) return false; + if (prob({{"W", Type::I16, 4, true}, {"S", Type::F32, 1, false}}).find("'S'") == std::string::npos) + return false; + if (prob({{"W", Type::I16, 4, true}, {"X", Type::F32, 8, true}}).find("'X'") == std::string::npos) + return false; + + // The channel total: scalar 1, vector `count`, waveform 1; at most kMaxChannels. + if (channel_count({{"S", Type::U8, 1, false}, {"V", Type::U8, 5, false}, {"W", Type::U8, 5, true}}) != 7) + return false; + if (!valid({{"V", Type::U8, 256, false}})) return false; + if (valid({{"V", Type::U8, 257, false}})) return false; + if (valid({{"V", Type::U8, 256, false}, {"S", Type::U8, 1, false}})) return false; + if (prob({{"V", Type::U8, 257, false}}).find("256") == std::string::npos) return false; + + // Channel decode: a `channels` vector feeds `count` columns of one value, a waveform one column of + // `count`. U16 = 0x1234, I32 = -2, then V[3] of U8 = 7 8 9, then a waveform of two I16 = 1 -1. + const std::vector cr = {{"A", Type::U16, 1, false}, {"B", Type::I32, 1, false}, + {"V", Type::U8, 3, false}, {"W", Type::I16, 2, true}}; + const std::uint8_t cd[] = {0x34, 0x12, 0xFE, 0xFF, 0xFF, 0xFF, 7, 8, 9, 0x01, 0x00, 0xFF, 0xFF}; + if (datagram_size(cr) != sizeof cd || channel_count(cr) != 6) return false; + std::vector> cc; + decode_channels(cr, cd, false, cc); + const std::vector> cwant = {{4660}, {-2}, {7}, {8}, {9}, {1, -1}}; + if (cc != cwant) return false; + + // The wide and unsigned types, little then big endian: a float64, uint16, int32, uint32 and int64. + // F64 -2.5, U16 65535, I32 -123456, U32 4000000000, I64 -5000000000. + const std::vector tr = {{"F", Type::F64}, {"U", Type::U16}, {"I", Type::I32}, + {"L", Type::U32}, {"Q", Type::I64}}; + const std::uint8_t tle[] = {0, 0, 0, 0, 0, 0, 0x04, 0xC0, // -2.5 as 0xC004000000000000 + 0xFF, 0xFF, // 65535 + 0xC0, 0x1D, 0xFE, 0xFF, // -123456 = 0xFFFE1DC0 + 0x00, 0x28, 0x6B, 0xEE, // 4000000000 = 0xEE6B2800 + 0x00, 0x0E, 0xFA, 0xD5, 0xFE, 0xFF, 0xFF, 0xFF}; // -5e9 = 0xFFFFFFFED5FA0E00 + if (datagram_size(tr) != sizeof tle) return false; + std::uint8_t tbe[sizeof tle]; + { + std::size_t off = 0; + for (const Row& r : tr) { + const std::size_t w = type_size(r.type); + for (std::size_t b = 0; b < w; ++b) tbe[off + b] = tle[off + w - 1 - b]; + off += w; + } + } + const double twant[5] = {-2.5, 65535.0, -123456.0, 4000000000.0, -5000000000.0}; + for (int pass = 0; pass < 2; ++pass) { + std::vector> cols; + decode_datagram(tr, pass == 0 ? tle : tbe, pass == 1, cols); + if (cols.size() != 5) return false; + for (std::size_t k = 0; k < 5; ++k) + if (cols[k].size() != 1 || cols[k][0] != twant[k]) return false; + } return true; } @@ -2115,8 +2194,6 @@ flatudp::FlatUdpSource* flatudp_source(App* a) { return a->acq != nullptr ? dynamic_cast(a->acq->source()) : nullptr; } -std::uint64_t g_flatudp_frames_before = 0; - /// The mock's signal k carries (i * (k + 1) + element) % 100 in packet i, so two consecutive /// samples of a scalar signal differ by k + 1 modulo 100. Returns the number of steps that do not, /// over a window clear of the head the producer is writing; -1 when it cannot be read. @@ -2281,7 +2358,91 @@ gboolean e1608_probe_reconnect(gpointer user) { return G_SOURCE_REMOVE; } -/// After the mock's gap: the stream's frame count moved on and the source is connected again. +/// The mock's channels, as the model must list them: a scalar `Name`, the elements of a `channels` +/// vector `Name[k]`, a waveform `Name` at `rate * count`. +struct FlatudpExpect { + std::vector names; + std::vector rates; +}; + +static FlatudpExpect flatudp_expect(const flatudp::FlatUdpSettings& s) { + FlatudpExpect e; + for (const flatudp::Row& r : s.rows) { + if (r.wave) { + e.names.push_back(r.name); + e.rates.push_back(s.rate * double(r.count)); + } else if (r.count == 1) { + e.names.push_back(r.name); + e.rates.push_back(s.rate); + } else { + for (std::uint32_t k = 0; k < r.count; ++k) { + e.names.push_back(r.name + "[" + std::to_string(k) + "]"); + e.rates.push_back(s.rate); + } + } + } + return e; +} + +/// Breaks in a waveform channel as the mock emits it: datagram i carries i, i+1 .. i+count-1, so +/// consecutive samples differ by 1 and the step into the next datagram by 2 - count (mod 100). +/// The block phase is read from the first such step, then held for the whole window. +static long flatudp_wave_breaks(App* a, std::size_t channel, long count) { + if (channel >= a->model->channels().size()) return -1; + const ChannelBuffer& ch = a->model->live().channel(channel); + const u64 n = ch.size(); + const u64 want = 400; + if (n <= want + 50) return -1; + std::vector v(static_cast(want)); + ch.raw(n - want - 50, n - 50, v.data()); + const auto step = [&](std::size_t i) { + const long d = std::lround(double(v[i])) - std::lround(double(v[i - 1])); + return ((d % 100) + 100) % 100; + }; + const long jump = (((2 - count) % 100) + 100) % 100; + long phase = -1; + for (std::size_t i = 1; i < v.size() && phase < 0; ++i) { + if (step(i) == jump) phase = static_cast(i % std::size_t(count)); + } + if (phase < 0) return static_cast(v.size()); + long breaks = 0; + for (std::size_t i = 1; i < v.size(); ++i) { + const long want_step = long(i % std::size_t(count)) == phase ? jump : 1; + if (step(i) != want_step) ++breaks; + } + return breaks; +} + +/// Samples of two channels at the same index, `b` minus `a`, modulo 100: the mock's element e of a +/// vector row adds e to the row's value, so element k+1 minus element k is 1 at every sample. +static long flatudp_offset_breaks(App* app, std::size_t a, std::size_t b, long offset) { + if (a >= app->model->channels().size() || b >= app->model->channels().size()) return -1; + const ChannelBuffer& ca = app->model->live().channel(a); + const ChannelBuffer& cb = app->model->live().channel(b); + const u64 n = std::min(ca.size(), cb.size()); + const u64 want = 400; + if (n <= want + 50) return -1; + std::vector va(static_cast(want)), vb(va.size()); + ca.raw(n - want - 50, n - 50, va.data()); + cb.raw(n - want - 50, n - 50, vb.data()); + long breaks = 0; + for (std::size_t i = 0; i < va.size(); ++i) { + const long d = std::lround(double(vb[i])) - std::lround(double(va[i])); + if (((d % 100) + 100) % 100 != offset) ++breaks; + } + return breaks; +} + +/// The silence probe is a small state machine on a 100 ms tick, so it needs no knowledge of when the +/// sender's gap falls (the mock's clock starts seconds before the app's): wait until the stream is +/// up, then for the 2 s silence to take the source down, then for it to come back. +struct FlatudpGapState { + int phase = 0; // 0 waiting for the stream, 1 for the silence, 2 for the return + int ticks = 0; + std::uint64_t frames_before = 0; +}; +static FlatudpGapState g_flatudp_gap; + gboolean flatudp_probe_resumed(gpointer user) { App* a = static_cast(user); if (a->window == nullptr) return G_SOURCE_REMOVE; @@ -2289,27 +2450,55 @@ gboolean flatudp_probe_resumed(gpointer user) { const std::uint64_t after = a->model->stats().frames.load(std::memory_order_relaxed); const bool up = src != nullptr && src->connected() && a->model->stats().connected.load(std::memory_order_relaxed); + if (!(up && after > g_flatudp_gap.frames_before) && ++g_flatudp_gap.ticks < 200) return G_SOURCE_CONTINUE; std::fprintf(stderr, "[flatudp] resumed: connected=%d frames_before=%" G_GUINT64_FORMAT " frames_after=%" G_GUINT64_FORMAT " ok=%d\n", - up ? 1 : 0, g_flatudp_frames_before, after, - up && after > g_flatudp_frames_before ? 1 : 0); + up ? 1 : 0, g_flatudp_gap.frames_before, after, + up && after > g_flatudp_gap.frames_before ? 1 : 0); std::fflush(stderr); return G_SOURCE_REMOVE; } -/// Inside the mock's gap: 2 s of silence must have raised `Disconnected`. -gboolean flatudp_probe_gap(gpointer user) { +/// Polls the silence: the source must go down within 20 s of the stream coming up. +static gboolean flatudp_probe_gap_poll(gpointer user) { App* a = static_cast(user); if (a->window == nullptr) return G_SOURCE_REMOVE; flatudp::FlatUdpSource* src = flatudp_source(a); if (src == nullptr) return G_SOURCE_REMOVE; + FlatudpGapState& g = g_flatudp_gap; const bool src_up = src->connected(); const bool model_up = a->model->stats().connected.load(std::memory_order_relaxed); - g_flatudp_frames_before = a->model->stats().frames.load(std::memory_order_relaxed); - std::fprintf(stderr, "[flatudp] silence: connected=%d model_connected=%d ok=%d\n", src_up ? 1 : 0, - model_up ? 1 : 0, !src_up && !model_up ? 1 : 0); - std::fflush(stderr); - g_timeout_add(4000, flatudp_probe_resumed, a); + ++g.ticks; + if (g.phase == 0) { + if (src_up && model_up) { + g.phase = 1; + g.ticks = 0; + } else if (g.ticks >= 200) { + std::fprintf(stderr, "[flatudp] silence: never connected ok=0\n"); + std::fflush(stderr); + return G_SOURCE_REMOVE; + } + return G_SOURCE_CONTINUE; + } + if (!src_up || g.ticks >= 200) { + // The model learns of it through the sink a moment after the source drops: read it a tick later. + if (!src_up && model_up && g.ticks < 200) return G_SOURCE_CONTINUE; + g.frames_before = a->model->stats().frames.load(std::memory_order_relaxed); + std::fprintf(stderr, "[flatudp] silence: connected=%d model_connected=%d after_ms=%d ok=%d\n", + src_up ? 1 : 0, model_up ? 1 : 0, g.ticks * 100, !src_up && !model_up ? 1 : 0); + std::fflush(stderr); + g.ticks = 0; + g.phase = 2; + g_timeout_add(100, flatudp_probe_resumed, a); + return G_SOURCE_REMOVE; + } + return G_SOURCE_CONTINUE; +} + +/// Starts the silence probe (see `FlatudpGapState`). +gboolean flatudp_probe_gap(gpointer user) { + g_flatudp_gap = FlatudpGapState{}; + g_timeout_add(100, flatudp_probe_gap_poll, user); return G_SOURCE_REMOVE; } @@ -2324,16 +2513,17 @@ gboolean flatudp_probe_report(gpointer user) { const bool up = a->model->stats().connected.load(std::memory_order_relaxed); const std::vector& sigs = a->model->signals(); + const FlatudpExpect ex = flatudp_expect(s); std::string names, expect_names; - std::size_t expect_channels = 0; // one channel per signal: a vector's elements are consecutive samples - for (const SignalInfo& si : sigs) names += (names.empty() ? "" : ",") + si.name; - for (const flatudp::Row& r : s.rows) { - expect_names += (expect_names.empty() ? "" : ",") + r.name; - ++expect_channels; + bool rates_ok = sigs.size() == ex.rates.size(); + for (std::size_t i = 0; i < sigs.size(); ++i) { + names += (names.empty() ? "" : ",") + sigs[i].name; + if (i < ex.rates.size() && std::fabs(sigs[i].sampling_rate - ex.rates[i]) > 1e-6) rates_ok = false; } + for (const std::string& n : ex.names) expect_names += (expect_names.empty() ? "" : ",") + n; const double rate = sigs.empty() ? 0.0 : sigs.front().sampling_rate; const std::uint64_t frames = a->model->stats().frames.load(std::memory_order_relaxed); - const bool stream_ok = up && names == expect_names && std::fabs(rate - s.rate) < 1e-6 && frames > 0; + const bool stream_ok = up && names == expect_names && rates_ok && frames > 0; std::fprintf(stderr, "[flatudp] stream: connected=%d signals=%zu names=%s expect=%s frames=%" G_GUINT64_FORMAT " rate=%.3f ok=%d\n", @@ -2341,18 +2531,38 @@ gboolean flatudp_probe_report(gpointer user) { stream_ok ? 1 : 0); const std::size_t channels = a->model->channels().size(); + const std::size_t expect_channels = static_cast(flatudp::channel_count(s.rows)); std::fprintf(stderr, "[flatudp] channels: count=%zu expect=%zu ok=%d\n", channels, expect_channels, channels == expect_channels ? 1 : 0); - // Only the first two signals are checked, and only in an all-scalar list: a vector row makes the - // model span each frame by its elements, and the scalar channels are then padded with zeros. - if (up && s.rows.size() >= 2 && flatudp::all_scalar(s.rows)) { - const long b0 = flatudp_breaks(a, 0, 1); - const long b1 = flatudp_breaks(a, 1, 2); - std::fprintf(stderr, "[flatudp] continuity: ch0_breaks=%ld ch1_breaks=%ld ok=%d\n", b0, b1, - b0 == 0 && b1 == 0 ? 1 : 0); + // The mock's row k, element e, carries (i * (k + 1) + e) % 100 in datagram i, so consecutive + // samples of a channel from row k differ by k + 1 (mod 100), and element e + 1 of a `channels` + // vector reads one more than element e at the same sample. A waveform channel steps by 1 inside + // a datagram and by 2 - count between two. + if (up && !s.rows.empty() && flatudp::any_wave(s.rows)) { + const long wb = flatudp_wave_breaks(a, 0, static_cast(s.rows[0].count)); + std::fprintf(stderr, "[flatudp] continuity: wave_breaks=%ld ok=%d\n", wb, wb == 0 ? 1 : 0); + } else if (up && !s.rows.empty()) { + std::string out; + bool ok = true; + std::size_t base = 0; + bool elem_done = false; + for (std::size_t k = 0; k < s.rows.size(); ++k) { + const flatudp::Row& r = s.rows[k]; + const long b = flatudp_breaks(a, base, static_cast(k) + 1); + out += " " + r.name + (r.count > 1 ? "[0]" : "") + "_breaks=" + std::to_string(b); + ok = ok && b == 0; + if (r.count > 1 && !elem_done) { + elem_done = true; + const long ob = flatudp_offset_breaks(a, base, base + 1, 1); + out += " " + r.name + "[1]-[0]_breaks=" + std::to_string(ob); + ok = ok && ob == 0; + } + base += r.count; + } + std::fprintf(stderr, "[flatudp] continuity:%s ok=%d\n", out.c_str(), ok ? 1 : 0); } else { - std::fprintf(stderr, "[flatudp] continuity: skipped (needs a connected all-scalar list of two or more)\n"); + std::fprintf(stderr, "[flatudp] continuity: skipped (not connected)\n"); } const std::uint64_t wrong = src->wrong_size_datagrams(); diff --git a/tools/mock_flatudp.py b/tools/mock_flatudp.py index 332d78b..f292be0 100755 --- a/tools/mock_flatudp.py +++ b/tools/mock_flatudp.py @@ -7,7 +7,9 @@ One datagram is one cycle: the signals concatenated in declaration order, no pad chosen byte order. Packet i carries, for the signal of index k (from 0) and its element e, the value (i * (k + 1) + e) % 100, so consecutive datagrams are predictable. - --signals name:type[count],... default Trigger:uint8,Ia:float32,Cnt:uint64,Wave:int16[4] + --signals name:type[count][w],... default Trigger:uint8,Ia:float32,Cnt:uint64,Wave:int16[4] + (the datagram is the same with or without the app's trailing w, which only says + how the app shows a vector: one channel per element, or one waveform channel) --rate cycles per second --big-endian pack big endian (default little) --gap S,D send nothing from S to S+D seconds after the start (a sender that stops) @@ -32,7 +34,7 @@ def parse_signals(text): count = 1 if "[" in ty: ty, _, n = ty.partition("[") - count = int(n.rstrip("]")) + count = int(n.strip().rstrip("w").rstrip().rstrip("]")) # a trailing w (waveform) is the app's out.append((name.strip(), CODES[ty.strip()], count)) return out