diff --git a/Client/udpscope/CMakeLists.txt b/Client/udpscope/CMakeLists.txt index ba24ee8..32ff7bb 100644 --- a/Client/udpscope/CMakeLists.txt +++ b/Client/udpscope/CMakeLists.txt @@ -83,6 +83,7 @@ set(CORE_SOURCES Decimate.cpp PaneTree.cpp TimeBase.cpp + FrameDecoder.cpp ) add_library(udpscope_core STATIC ${CORE_SOURCES}) diff --git a/Client/udpscope/FrameDecoder.cpp b/Client/udpscope/FrameDecoder.cpp new file mode 100644 index 0000000..a9afbdf --- /dev/null +++ b/Client/udpscope/FrameDecoder.cpp @@ -0,0 +1,173 @@ +#include "FrameDecoder.h" + +namespace udpscope { + +/** Fallback cycle period before the first inter-packet gap is known. */ +static constexpr double kDefaultDt = 1.0e-3; + +void FrameDecoder::setSignals(const std::vector& signals) { + signals_ = signals; + state_.assign(signals_.size(), SigState{}); + hrtFit_.reset(); +} + +void FrameDecoder::reset() { + state_.assign(signals_.size(), SigState{}); + hrtFit_.reset(); +} + +void FrameDecoder::beginFrame(const FrameView& f) { + if (f.hrt != 0u) { hrtFit_.add(f.hrt, f.recvTime); } +} + +bool FrameDecoder::packetBurst(uint32_t idx, uint32_t nElems, double wallNow, + std::vector& tsOut) { + SigState& st = state_[idx]; + if (!st.lastPacketValid || wallNow <= st.lastPacketWall) { + /* No previous arrival to span from, or time went backwards. Remember + * this one and drop the samples rather than store them at made-up + * spacing. */ + st.lastPacketWall = wallNow; + st.lastPacketValid = true; + return false; + } + + const double dt = (wallNow - st.lastPacketWall) / static_cast(nElems); + tsOut.resize(nElems); + for (uint32_t e = 0; e < nElems; e++) { + tsOut[e] = st.lastPacketWall + static_cast(e + 1u) * dt; + } + st.lastPacketWall = wallNow; + return true; +} + +bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx, + std::vector& tsOut) { + tsOut.clear(); + if (idx >= signals_.size() || idx >= f.numSignals || f.counts == nullptr) { + return false; + } + + const SignalMeta& d = signals_[idx]; + const uint32_t nElems = f.counts[idx]; + if (nElems == 0u) { return false; } + + const double wallNow = f.recvTime; + SigState& st = state_[idx]; + + const bool hasTimeSig = d.hasTimeSignal(f.numSignals); + const uint32_t tIdx = hasTimeSig ? d.timeSignalIdx : 0u; + const double tScale = hasTimeSig + ? TimeSignalScale(signals_[tIdx].typeCode) + : 1.0e-6; + + /* Rule 1: one stamp per element, straight from the time signal. */ + if (d.timeMode == kTimeFullArray && hasTimeSig && + f.counts[tIdx] >= nElems && f.values[tIdx] != nullptr) { + const double* tv = f.values[tIdx]; + const double t0 = tv[0] * tScale; + (void) st.offset.map(t0, wallNow); + const double base = st.offset.offset(); + tsOut.resize(nElems); + for (uint32_t e = 0; e < nElems; e++) { + tsOut[e] = base + tv[e] * tScale; + } + return true; + } + + /* Rule 2: anchor from the time signal, spread by the sampling rate. */ + if ((d.timeMode == kTimeFirstSample || d.timeMode == kTimeLastSample) && + hasTimeSig && f.counts[tIdx] >= 1u && f.values[tIdx] != nullptr) { + const double anchor = st.offset.map(f.values[tIdx][0] * tScale, wallNow); + const double dt = (d.samplingRate > 0.0) ? (1.0 / d.samplingRate) : 0.0; + tsOut.resize(nElems); + for (uint32_t e = 0; e < nElems; e++) { + tsOut[e] = (d.timeMode == kTimeFirstSample) + ? (anchor + static_cast(e) * dt) + : (anchor - static_cast(nElems - 1u - e) * dt); + } + return true; + } + + /* Rule 3: accumulated scalar, based on declared sampling rate or hrt. + * + * When samplingRate is declared the inter-element step is exact and we + * anchor from the end of the previous burst rather than from arrival time + * or hrt. This makes the output immune to arrival jitter: even when the + * kernel delivers two packets microseconds apart each burst starts exactly + * one sample period after the previous burst ended. + * + * When samplingRate is absent we must derive dt from the hrt gap, which + * requires the HrtRateFit to be ready. Until then we fall back to + * packetBurst (arrival-time spanning), which is accurate during the normal + * pre-burst delivery phase that precedes the fit becoming ready. */ + if (d.numElements() == 1u && nElems > 1u) { + const double dt = (d.samplingRate > 0.0) + ? (1.0 / d.samplingRate) + : 0.0; + + if (d.samplingRate > 0.0) { + /* Forward-chain anchor: t[0] = lastEnd + dt, or wallNow on first + * packet (arrival-time for the very first burst only). */ + double base; + if (st.lastEmittedValid) { + base = st.lastEmittedEnd + dt; + } else { + /* First packet: anchor element 0 at arrival time. + * This one packet may be slightly off, but subsequent packets + * chain from this end and jitter is suppressed thereafter. */ + base = wallNow - static_cast(nElems - 1u) * dt; + /* Calibrate the clock-offset so later hrt-based paths (if any) + * are consistent, but we don't use it in this branch. */ + if (f.hrt != 0u && hrtFit_.ready()) { + const double hrtSec = hrtFit_.toSeconds(f.hrt); + (void) st.offset.map(hrtSec, wallNow); + } + } + tsOut.resize(nElems); + for (uint32_t e = 0; e < nElems; e++) { + tsOut[e] = base + static_cast(e) * dt; + } + st.lastEmittedEnd = tsOut[nElems - 1u]; + st.lastEmittedValid = true; + return true; + } + + /* No declared rate: need hrt-derived dt. */ + if (!hrtFit_.ready()) { + return packetBurst(idx, nElems, wallNow, tsOut); + } + const double hrtSec = hrtFit_.toSeconds(f.hrt); + const double base = st.offset.map(hrtSec, wallNow); + + double hrtDt; + if (st.lastAccValid && st.prevAccCount > 0u && hrtSec > st.lastAccHrtSec) { + /* The flushes carry contiguous RT cycles, so the gap divided by the + * previous packet's sample count is exactly one cycle period. */ + hrtDt = (hrtSec - st.lastAccHrtSec) / + static_cast(st.prevAccCount); + } else { + hrtDt = kDefaultDt; + } + + tsOut.resize(nElems); + for (uint32_t e = 0; e < nElems; e++) { + tsOut[e] = base + static_cast(e) * hrtDt; + } + st.lastAccHrtSec = hrtSec; + st.lastAccValid = true; + st.prevAccCount = nElems; + return true; + } + + /* Rule 4: PACKET burst with no time reference at all. */ + if (nElems > 1u) { + return packetBurst(idx, nElems, wallNow, tsOut); + } + + /* Rule 5: plain scalar. */ + tsOut.assign(1, wallNow); + return true; +} + +} /* namespace udpscope */ diff --git a/Client/udpscope/FrameDecoder.h b/Client/udpscope/FrameDecoder.h new file mode 100644 index 0000000..43e847e --- /dev/null +++ b/Client/udpscope/FrameDecoder.h @@ -0,0 +1,68 @@ +/** + * @file FrameDecoder.h + * @brief Per-element timestamp reconstruction for UDPS frames. + * + * The C client's udps_frame_element_time() is explicitly an arrival-anchored + * estimate. It is not sufficient: the kernel frequently delivers several queued + * datagrams in one burst, so two packets are processed microseconds apart even + * though each represents ~10 ms of signal, and arrival-time interpolation then + * crams a packet's samples into that tiny gap — the trace renders as a sawtooth. + * Source/Applications/StreamHub/UDPSourceSession.cpp documents this failure and + * solves it; these are the same rules, computed from udps_frame_t's own fields + * so the scope and StreamHub agree on the same stream. + */ +#pragma once + +#include "TimeBase.h" +#include "Types.h" + +#include + +namespace udpscope { + +class FrameDecoder { +public: + /** Installs the signal table. Clears all per-signal timing history. */ + void setSignals(const std::vector& signals); + + const std::vector& signals() const { return signals_; } + + /** Call once per frame, before any timestamps() call for that frame. */ + void beginFrame(const FrameView& f); + + /** + * @brief Timestamps for every value of signal @p idx in this frame. + * @return false when the signal produced nothing usable — an empty slot, or + * the first PACKET burst after connect, which has no previous + * arrival to span from and would otherwise poison the ring with + * wrongly spaced timestamps. + */ + bool timestamps(const FrameView& f, uint32_t idx, std::vector& tsOut); + + /** Forgets all timing history; call on reconnect. */ + void reset(); + +private: + bool packetBurst(uint32_t idx, uint32_t nElems, double wallNow, + std::vector& tsOut); + + struct SigState { + ClockOffset offset; + double lastPacketWall = 0.0; + bool lastPacketValid = false; + double lastAccHrtSec = 0.0; + bool lastAccValid = false; + uint32_t prevAccCount = 0; + /** For accumulated scalars with a declared sampling rate: end timestamp + * of the most recently emitted burst, used as a forward-chain anchor + * that is immune to arrival-time jitter. */ + double lastEmittedEnd = 0.0; + bool lastEmittedValid = false; + }; + + std::vector signals_; + std::vector state_; + HrtRateFit hrtFit_; +}; + +} /* namespace udpscope */ diff --git a/Client/udpscope/Types.h b/Client/udpscope/Types.h index 878f640..07805fc 100644 --- a/Client/udpscope/Types.h +++ b/Client/udpscope/Types.h @@ -35,4 +35,69 @@ struct Rect { } }; +/* Protocol constants, spelled out rather than included, so the framework-free + * modules stay independent of udps_client.h. They mirror Common/UDP/UDPSProtocol.h. */ +constexpr uint8_t kTimePacket = 0; +constexpr uint8_t kTimeFullArray = 1; +constexpr uint8_t kTimeFirstSample = 2; +constexpr uint8_t kTimeLastSample = 3; +constexpr uint32_t kNoTimeSignal = 0xFFFFFFFFu; + +/** Framework-free mirror of udps_signal_t, plus UI state. */ +struct SignalMeta { + std::string name; + uint8_t typeCode = 255; + uint8_t quantType = 0; + uint32_t numRows = 1; + uint32_t numCols = 1; + double rangeMin = 0.0; + double rangeMax = 0.0; + uint8_t timeMode = kTimePacket; + double samplingRate = 0.0; + uint32_t timeSignalIdx = kNoTimeSignal; + std::string unit; + + /** User override: treat an ambiguous PACKET array as a profile, not a burst. */ + bool profileOverride = false; + + uint32_t numElements() const { + const uint64_t n = static_cast(numRows ? numRows : 1u) * + static_cast(numCols ? numCols : 1u); + return n == 0u ? 1u : static_cast(n); + } + + bool hasTimeSignal(uint32_t numSignals) const { + return timeSignalIdx != kNoTimeSignal && timeSignalIdx < numSignals; + } + + /** + * @brief True when this array should be plotted against element index + * rather than unrolled onto the time axis. + * + * Only PACKET arrays are ambiguous: the producer stamped the whole datagram + * with one time, which is what a genuine vector looks like and also what a + * burst carrying no time metadata looks like. Default is burst, matching + * UDPSourceSession, with this flag as the user's override. + */ + bool isVectorProfile() const { + return profileOverride && numElements() > 1u && timeMode == kTimePacket; + } +}; + +/** + * @brief Non-owning mirror of udps_frame_t. + * + * Kept separate from the C struct so FrameDecoder can be tested with plain + * arrays and no socket. Points at memory owned by the caller. + */ +struct FrameView { + uint32_t counter = 0; + uint64_t hrt = 0; + double recvTime = 0.0; + uint32_t numSamples = 1; + uint32_t numSignals = 0; + const double* const* values = nullptr; /**< values[i][0..counts[i]) */ + const uint32_t* counts = nullptr; +}; + } /* namespace udpscope */ diff --git a/Client/udpscope/tests/FrameDecoderTest.cpp b/Client/udpscope/tests/FrameDecoderTest.cpp new file mode 100644 index 0000000..e923d12 --- /dev/null +++ b/Client/udpscope/tests/FrameDecoderTest.cpp @@ -0,0 +1,276 @@ +#include "FrameDecoder.h" + +#include + +#include + +using namespace udpscope; + +namespace { + +/** Builds a FrameView over vectors the test owns. */ +struct FrameBuilder { + std::vector> storage; + std::vector ptrs; + std::vector counts; + FrameView view; + + void addSignal(std::vector vals) { + storage.push_back(std::move(vals)); + } + + const FrameView& build(uint64_t hrt, double recvTime, uint32_t numSamples = 1) { + ptrs.clear(); + counts.clear(); + for (const auto& s : storage) { + ptrs.push_back(s.data()); + counts.push_back(static_cast(s.size())); + } + view.hrt = hrt; + view.recvTime = recvTime; + view.numSamples = numSamples; + view.numSignals = static_cast(storage.size()); + view.values = ptrs.data(); + view.counts = counts.data(); + return view; + } +}; + +SignalMeta burst(const char* name, uint8_t timeMode, double rate, + uint32_t elems, uint32_t timeIdx) { + SignalMeta m; + m.name = name; + m.typeCode = 8; /* float32 */ + m.numRows = elems; + m.numCols = 1; + m.timeMode = timeMode; + m.samplingRate = rate; + m.timeSignalIdx = timeIdx; + return m; +} + +SignalMeta timeSignal(const char* name, uint32_t elems) { + SignalMeta m; + m.name = name; + m.typeCode = 6; /* uint64 -> nanoseconds */ + m.numRows = elems; + m.numCols = 1; + return m; +} + +} /* namespace */ + +TEST(FrameDecoder, FullArrayTakesOneStampPerElementFromTheTimeSignal) { + FrameDecoder dec; + dec.setSignals({burst("Sine", kTimeFullArray, 1000.0, 4, 1), + timeSignal("Time", 4)}); + + FrameBuilder fb; + fb.addSignal({1.0, 2.0, 3.0, 4.0}); + /* Nanoseconds: 5.000, 5.001, 5.002, 5.003 s of producer time. */ + fb.addSignal({5.0e9, 5.001e9, 5.002e9, 5.003e9}); + const FrameView& f = fb.build(0, 1000.0); + + dec.beginFrame(f); + std::vector ts; + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + ASSERT_EQ(ts.size(), 4u); + + /* Element 0 lands on the arrival time; the rest keep the producer spacing. */ + EXPECT_NEAR(ts[0], 1000.000, 1e-9); + EXPECT_NEAR(ts[1], 1000.001, 1e-9); + EXPECT_NEAR(ts[2], 1000.002, 1e-9); + EXPECT_NEAR(ts[3], 1000.003, 1e-9); +} + +TEST(FrameDecoder, FirstSampleAnchorsElementZeroAndCountsForward) { + FrameDecoder dec; + dec.setSignals({burst("Sine", kTimeFirstSample, 1000.0, 4, 1), + timeSignal("Time", 1)}); + + FrameBuilder fb; + fb.addSignal({1.0, 2.0, 3.0, 4.0}); + fb.addSignal({7.0e9}); + const FrameView& f = fb.build(0, 2000.0); + + dec.beginFrame(f); + std::vector ts; + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + ASSERT_EQ(ts.size(), 4u); + EXPECT_NEAR(ts[0], 2000.000, 1e-9); + EXPECT_NEAR(ts[3], 2000.003, 1e-9); +} + +TEST(FrameDecoder, LastSampleAnchorsTheFinalElementAndCountsBackward) { + FrameDecoder dec; + dec.setSignals({burst("Sine", kTimeLastSample, 1000.0, 4, 1), + timeSignal("Time", 1)}); + + FrameBuilder fb; + fb.addSignal({1.0, 2.0, 3.0, 4.0}); + fb.addSignal({7.0e9}); + const FrameView& f = fb.build(0, 3000.0); + + dec.beginFrame(f); + std::vector ts; + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + ASSERT_EQ(ts.size(), 4u); + EXPECT_NEAR(ts[3], 3000.000, 1e-9); + EXPECT_NEAR(ts[0], 3000.000 - 0.003, 1e-9); +} + +TEST(FrameDecoder, PlainScalarUsesArrivalTime) { + FrameDecoder dec; + SignalMeta m; + m.name = "Level"; + m.typeCode = 9; + dec.setSignals({m}); + + FrameBuilder fb; + fb.addSignal({42.0}); + const FrameView& f = fb.build(0, 1234.5); + + dec.beginFrame(f); + std::vector ts; + ASSERT_TRUE(dec.timestamps(f, 0, ts)); + ASSERT_EQ(ts.size(), 1u); + EXPECT_DOUBLE_EQ(ts[0], 1234.5); +} + +// This is the failure UDPSourceSession.cpp:560 documents. The kernel delivers +// two queued datagrams microseconds apart even though each carries 10 ms of +// signal. Dating from arrival crams the second packet's samples into that gap +// and the trace becomes a sawtooth; dating from the producer hrt does not. +TEST(FrameDecoder, AccumulatedScalarSurvivesBurstyDelivery) { + FrameDecoder dec; + SignalMeta m; + m.name = "Acc"; + m.typeCode = 9; + m.numRows = 1; + m.samplingRate = 1000.0; /* 1 kHz, 10 samples = 10 ms per packet */ + dec.setSignals({m}); + + const double ticks = 1.0e9; + std::vector all; + + for (int p = 0; p < 40; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(10, static_cast(p))); + const double producerSec = 100.0 + p * 0.010; + /* Packets 20+ arrive in a burst, all within 50 us of each other. */ + const double arrival = (p < 20) ? (500.0 + p * 0.010) + : (500.2 + (p - 20) * 0.00005); + const FrameView& f = fb.build(static_cast(producerSec * ticks), + arrival, 10); + dec.beginFrame(f); + std::vector ts; + if (dec.timestamps(f, 0, ts)) { + all.insert(all.end(), ts.begin(), ts.end()); + } + } + + ASSERT_GT(all.size(), 300u); + for (size_t i = 1; i < all.size(); i++) { + EXPECT_GT(all[i], all[i - 1]) << "non-monotonic at " << i; + EXPECT_NEAR(all[i] - all[i - 1], 0.001, 2e-4) + << "spacing collapsed at " << i << " (sawtooth)"; + } +} + +TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) { + FrameDecoder dec; + SignalMeta m; + m.name = "Acc"; + m.typeCode = 9; + m.samplingRate = 0.0; /* undeclared */ + dec.setSignals({m}); + + const double ticks = 1.0e9; + std::vector last; + for (int p = 0; p < 40; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(10, 1.0)); + const double producerSec = 100.0 + p * 0.010; /* 10 ms per packet */ + const FrameView& f = fb.build(static_cast(producerSec * ticks), + 700.0 + p * 0.010, 10); + dec.beginFrame(f); + std::vector ts; + if (dec.timestamps(f, 0, ts)) { last = ts; } + } + + ASSERT_EQ(last.size(), 10u); + /* 10 ms of producer time across 10 samples is a 1 ms period. */ + EXPECT_NEAR(last[1] - last[0], 0.001, 1e-5); +} + +// A PACKET burst has no per-element time at all. Elements span +// (lastPacket, thisPacket] — backwards from arrival, because the samples were +// acquired before the packet landed. Forward extrapolation would let a jittered +// packet overlap the next one and break ring monotonicity. +TEST(FrameDecoder, PacketBurstDropsTheFirstFrameThenSpansBackwards) { + FrameDecoder dec; + dec.setSignals({burst("Raw", kTimePacket, 0.0, 5, kNoTimeSignal)}); + + FrameBuilder fb1; + fb1.addSignal({1.0, 2.0, 3.0, 4.0, 5.0}); + const FrameView& f1 = fb1.build(0, 10.0); + dec.beginFrame(f1); + std::vector ts; + EXPECT_FALSE(dec.timestamps(f1, 0, ts)) + << "the first packet has no previous arrival to span from"; + + FrameBuilder fb2; + fb2.addSignal({6.0, 7.0, 8.0, 9.0, 10.0}); + const FrameView& f2 = fb2.build(0, 10.05); + dec.beginFrame(f2); + ASSERT_TRUE(dec.timestamps(f2, 0, ts)); + ASSERT_EQ(ts.size(), 5u); + EXPECT_GT(ts[0], 10.0); + EXPECT_NEAR(ts[4], 10.05, 1e-12); + EXPECT_NEAR(ts[1] - ts[0], 0.01, 1e-12); +} + +TEST(FrameDecoder, PacketBurstStaysMonotonicUnderJitteredArrivals) { + FrameDecoder dec; + dec.setSignals({burst("Raw", kTimePacket, 0.0, 8, kNoTimeSignal)}); + + const double jitter[] = {0.0, 0.004, -0.003, 0.006, -0.002, 0.0, 0.005, -0.004}; + std::vector all; + for (int p = 0; p < 8; p++) { + FrameBuilder fb; + fb.addSignal(std::vector(8, 1.0)); + const FrameView& f = fb.build(0, 20.0 + p * 0.05 + jitter[p]); + dec.beginFrame(f); + std::vector ts; + if (dec.timestamps(f, 0, ts)) { + all.insert(all.end(), ts.begin(), ts.end()); + } + } + + ASSERT_GT(all.size(), 8u); + for (size_t i = 1; i < all.size(); i++) { + EXPECT_GT(all[i], all[i - 1]) << "packets overlapped at " << i; + } +} + +TEST(FrameDecoder, ResetForgetsPerSignalHistory) { + FrameDecoder dec; + dec.setSignals({burst("Raw", kTimePacket, 0.0, 4, kNoTimeSignal)}); + + FrameBuilder fb; + fb.addSignal({1.0, 2.0, 3.0, 4.0}); + const FrameView& f = fb.build(0, 5.0); + dec.beginFrame(f); + std::vector ts; + EXPECT_FALSE(dec.timestamps(f, 0, ts)); + + const FrameView& f2 = fb.build(0, 5.1); + dec.beginFrame(f2); + EXPECT_TRUE(dec.timestamps(f2, 0, ts)); + + dec.reset(); + const FrameView& f3 = fb.build(0, 5.2); + dec.beginFrame(f3); + EXPECT_FALSE(dec.timestamps(f3, 0, ts)) + << "after reset the next packet is again the first one"; +}