fix(udpscope): bound accumulated-burst chaining so packet loss cannot displace the trace
Forward-chaining each accumulated burst onto the previous one suppresses arrival jitter, but an unchecked chain never recovers: one lost datagram, or a declared sampling rate that differs from the producer's real one, dates every later sample early for the rest of the run. The chain is now a prediction, compared each packet against the arrival anchor and abandoned beyond kBurstResyncThresholdS, which bounds the error instead of accumulating it. Plan amended so the hrt-fit fallback (unusable here: the fit needs 32 packets and is itself corrupted by bursty arrivals) cannot come back. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.6
parent
5a8479cda9
commit
892e3eae28
@@ -1,10 +1,25 @@
|
|||||||
#include "FrameDecoder.h"
|
#include "FrameDecoder.h"
|
||||||
|
|
||||||
|
#include <cmath>
|
||||||
|
|
||||||
namespace udpscope {
|
namespace udpscope {
|
||||||
|
|
||||||
/** Fallback cycle period before the first inter-packet gap is known. */
|
/** Fallback cycle period before the first inter-packet gap is known. */
|
||||||
static constexpr double kDefaultDt = 1.0e-3;
|
static constexpr double kDefaultDt = 1.0e-3;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* How far the forward-chained prediction for an accumulated burst may sit from
|
||||||
|
* where arrival time says it should be before the chain is abandoned.
|
||||||
|
*
|
||||||
|
* A kernel draining a backlog of queued datagrams can legitimately put the
|
||||||
|
* prediction a few hundred milliseconds ahead of arrival, so the threshold has
|
||||||
|
* to be well clear of that. Anything larger is not delivery jitter: it is lost
|
||||||
|
* packets or a declared sampling rate that does not match the producer's real
|
||||||
|
* one, and both must resynchronise rather than accumulate forever. Same value
|
||||||
|
* and same reasoning as ClockOffset::kRecalibThresholdS.
|
||||||
|
*/
|
||||||
|
static constexpr double kBurstResyncThresholdS = 0.5;
|
||||||
|
|
||||||
void FrameDecoder::setSignals(const std::vector<SignalMeta>& signals) {
|
void FrameDecoder::setSignals(const std::vector<SignalMeta>& signals) {
|
||||||
signals_ = signals;
|
signals_ = signals;
|
||||||
state_.assign(signals_.size(), SigState{});
|
state_.assign(signals_.size(), SigState{});
|
||||||
@@ -107,21 +122,25 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx,
|
|||||||
: 0.0;
|
: 0.0;
|
||||||
|
|
||||||
if (d.samplingRate > 0.0) {
|
if (d.samplingRate > 0.0) {
|
||||||
/* Forward-chain anchor: t[0] = lastEnd + dt, or wallNow on first
|
/* Where arrival time says this burst begins: its last element was
|
||||||
* packet (arrival-time for the very first burst only). */
|
* acquired just before the packet landed. */
|
||||||
double base;
|
const double arrivalAnchor =
|
||||||
|
wallNow - static_cast<double>(nElems - 1u) * dt;
|
||||||
|
|
||||||
|
/* Chaining from the end of the previous burst is immune to arrival
|
||||||
|
* jitter — a kernel draining several queued datagrams microseconds
|
||||||
|
* apart still yields contiguous timestamps. But a pure chain is
|
||||||
|
* blind: one lost datagram, or a declared rate that does not match
|
||||||
|
* the producer's real one, displaces every later sample and never
|
||||||
|
* recovers. So the chain is a PREDICTION, checked each packet
|
||||||
|
* against arrival and abandoned when the two disagree by more than
|
||||||
|
* a delivery backlog can explain. That bounds the error instead of
|
||||||
|
* letting it accumulate. */
|
||||||
|
double base = arrivalAnchor;
|
||||||
if (st.lastEmittedValid) {
|
if (st.lastEmittedValid) {
|
||||||
base = st.lastEmittedEnd + dt;
|
const double predicted = st.lastEmittedEnd + dt;
|
||||||
} else {
|
if (std::fabs(predicted - arrivalAnchor) <= kBurstResyncThresholdS) {
|
||||||
/* First packet: anchor element 0 at arrival time.
|
base = predicted;
|
||||||
* This one packet may be slightly off, but subsequent packets
|
|
||||||
* chain from this end and jitter is suppressed thereafter. */
|
|
||||||
base = wallNow - static_cast<double>(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);
|
tsOut.resize(nElems);
|
||||||
|
|||||||
@@ -54,8 +54,11 @@ private:
|
|||||||
bool lastAccValid = false;
|
bool lastAccValid = false;
|
||||||
uint32_t prevAccCount = 0;
|
uint32_t prevAccCount = 0;
|
||||||
/** For accumulated scalars with a declared sampling rate: end timestamp
|
/** For accumulated scalars with a declared sampling rate: end timestamp
|
||||||
* of the most recently emitted burst, used as a forward-chain anchor
|
* of the most recently emitted burst. The next burst is PREDICTED to
|
||||||
* that is immune to arrival-time jitter. */
|
* start one sample period after it — immune to arrival-time jitter —
|
||||||
|
* but the prediction is discarded when arrival time disagrees with it
|
||||||
|
* by more than a delivery backlog can explain, so packet loss cannot
|
||||||
|
* displace the trace permanently. */
|
||||||
double lastEmittedEnd = 0.0;
|
double lastEmittedEnd = 0.0;
|
||||||
bool lastEmittedValid = false;
|
bool lastEmittedValid = false;
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -177,6 +177,45 @@ TEST(FrameDecoder, AccumulatedScalarSurvivesBurstyDelivery) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// The counterweight to the test above. Suppressing arrival jitter by chaining
|
||||||
|
// each burst onto the previous one is only safe while the chain is checked: on
|
||||||
|
// UDP, packets are lost, and a chain that ignores arrival entirely closes the
|
||||||
|
// hole silently and dates every later sample a full second early — for the rest
|
||||||
|
// of the run, because nothing ever pulls it back. The prediction has to be
|
||||||
|
// abandoned once arrival contradicts it by more than a delivery backlog could.
|
||||||
|
TEST(FrameDecoder, AccumulatedScalarResynchronisesAfterLostPackets) {
|
||||||
|
FrameDecoder dec;
|
||||||
|
SignalMeta m;
|
||||||
|
m.name = "Acc";
|
||||||
|
m.typeCode = 9;
|
||||||
|
m.numRows = 1;
|
||||||
|
m.samplingRate = 1000.0; /* 10 samples = 10 ms per packet */
|
||||||
|
dec.setSignals({m});
|
||||||
|
|
||||||
|
std::vector<double> ts;
|
||||||
|
for (int p = 0; p < 10; p++) {
|
||||||
|
FrameBuilder fb;
|
||||||
|
fb.addSignal(std::vector<double>(10, 1.0));
|
||||||
|
const FrameView& f = fb.build(0, 500.0 + p * 0.010, 10);
|
||||||
|
dec.beginFrame(f);
|
||||||
|
ASSERT_TRUE(dec.timestamps(f, 0, ts));
|
||||||
|
}
|
||||||
|
/* Contiguous so far: burst 9 ends at 500.090. */
|
||||||
|
EXPECT_NEAR(ts[9], 500.090, 1e-9);
|
||||||
|
|
||||||
|
/* A full second of packets never arrives. The next one lands at 501.100. */
|
||||||
|
FrameBuilder fb;
|
||||||
|
fb.addSignal(std::vector<double>(10, 1.0));
|
||||||
|
const FrameView& f = fb.build(0, 501.100, 10);
|
||||||
|
dec.beginFrame(f);
|
||||||
|
ASSERT_TRUE(dec.timestamps(f, 0, ts));
|
||||||
|
|
||||||
|
/* Chaining blindly would put this burst at 500.091..500.100, overlapping
|
||||||
|
* the gap as though no data were missing. */
|
||||||
|
EXPECT_NEAR(ts[0], 501.091, 1e-9);
|
||||||
|
EXPECT_NEAR(ts[9], 501.100, 1e-9);
|
||||||
|
}
|
||||||
|
|
||||||
TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) {
|
TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) {
|
||||||
FrameDecoder dec;
|
FrameDecoder dec;
|
||||||
SignalMeta m;
|
SignalMeta m;
|
||||||
|
|||||||
@@ -1609,6 +1609,42 @@ TEST(FrameDecoder, AccumulatedScalarSurvivesBurstyDelivery) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ADDED in Task 4 review. The counterweight to the test above: suppressing
|
||||||
|
// arrival jitter by chaining bursts is only safe while the chain is CHECKED. On
|
||||||
|
// UDP packets are lost, and an unchecked chain closes the hole silently and
|
||||||
|
// dates every later sample early for the rest of the run.
|
||||||
|
TEST(FrameDecoder, AccumulatedScalarResynchronisesAfterLostPackets) {
|
||||||
|
FrameDecoder dec;
|
||||||
|
SignalMeta m;
|
||||||
|
m.name = "Acc";
|
||||||
|
m.typeCode = 9;
|
||||||
|
m.numRows = 1;
|
||||||
|
m.samplingRate = 1000.0; /* 10 samples = 10 ms per packet */
|
||||||
|
dec.setSignals({m});
|
||||||
|
|
||||||
|
std::vector<double> ts;
|
||||||
|
for (int p = 0; p < 10; p++) {
|
||||||
|
FrameBuilder fb;
|
||||||
|
fb.addSignal(std::vector<double>(10, 1.0));
|
||||||
|
const FrameView& f = fb.build(0, 500.0 + p * 0.010, 10);
|
||||||
|
dec.beginFrame(f);
|
||||||
|
ASSERT_TRUE(dec.timestamps(f, 0, ts));
|
||||||
|
}
|
||||||
|
EXPECT_NEAR(ts[9], 500.090, 1e-9);
|
||||||
|
|
||||||
|
/* A full second of packets never arrives. The next one lands at 501.100. */
|
||||||
|
FrameBuilder fb;
|
||||||
|
fb.addSignal(std::vector<double>(10, 1.0));
|
||||||
|
const FrameView& f = fb.build(0, 501.100, 10);
|
||||||
|
dec.beginFrame(f);
|
||||||
|
ASSERT_TRUE(dec.timestamps(f, 0, ts));
|
||||||
|
|
||||||
|
/* Chaining blindly would put this burst at 500.091..500.100, as though no
|
||||||
|
* data were missing. */
|
||||||
|
EXPECT_NEAR(ts[0], 501.091, 1e-9);
|
||||||
|
EXPECT_NEAR(ts[9], 501.100, 1e-9);
|
||||||
|
}
|
||||||
|
|
||||||
TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) {
|
TEST(FrameDecoder, AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared) {
|
||||||
FrameDecoder dec;
|
FrameDecoder dec;
|
||||||
SignalMeta m;
|
SignalMeta m;
|
||||||
@@ -1882,8 +1918,44 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx,
|
|||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
/* Rule 3: accumulated scalar, based on the producer's own hrt. */
|
/* Rule 3: accumulated scalar.
|
||||||
|
*
|
||||||
|
* AMENDED after Task 4 review. The version below originally sent EVERY
|
||||||
|
* accumulated scalar through the hrt fit, falling back to packetBurst until
|
||||||
|
* the fit was ready. That cannot work when a declared samplingRate is
|
||||||
|
* present: HrtRateFit needs 32 packets, bursty delivery can begin before
|
||||||
|
* that, and packetBurst then crams a 10 ms burst into a 50 us arrival gap —
|
||||||
|
* exactly the sawtooth this rule exists to prevent. Worse, HrtRateFit fits
|
||||||
|
* hrt against ARRIVAL time, so a burst episode corrupts the very rate the
|
||||||
|
* fallback is waiting on.
|
||||||
|
*
|
||||||
|
* With a declared rate none of that is needed: the intra-packet step is
|
||||||
|
* exact, and bursts are contiguous, so the next burst is PREDICTED at
|
||||||
|
* lastEmittedEnd + dt. The prediction must be checked, not trusted — a pure
|
||||||
|
* chain silently closes the hole left by a lost datagram and dates every
|
||||||
|
* later sample early for the rest of the run. So each packet compares the
|
||||||
|
* prediction against the arrival anchor and abandons it beyond
|
||||||
|
* kBurstResyncThresholdS. The hrt path below remains for samplingRate == 0. */
|
||||||
if (d.numElements() == 1u && nElems > 1u) {
|
if (d.numElements() == 1u && nElems > 1u) {
|
||||||
|
const double dtDeclared = (d.samplingRate > 0.0) ? (1.0 / d.samplingRate) : 0.0;
|
||||||
|
if (d.samplingRate > 0.0) {
|
||||||
|
const double arrivalAnchor =
|
||||||
|
wallNow - static_cast<double>(nElems - 1u) * dtDeclared;
|
||||||
|
double base = arrivalAnchor;
|
||||||
|
if (st.lastEmittedValid) {
|
||||||
|
const double predicted = st.lastEmittedEnd + dtDeclared;
|
||||||
|
if (std::fabs(predicted - arrivalAnchor) <= kBurstResyncThresholdS) {
|
||||||
|
base = predicted;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
tsOut.resize(nElems);
|
||||||
|
for (uint32_t e = 0; e < nElems; e++) {
|
||||||
|
tsOut[e] = base + static_cast<double>(e) * dtDeclared;
|
||||||
|
}
|
||||||
|
st.lastEmittedEnd = tsOut[nElems - 1u];
|
||||||
|
st.lastEmittedValid = true;
|
||||||
|
return true;
|
||||||
|
}
|
||||||
if (!hrtFit_.ready()) {
|
if (!hrtFit_.ready()) {
|
||||||
return packetBurst(idx, nElems, wallNow, tsOut);
|
return packetBurst(idx, nElems, wallNow, tsOut);
|
||||||
}
|
}
|
||||||
@@ -1891,9 +1963,7 @@ bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx,
|
|||||||
const double base = st.offset.map(hrtSec, wallNow);
|
const double base = st.offset.map(hrtSec, wallNow);
|
||||||
|
|
||||||
double dt;
|
double dt;
|
||||||
if (d.samplingRate > 0.0) {
|
if (st.lastAccValid && st.prevAccCount > 0u &&
|
||||||
dt = 1.0 / d.samplingRate;
|
|
||||||
} else if (st.lastAccValid && st.prevAccCount > 0u &&
|
|
||||||
hrtSec > st.lastAccHrtSec) {
|
hrtSec > st.lastAccHrtSec) {
|
||||||
/* The flushes carry contiguous RT cycles, so the gap divided by the
|
/* The flushes carry contiguous RT cycles, so the gap divided by the
|
||||||
* previous packet's sample count is exactly one cycle period. */
|
* previous packet's sample count is exactly one cycle period. */
|
||||||
@@ -1945,7 +2015,7 @@ cd Client/udpscope && cmake --build build -j && ./build/udpscope_tests --gtest_f
|
|||||||
|
|
||||||
Expected: PASS, 9 tests.
|
Expected: PASS, 9 tests.
|
||||||
|
|
||||||
If `AccumulatedScalarSurvivesBurstyDelivery` fails on the first few samples, check that `beginFrame()` is being called before `timestamps()` — the hrt fit needs 32 packets before rule 3 engages, and the packets before that legitimately go through rule 4.
|
If `AccumulatedScalarSurvivesBurstyDelivery` fails, do NOT reach for the hrt fit: with a declared `samplingRate` rule 3 never consults it, precisely because the fit is not ready for the first 32 packets and is itself corrupted by bursty arrivals. Check instead that `lastEmittedEnd`/`lastEmittedValid` are being updated on every emitted burst. The only test that may legitimately fall through to rule 4 early is `AccumulatedScalarDerivesDtFromTheHrtGapWhenNoRateIsDeclared`, whose arrivals are uniform, so `packetBurst` is accurate there.
|
||||||
|
|
||||||
- [ ] **Step 8: Commit**
|
- [ ] **Step 8: Commit**
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user