Martino FerrariandClaude Opus 4.6
1c61e814c0
The counter is the denominator of the very period lastAccHrt is the numerator of, so any packet that cannot move the tick reference must not move the counter either. Two paths were violating that: a reordered datagram rolled the counter back while the reference correctly held (next burst drawn 0.048x too narrow at distance 20, 83.3% worst spacing error under 2% sustained reordering), and a stray hrt == 0 packet advanced the counter from the warm-up branch without a tick to match (+22.5 ms of future-dating per stray packet). Rules 1 and 2 now record a counter too. The duplicate-datagram guard is keyed on one, so an array rule that recorded none was exempt and plotted every doubly-delivered update twice. Also: rule 2 divides by the count the anchor actually spans and falls back to the last period it derived; the counter-gap test is wrap-safe so 2^32 rollover reads as no information rather than 2e9 lost packets. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
678 lines
37 KiB
C++
678 lines
37 KiB
C++
#include "FrameDecoder.h"
|
|
|
|
#include <cmath>
|
|
|
|
namespace udpscope {
|
|
|
|
/** Fallback cycle period before the first inter-packet gap is known. */
|
|
static constexpr double kDefaultDt = 1.0e-3;
|
|
|
|
/**
|
|
* How far a chained burst prediction may sit from where arrival time says it
|
|
* should be before the chain is abandoned and time is re-anchored on arrival.
|
|
*
|
|
* This is a backstop, not the primary mechanism: the packet counter normally
|
|
* accounts for lost datagrams exactly, so the prediction and arrival agree.
|
|
* It catches what the counter cannot describe — a producer restart (the
|
|
* counter returns to zero), a counter that never advances, and a declared
|
|
* sampling rate that does not match the producer's real one. A kernel draining
|
|
* a backlog of queued datagrams can legitimately put the prediction a couple of
|
|
* hundred milliseconds from arrival, so the threshold sits well clear of that.
|
|
* Same value and same reasoning as ClockOffset::kRecalibThresholdS.
|
|
*/
|
|
static constexpr double kBurstResyncThresholdS = 0.5;
|
|
|
|
/**
|
|
* Narrowest a burst may be drawn, as a fraction of its nominal width, while a
|
|
* leading timeline is being pulled back. Only a floor: the squeeze is normally
|
|
* proportional to the excess and removes it in a single burst. See the sole use
|
|
* site.
|
|
*/
|
|
static constexpr double kMinBleedFactor = 0.05;
|
|
|
|
/**
|
|
* Largest share of the wall time elapsed since a signal's previous burst that
|
|
* that signal's next burst may advance its own timeline by, while a lead is
|
|
* being pulled back.
|
|
*
|
|
* This, and not kMinBleedFactor, is what makes a lead converge. A fraction of
|
|
* the NOMINAL burst width cannot: the floor still advances the timeline by
|
|
* kMinBleedFactor * nominal per packet while the wall advances one packet
|
|
* interval, so it diverges outright whenever nominal exceeds
|
|
* (1 / kMinBleedFactor) packet intervals — a declared SamplingRate of 30 against
|
|
* a producer really flushing 10 samples at 1 kHz put the trace 667 s ahead after
|
|
* 1000 s of stream. Measuring the allowance against elapsed WALL time instead
|
|
* bounds the advance below the wall's own advance for any declared rate, so the
|
|
* lead strictly falls whatever the config says. Any fraction under 1 converges;
|
|
* a half both converges quickly and leaves the burst visibly compressed rather
|
|
* than frozen.
|
|
*/
|
|
static constexpr double kWallBleedFraction = 0.5;
|
|
|
|
/**
|
|
* A backward jump in producer hrt larger than this is a producer RESTART; a
|
|
* smaller one is a reordered datagram.
|
|
*
|
|
* The two need opposite handling — a reorder must leave the hrt reference
|
|
* untouched (its interval was already counted by the packet that overtook it),
|
|
* a restart must rebase onto the new epoch or the signal never advances again —
|
|
* and nothing but the size of the jump distinguishes them. A second of producer
|
|
* time is orders of magnitude more than any reordering window a UDP path can
|
|
* produce (a few packet intervals) and orders of magnitude less than a restart,
|
|
* which drops hrt from the producer's whole uptime back to near zero.
|
|
*
|
|
* Deliberately NOT kBurstResyncThresholdS: that one asks how far a WALL-clock
|
|
* prediction may sit from arrival, a different quantity in a different clock
|
|
* that happens to be tuned for delivery jitter. Sharing the number would couple
|
|
* two unrelated tunings.
|
|
*/
|
|
static constexpr double kProducerRestartS = 1.0;
|
|
|
|
/**
|
|
* The declared sampling rate, or 0 when there is none to trust.
|
|
*
|
|
* samplingRate arrives unvalidated from a signal descriptor on the wire. A
|
|
* malformed +inf reaches the reciprocal as dt == 0, which makes a burst's
|
|
* nominal width zero and the proportional squeeze compute 0.0/0.0 — and a NaN
|
|
* factor is not caught by the floor, since every comparison against NaN is
|
|
* false, so the whole burst is emitted as NaN. Rejecting it at the boundary
|
|
* costs one test and removes the entire class.
|
|
*/
|
|
static double DeclaredRate(double samplingRate) {
|
|
return (std::isfinite(samplingRate) && samplingRate > 0.0) ? samplingRate
|
|
: 0.0;
|
|
}
|
|
|
|
void FrameDecoder::setSignals(const std::vector<SignalMeta>& 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<double>& 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<double>(nElems);
|
|
tsOut.resize(nElems);
|
|
for (uint32_t e = 0; e < nElems; e++) {
|
|
tsOut[e] = st.lastPacketWall + static_cast<double>(e + 1u) * dt;
|
|
}
|
|
st.lastPacketWall = wallNow;
|
|
return true;
|
|
}
|
|
|
|
bool FrameDecoder::timestamps(const FrameView& f, uint32_t idx,
|
|
std::vector<double>& tsOut) {
|
|
tsOut.clear();
|
|
if (idx >= signals_.size() || idx >= f.numSignals ||
|
|
f.counts == nullptr || f.values == 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];
|
|
|
|
/* A repeated counter is a duplicated datagram — the same update arriving
|
|
* twice because the host joined the multicast group on two interfaces, say.
|
|
* The C client only de-duplicates fragments, so an unfragmented update
|
|
* reaches us intact both times; emitting it again would double the values
|
|
* and advance the timeline by a burst that never existed. Counter zero is
|
|
* excluded because a producer that never sets one leaves it there.
|
|
*
|
|
* Keyed on counterValid, not lastEmittedValid: rules 1 and 2 are exposed to
|
|
* the same double delivery and would otherwise plot every array twice, since
|
|
* neither of them ever joins rule 3's emitted chain. */
|
|
if (st.counterValid && f.counter != 0u && f.counter == st.lastCounter) {
|
|
return false;
|
|
}
|
|
|
|
/* hasTimeSignal() bounds the index against the FRAME's signal count, but
|
|
* the time signal's type code is read from our own table, whose size is
|
|
* independent — a frame carrying more signals than the installed table
|
|
* (briefly possible after a CONFIG change) would otherwise read past it. */
|
|
const bool hasTimeSig = d.hasTimeSignal(f.numSignals) &&
|
|
d.timeSignalIdx < signals_.size();
|
|
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;
|
|
}
|
|
/* Only so the duplicate-datagram guard above has something to compare
|
|
* against; nothing in this rule reads it back. */
|
|
st.lastCounter = f.counter;
|
|
st.counterValid = true;
|
|
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 prodSec = f.values[tIdx][0] * tScale;
|
|
const double anchor = st.offset.map(prodSec, wallNow);
|
|
const double rate = DeclaredRate(d.samplingRate);
|
|
double dt = (rate > 0.0) ? (1.0 / rate) : 0.0;
|
|
|
|
/* No rate declared. UDPSourceSession.cpp:522 leaves dt at zero here,
|
|
* which stacks every element of the array on one instant — harmless for
|
|
* a host-local consumer that only stores them, but this scope's ring,
|
|
* decimator and trigger all require a signal's stamps to increase, and a
|
|
* plot of N points at one X is not a trace.
|
|
*
|
|
* The spread is recoverable without a rate: consecutive anchors come
|
|
* from the time signal, so their difference is the burst's true duration
|
|
* in producer seconds, measured on the producer's own clock rather than
|
|
* on arrival — immune to the bursty delivery that corrupts everything
|
|
* arrival-derived. Divide by the counter gap for the same reason rule 3
|
|
* does: a lost datagram widens the anchor difference without widening
|
|
* the array.
|
|
*
|
|
* Which array, though, is a question of which end is anchored. For
|
|
* LAST_SAMPLE the anchors bracket THIS packet's elements, so the divisor
|
|
* is nElems; for FIRST_SAMPLE they bracket the PREVIOUS packet's, so it
|
|
* is that packet's count. Accumulate mode flushes on a timer, so the
|
|
* count really does vary between packets — using the wrong one against a
|
|
* 10,10,2,10,20 pattern gave 5x, 0.2x and 0.5x the true period and one
|
|
* backward step of 3 ms.
|
|
*
|
|
* Two packets cannot always be measured. The first has no predecessor,
|
|
* and a reordered one has an anchor behind its predecessor's; rather
|
|
* than stack the whole array on one instant — the very defect this
|
|
* paragraph exists to remove — reuse the last period actually measured,
|
|
* exactly as the hrt branch reuses lastHrtDt. Only the genuine first
|
|
* packet of a run stacks, and only until the second arrives. */
|
|
if (!(dt > 0.0) && nElems > 1u) {
|
|
const uint32_t divisor = (d.timeMode == kTimeLastSample)
|
|
? nElems : st.prevAnchorCount;
|
|
const uint32_t rawGap = f.counter - st.lastCounter;
|
|
const bool fwdGap = (f.counter != 0u) && st.counterValid &&
|
|
(rawGap != 0u) && (rawGap < 0x80000000u);
|
|
if (st.prevAnchorValid && prodSec > st.prevAnchorProdSec &&
|
|
divisor > 0u) {
|
|
dt = (prodSec - st.prevAnchorProdSec) /
|
|
(static_cast<double>(divisor) *
|
|
static_cast<double>(fwdGap ? rawGap : 1u));
|
|
st.prevAnchorDt = dt;
|
|
} else if (st.prevAnchorDt > 0.0) {
|
|
dt = st.prevAnchorDt;
|
|
}
|
|
}
|
|
/* All four move together or not at all: a reordered packet must not
|
|
* leave a newer anchor and an older counter behind for the next one to
|
|
* divide one by the other. */
|
|
if (!st.prevAnchorValid || prodSec > st.prevAnchorProdSec) {
|
|
st.prevAnchorProdSec = prodSec;
|
|
st.prevAnchorCount = nElems;
|
|
st.prevAnchorValid = true;
|
|
st.lastCounter = f.counter;
|
|
st.counterValid = true;
|
|
}
|
|
|
|
tsOut.resize(nElems);
|
|
for (uint32_t e = 0; e < nElems; e++) {
|
|
tsOut[e] = (d.timeMode == kTimeFirstSample)
|
|
? (anchor + static_cast<double>(e) * dt)
|
|
: (anchor - static_cast<double>(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.
|
|
*
|
|
* A signal that has already produced a burst stays on this rule even when a
|
|
* later packet carries a single sample — Accumulate mode flushes on a timer,
|
|
* so a short cycle legitimately yields one. Dropping such a packet to rule 5
|
|
* would date it from arrival while its neighbours are chained, and would
|
|
* leave lastCounter behind so the next real burst read the skip as a lost
|
|
* datagram and reinstated a hole that never existed. A signal that has never
|
|
* burst is a genuine scalar and is left to rule 5. */
|
|
if (d.numElements() == 1u && (nElems > 1u || st.lastEmittedValid)) {
|
|
/* A rate that is not a finite positive number is no rate at all; see
|
|
* DeclaredRate(). Such a signal takes the hrt branch below. */
|
|
const double declared = DeclaredRate(d.samplingRate);
|
|
const double dt = (declared > 0.0) ? (1.0 / declared) : 0.0;
|
|
|
|
if (declared > 0.0) {
|
|
/* Where arrival time says this burst begins: its last element was
|
|
* acquired just before the packet landed. */
|
|
const double arrivalAnchor =
|
|
wallNow - static_cast<double>(nElems - 1u) * dt;
|
|
|
|
/* Chaining onto the end of the previous burst is immune to arrival
|
|
* jitter — a kernel draining several queued datagrams microseconds
|
|
* apart still yields contiguous timestamps. What a bare chain gets
|
|
* wrong is loss: it closes the hole a dropped datagram left, and
|
|
* every later sample is then dated early for the rest of the run.
|
|
*
|
|
* The wire says exactly how much is missing. counter increments
|
|
* once per update, so a gap of g means g-1 lost packets, each
|
|
* carrying (as far as we can tell) as many samples as the last one
|
|
* we saw. Reinstating that duration keeps the chain honest without
|
|
* consulting arrival time at all. */
|
|
double base = arrivalAnchor;
|
|
double step = dt;
|
|
if (st.lastEmittedValid) {
|
|
/* Unsigned subtraction wraps, so this stays right across the
|
|
* counter's own 2^32 rollover.
|
|
*
|
|
* A producer restart or a reordered datagram makes the wrapped
|
|
* gap enormous, and this deliberately does NOT special-case
|
|
* that: an absurd gap yields an absurd prediction, which the
|
|
* arrival backstop below then rejects on its own.
|
|
*
|
|
* Clamping the gap first is not a harmless earlier version of
|
|
* the same decision — it reaches the OPPOSITE answer. A clamp
|
|
* that treats gap > kMaxCounterGap as unknowable has to fall
|
|
* back to lost == 0, so the prediction becomes
|
|
* lastEmittedEnd + dt: one sample period after the last burst,
|
|
* which is exactly the shape of a healthy chain and therefore
|
|
* lands INSIDE the arrival backstop, is accepted, and silently
|
|
* closes an outage of arbitrary length. Letting the absurd gap
|
|
* through produces an absurd prediction that the backstop
|
|
* catches, and the burst re-anchors on arrival — which is the
|
|
* right answer, and what
|
|
* AccumulatedScalarSurvivesAProducerRestart pins down. The
|
|
* arithmetic cannot overflow: gap and prevAccCount are both
|
|
* bounded by 2^32-1, so lost is at most ~1.8e19, finite, and
|
|
* always rejected. */
|
|
const uint32_t gap = f.counter - st.lastCounter;
|
|
const double lost = (gap > 1u)
|
|
? static_cast<double>(gap - 1u) *
|
|
static_cast<double>(st.prevAccCount)
|
|
: 0.0;
|
|
const double predicted = st.lastEmittedEnd + dt * (1.0 + lost);
|
|
|
|
/* Backstop for what the counter cannot express: a producer
|
|
* restart, a counter stuck at zero, or a declared rate that is
|
|
* simply wrong. Beyond this the chain is not recoverable and
|
|
* arrival time is the better of two bad answers. */
|
|
if (std::fabs(predicted - arrivalAnchor) <= kBurstResyncThresholdS) {
|
|
base = predicted;
|
|
}
|
|
|
|
if (base <= st.lastEmittedEnd) {
|
|
/* Re-anchoring here would step backwards, and the ring, the
|
|
* trigger and the exporter all require a signal's stamps to
|
|
* increase. Rejecting the correction outright is not an
|
|
* option either: `predicted` is never less than
|
|
* lastEmittedEnd + dt, so rejection would make the backstop
|
|
* one-directional and let a timeline that runs FAST — two
|
|
* hosts' crystals differ by tens of ppm, so this is certain
|
|
* on a long session, not hypothetical — drift ahead of the
|
|
* wall clock without bound.
|
|
*
|
|
* So compress instead of stepping back: start immediately
|
|
* after the previous burst and spread this one out to
|
|
* arrival. A single packet is drawn narrower than its true
|
|
* width, and in exchange the timeline is back in step. */
|
|
if (wallNow > st.lastEmittedEnd) {
|
|
step = (wallNow - st.lastEmittedEnd) /
|
|
static_cast<double>(nElems);
|
|
base = st.lastEmittedEnd + step;
|
|
} else {
|
|
/* The timeline has run PAST arrival: our last burst is
|
|
* dated later than the moment this packet landed, so
|
|
* there is no room to spread into and no burst can end
|
|
* on arrival without starting before it. Squeeze this
|
|
* one instead, by the excess and by the wall time that
|
|
* has really elapsed since this signal's last burst.
|
|
*
|
|
* Both terms are needed, and only the second one
|
|
* converges. Within this call the wall clock is frozen
|
|
* at wallNow, so ANY positive step increases the lead
|
|
* measured at this instant; the lead falls only because
|
|
* the wall advances BETWEEN packets. A step expressed
|
|
* purely as a fraction of the nominal burst width
|
|
* therefore diverges as soon as the nominal width
|
|
* outruns the packet interval — with the floor alone, a
|
|
* declared 30 Hz against a producer really flushing 10
|
|
* samples at 1 kHz gained ~0.67 s of lead per second of
|
|
* stream, without bound. Capping the burst's total
|
|
* advance at kWallBleedFraction of the elapsed wall time
|
|
* makes it advance strictly slower than the wall for any
|
|
* declared rate, so the lead strictly falls.
|
|
*
|
|
* The proportional term still does the fine work: when
|
|
* the excess is smaller than a burst it removes it in
|
|
* one packet. kMinBleedFactor only keeps that term
|
|
* positive when the excess exceeds a whole burst.
|
|
*
|
|
* Steady state is a sawtooth, not a fixed offset: the
|
|
* squeeze pulls the lead down, ordinary chaining resumes
|
|
* on the very next packet and pushes it back up until
|
|
* the prediction misses arrival by more than
|
|
* kBurstResyncThresholdS. So the lead cycles between a
|
|
* fraction of a millisecond and roughly that threshold —
|
|
* bounded, which is what matters, but not zero. */
|
|
const double nominal = static_cast<double>(nElems) * dt;
|
|
const double excess = st.lastEmittedEnd - wallNow;
|
|
double factor = 1.0 - excess / nominal;
|
|
if (factor < kMinBleedFactor) { factor = kMinBleedFactor; }
|
|
double advance = nominal * factor;
|
|
|
|
/* A non-positive elapsed means the wall has not moved
|
|
* since this signal's previous burst. Skipping the cap
|
|
* then is deliberate and, more to the point, makes no
|
|
* difference: forcing the cap to zero instead sends step
|
|
* through the floor below to dt * kMinBleedFactor, which
|
|
* is the same number the proportional factor already
|
|
* yields once the excess exceeds one burst. Both leave
|
|
* the same-tick case diverging; only real elapsed wall
|
|
* time can bleed lead off, and a recv_time from
|
|
* CLOCK_REALTIME (udps_client.c:120) does not repeat. */
|
|
const double wallElapsed = wallNow - st.lastEmittedWall;
|
|
if (wallElapsed > 0.0) {
|
|
const double cap = kWallBleedFraction * wallElapsed;
|
|
if (cap < advance) { advance = cap; }
|
|
}
|
|
step = advance / static_cast<double>(nElems);
|
|
/* Unreachable with a finite positive dt — kept because
|
|
* downstream monotonicity must not depend on that
|
|
* argument holding for every value off the wire. */
|
|
if (!(step > 0.0)) { step = dt * kMinBleedFactor; }
|
|
base = st.lastEmittedEnd + step;
|
|
}
|
|
}
|
|
}
|
|
tsOut.resize(nElems);
|
|
for (uint32_t e = 0; e < nElems; e++) {
|
|
tsOut[e] = base + static_cast<double>(e) * step;
|
|
}
|
|
st.lastEmittedEnd = tsOut[nElems - 1u];
|
|
st.lastEmittedWall = wallNow;
|
|
st.lastCounter = f.counter;
|
|
st.counterValid = true;
|
|
st.prevAccCount = nElems;
|
|
st.lastEmittedValid = true;
|
|
return true;
|
|
}
|
|
|
|
/* No declared rate: need hrt-derived dt. */
|
|
if (!hrtFit_.ready() || f.hrt == 0u) {
|
|
const bool ok = packetBurst(idx, nElems, wallNow, tsOut);
|
|
/* Carry the warm-up's state into the hrt branch, or the handover
|
|
* from one to the other is a discontinuity in both directions.
|
|
*
|
|
* The producer-clock reference (lastAccHrt, prevAccCount) matters
|
|
* most. Without it the first hrt packet has no previous tick to
|
|
* subtract, falls back to kDefaultDt for its inter-element step and
|
|
* latches ClockOffset against wallNow - (nElems-1)*kDefaultDt.
|
|
* kDefaultDt is only right when the burst happens to run at 1 kHz;
|
|
* at 100 samples per 10 ms packet it is ten times too wide and the
|
|
* latch lands 89 ms in the past — permanently, since it is below
|
|
* ClockOffset's recalibration threshold. Seeding here means the
|
|
* first hrt packet measures a real tick delta and latches correctly.
|
|
*
|
|
* The emitted-timeline reference (lastEmitted*) then only has to
|
|
* cover residual disagreement, but it is what keeps the handover
|
|
* MONOTONIC: packetBurst ends its burst at wallNow while the hrt
|
|
* branch ends at wallNow - (nElems-1)*hrtDt, and without a previous
|
|
* end to clamp against the first hrt packet steps the signal
|
|
* backwards by up to a whole burst width. */
|
|
if (f.hrt != 0u) {
|
|
st.lastAccHrt = f.hrt;
|
|
st.lastAccValid = true;
|
|
}
|
|
/* Same lockstep rule as the hrt branch below. A packet with hrt == 0
|
|
* lands here mid-stream and cannot move the tick reference, so it
|
|
* must not move the counter either: advancing the counter alone
|
|
* makes the next packet's elapsed span two intervals while its gap
|
|
* reports one, drawing that burst twice too wide and ending it
|
|
* 22.5 ms in the future at 10 samples per 25 ms packet (6x and
|
|
* +112 ms after five such packets). Before any tick reference exists
|
|
* nothing is keyed to the counter, so it is free to advance and arm
|
|
* the duplicate guard for a producer that never sets hrt at all. */
|
|
if (f.hrt != 0u || !st.lastAccValid) {
|
|
st.lastCounter = f.counter;
|
|
st.counterValid = true;
|
|
}
|
|
st.prevAccCount = nElems;
|
|
if (!ok) { return false; }
|
|
st.lastEmittedEnd = tsOut[nElems - 1u];
|
|
st.lastEmittedWall = wallNow;
|
|
st.lastEmittedValid = true;
|
|
return true;
|
|
}
|
|
const double rate = hrtFit_.ticksPerSecond();
|
|
|
|
/* Integrate short tick DELTAS. Never convert an absolute tick count, and
|
|
* never subtract two such conversions.
|
|
*
|
|
* hrt counts from the producer's boot, so it is already ~1e11 ticks when
|
|
* the scope attaches, while the fit is re-estimated on every packet and
|
|
* wobbles by a few parts in 1e4. Any absolute hrt/rate therefore carries
|
|
* that relative wobble multiplied by the whole elapsed epoch — tens of
|
|
* milliseconds, moving in either direction from one packet to the next.
|
|
* As a burst's position that is not merely imprecise, it is
|
|
* NON-MONOTONIC: on a 2 h stream with ordinary scheduling jitter a few
|
|
* percent of samples land before their own predecessor.
|
|
*
|
|
* A delta spans one packet, so its share of the wobble is microseconds,
|
|
* and summing deltas keeps it there. ClockOffset then latches the
|
|
* arbitrary epoch that leaves behind, exactly as it would have latched
|
|
* the producer's boot epoch. */
|
|
double elapsed = 0.0;
|
|
/* Whether lastAccHrt should take this packet's value. Only a packet
|
|
* that legitimately defines the new front of producer time may move it;
|
|
* see the backward case below. */
|
|
bool takeHrt = true;
|
|
if (st.lastAccValid) {
|
|
if (f.hrt > st.lastAccHrt) {
|
|
elapsed = static_cast<double>(f.hrt - st.lastAccHrt) / rate;
|
|
} else {
|
|
/* hrt went backwards. Two entirely different events look like
|
|
* this and only the SIZE of the jump separates them.
|
|
*
|
|
* A small one is a reordered datagram: the packet that overtook
|
|
* it already counted the interval it covers, so it must
|
|
* contribute nothing — and must also leave lastAccHrt alone.
|
|
* Letting it write lastAccHrt anyway (which is what this code
|
|
* used to do unconditionally) rolls the reference back one
|
|
* interval, so the NEXT packet's delta spans two and fabricates
|
|
* a whole extra packet of producer time. It never heals:
|
|
* ClockOffset would correct it, but the monotonic clamp below
|
|
* discards every backward correction. A hundred swaps on a
|
|
* 25 ms stream left the trace 3.5 s ahead, permanently. The C
|
|
* client does not reorder for us — udps_client.c only COUNTS
|
|
* counter gaps — so this is reachable on any real network.
|
|
*
|
|
* A large one is a producer restart: hrt drops from the
|
|
* producer's whole uptime to near zero. Here the unconditional
|
|
* write was the right behaviour and must be kept, because
|
|
* refusing to regress would leave every subsequent packet below
|
|
* lastAccHrt forever, elapsed permanently zero and the signal
|
|
* frozen. Rebase, and reset the offset so it re-latches against
|
|
* the new epoch instead of being dragged there by recalibration. */
|
|
const double backward =
|
|
static_cast<double>(st.lastAccHrt - f.hrt) / rate;
|
|
if (backward > kProducerRestartS) {
|
|
st.offset.reset();
|
|
} else {
|
|
takeHrt = false;
|
|
}
|
|
}
|
|
}
|
|
st.accProdSec += elapsed;
|
|
|
|
/* The flushes carry contiguous RT cycles, so the tick gap divided by the
|
|
* number of cycles it spans is exactly one cycle period. That count is
|
|
* NOT prevAccCount: elapsed spans every packet since the last one we
|
|
* saw, so a lost datagram makes the tick gap wider without making
|
|
* prevAccCount larger. Dividing by prevAccCount alone therefore returns
|
|
* a period scaled by the whole counter gap — 2x for one lost datagram,
|
|
* 11x for ten — which draws the recovery burst that many times too wide
|
|
* and, because the burst is anchored on its LAST element, ends it in the
|
|
* FUTURE (measured: +22.5 ms for one loss, +225 ms for ten, at 10
|
|
* samples per 25 ms packet). At 1% loss that mis-spaced 4.0% of all
|
|
* samples. The declared branch already reads the counter for exactly
|
|
* this purpose (`lost`, above); the hrt branch must too.
|
|
*
|
|
* The gap and `elapsed` must be measured from the SAME packet, or the
|
|
* division mixes references. That is why lastCounter is written under
|
|
* takeHrt below, in lockstep with lastAccHrt: a reordered datagram that
|
|
* rolled lastCounter back while leaving lastAccHrt alone would give the
|
|
* next in-order packet a gap of d+1 against an elapsed spanning one
|
|
* interval, dividing its period by d+1 — measured 0.5x the true spacing
|
|
* for a swap of neighbours, 0.048x for a distance of twenty.
|
|
*
|
|
* Unsigned subtraction wraps, which is what makes this right across the
|
|
* counter's own 2^32 rollover. A gap in the top half of the range is not
|
|
* a forward gap at all but a backward one seen through the wrap, so it
|
|
* is treated as no information rather than as 2 billion lost packets. */
|
|
const uint32_t rawGap = f.counter - st.lastCounter;
|
|
const bool fwdGap = (f.counter != 0u) && st.counterValid &&
|
|
(rawGap != 0u) && (rawGap < 0x80000000u);
|
|
const double cycles = static_cast<double>(st.prevAccCount) *
|
|
static_cast<double>(fwdGap ? rawGap : 1u);
|
|
|
|
/* Falling back to kDefaultDt is a last resort, not a default: see
|
|
* SigState::lastHrtDt. What makes the fallback matter is the producer
|
|
* restart, because that is the one path where elapsed is zero AND
|
|
* offset.reset() has just forced a re-latch, so the burst width used
|
|
* here is the one calibrated against — measured 13.5 ms of standing
|
|
* displacement at 10 samples per 25 ms packet, 89 ms at 100 per 10 ms,
|
|
* both below kRecalibThresholdS and so never corrected. A reorder also
|
|
* reaches the fallback but does not re-latch, and its measured standing
|
|
* error is 0.013 ms, i.e. nothing.
|
|
*
|
|
* This is better than kDefaultDt at every cadence except one: a producer
|
|
* that restarts having CHANGED its cycle time is remembered wrongly, and
|
|
* a tenfold change measured -20 ms against kDefaultDt's -6.8 ms. Both
|
|
* are bounded and neither is correct; the remembered period wins the
|
|
* case that actually happens. */
|
|
double hrtDt;
|
|
if (elapsed > 0.0 && cycles > 0.0) {
|
|
hrtDt = elapsed / cycles;
|
|
st.lastHrtDt = hrtDt;
|
|
} else if (st.lastHrtDt > 0.0) {
|
|
hrtDt = st.lastHrtDt;
|
|
} else {
|
|
hrtDt = kDefaultDt;
|
|
}
|
|
|
|
/* Anchor the burst's LAST element on arrival, not its first. The
|
|
* packet's hrt is the tick count of sample 0 (UDPSourceSession.cpp:574),
|
|
* so stepping forward from it is right — but ClockOffset latches
|
|
* offset = wall - producerSec on its first call, and passing the raw
|
|
* arrival would put sample 0 at the instant the packet carrying the
|
|
* whole burst LANDED, dating every sample in it late by a burst. The
|
|
* declared-rate branch above already anchors on the burst end
|
|
* (arrivalAnchor), and two accumulated scalars in one scope, one with a
|
|
* declared rate and one without, would otherwise sit a burst apart on a
|
|
* shared X axis — 9 ms for 10 samples at 1 kHz, plain to see at a 200 ms
|
|
* window. Since map() latches once, this is a constant shift applied at
|
|
* latch and recalibration only; it changes no spacing. */
|
|
double base = st.offset.map(
|
|
st.accProdSec,
|
|
wallNow - static_cast<double>(nElems - 1u) * hrtDt);
|
|
double step = hrtDt;
|
|
|
|
/* ClockOffset recalibrates once true drift passes its threshold, and a
|
|
* recalibration can land behind where this signal already is.
|
|
* Downstream requires increasing stamps, so step forward minimally —
|
|
* but a bare forward step is one-directional, exactly the defect the
|
|
* declared branch's squeeze exists to avoid. A backward wall step (an
|
|
* NTP correction, a suspend/resume) would otherwise leave this signal
|
|
* permanently ahead of the wall clock, since the recalibrated base is
|
|
* behind lastEmittedEnd on every later packet too and the clamp keeps
|
|
* discarding it. So cap the burst's total advance against the wall time
|
|
* elapsed since this signal's previous burst, for the reason spelled out
|
|
* at kWallBleedFraction: only that makes the lead bleed off. */
|
|
if (st.lastEmittedValid && base <= st.lastEmittedEnd) {
|
|
const double wallElapsed = wallNow - st.lastEmittedWall;
|
|
if (wallElapsed > 0.0) {
|
|
const double cap = kWallBleedFraction * wallElapsed /
|
|
static_cast<double>(nElems);
|
|
if (cap < step) { step = cap; }
|
|
}
|
|
base = st.lastEmittedEnd + step;
|
|
}
|
|
|
|
tsOut.resize(nElems);
|
|
for (uint32_t e = 0; e < nElems; e++) {
|
|
tsOut[e] = base + static_cast<double>(e) * step;
|
|
}
|
|
if (takeHrt) { st.lastAccHrt = f.hrt; }
|
|
st.lastAccValid = true;
|
|
st.prevAccCount = nElems;
|
|
st.lastEmittedEnd = tsOut[nElems - 1u];
|
|
st.lastEmittedWall = wallNow;
|
|
/* Keep packetBurst's reference current even though this branch does not
|
|
* use it. A single packet with hrt == 0 re-enters the warm-up branch
|
|
* above, and packetBurst would otherwise span from whenever this signal
|
|
* last took that branch — the whole session. Measured: 153 packets into
|
|
* a 25 ms stream, one zero-hrt packet emitted a burst starting 2.74 s
|
|
* behind the trace, and that figure grows with session length. */
|
|
st.lastPacketWall = wallNow;
|
|
st.lastPacketValid = true;
|
|
/* In lockstep with lastAccHrt, and for the same reason: these two are
|
|
* the numerator and the denominator of the next packet's period. Only a
|
|
* packet that defines the new front of producer time may move either.
|
|
* Advancing the counter alone on a reordered datagram halves the next
|
|
* packet's spacing; see the gap comment above. Leaving a counter behind
|
|
* at all is what lets the duplicate-datagram guard at the top of
|
|
* timestamps() fire — a host joined on two interfaces receives every
|
|
* unfragmented update twice, and the original always sets takeHrt. */
|
|
if (takeHrt) {
|
|
st.lastCounter = f.counter;
|
|
st.counterValid = true;
|
|
}
|
|
st.lastEmittedValid = true;
|
|
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 */
|