Martino FerrariandClaude Opus 4.6
f97fd825c4
On the hrt branch the derived period is not just a spacing: it is the burst width ClockOffset latches against, so a wrong one shifts the whole trace by an amount that is usually too small for kRecalibThresholdS to ever heal. Three routes to a wrong period were open. Packet loss. elapsed spans every packet since the last one seen, but it was divided by prevAccCount alone, so a lost datagram scaled the period by the whole counter gap. Since a burst is anchored on its LAST element, too wide means it ends in the FUTURE: +22.5 ms for one loss, +225 ms for ten, at 10 samples per 25 ms packet, mis-spacing 2.7% of all samples at 1% loss. The declared branch already reads the counter for exactly this; the hrt branch now does too. Producer restart and reorder. Both leave elapsed at zero, so no period can be measured -- and the restart packet is also the one that re-latches after offset.reset(). Falling back to kDefaultDt is only right at 1 kHz; measured standing displacement was +13.5 ms at 10 samples per 25 ms and -89 ms at 100 per 10 ms. Remember the last measured period instead. A stray hrt == 0 packet re-enters the warm-up branch, which spans from packetBurst's lastPacketWall -- a field the hrt branch never wrote, so it still held the start of the session. After 153 packets that emitted a burst 3.8 s in the past, worse the longer the scope had run. Also: rule 2 with no declared rate stacked every element of the array on one instant (as UDPSourceSession.cpp:522 does, harmlessly, for a host-local consumer). Spread it from consecutive time-signal anchors, which measure the burst on the producer's own clock. Reverts the previous commit's wallElapsed <= 0 change: it was measurably inert -- the step floor two lines below already yields the same number -- and its comment claimed a divergence it did not stop. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
602 lines
32 KiB
C++
602 lines
32 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. */
|
|
if (st.lastEmittedValid && 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;
|
|
}
|
|
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. Until a second packet arrives there is nothing to measure
|
|
* and the elements do stack; that is one packet, not the whole run. */
|
|
if (!(dt > 0.0) && nElems > 1u && st.prevAnchorValid &&
|
|
prodSec > st.prevAnchorProdSec) {
|
|
const uint32_t gap = (f.counter != 0u && f.counter > st.lastCounter)
|
|
? (f.counter - st.lastCounter) : 1u;
|
|
dt = (prodSec - st.prevAnchorProdSec) /
|
|
(static_cast<double>(nElems) * static_cast<double>(gap));
|
|
}
|
|
st.prevAnchorProdSec = prodSec;
|
|
st.prevAnchorValid = true;
|
|
st.lastCounter = f.counter;
|
|
|
|
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.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;
|
|
}
|
|
st.prevAccCount = nElems;
|
|
if (!ok) { return false; }
|
|
st.lastEmittedEnd = tsOut[nElems - 1u];
|
|
st.lastEmittedWall = wallNow;
|
|
st.lastCounter = f.counter;
|
|
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 2.7% of all
|
|
* samples. The declared branch already reads the counter for exactly
|
|
* this purpose (`lost`, above); the hrt branch must too.
|
|
*
|
|
* Only a FORWARD gap counts. A backward or repeated counter is the
|
|
* reorder case handled above, where elapsed is zero anyway. */
|
|
const uint32_t accGap = (f.counter != 0u && st.lastEmittedValid &&
|
|
f.counter > st.lastCounter)
|
|
? (f.counter - st.lastCounter) : 1u;
|
|
const double cycles = static_cast<double>(st.prevAccCount) *
|
|
static_cast<double>(accGap);
|
|
|
|
/* Falling back to kDefaultDt is a last resort, not a default: see
|
|
* SigState::lastHrtDt. The fallback is reached on the first hrt packet
|
|
* of a producer restart (elapsed is zero because hrt went backwards) and
|
|
* on a reordered datagram, and in both cases the wrong burst width is
|
|
* latched into ClockOffset permanently — 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. */
|
|
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: after 153 hrt
|
|
* packets, one zero-hrt packet emitted a burst starting 3.8 s in the
|
|
* past, growing without bound with session length. */
|
|
st.lastPacketWall = wallNow;
|
|
st.lastPacketValid = true;
|
|
/* Same duplicate-datagram exposure as the declared branch: a host joined
|
|
* on two interfaces receives every unfragmented update twice, and the
|
|
* guard at the top of timestamps() can only fire if this branch leaves a
|
|
* counter behind for it to compare against. */
|
|
st.lastCounter = f.counter;
|
|
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 */
|