fix: a lost packet reserves the samples it owed instead of vanishing from the timeline

This commit is contained in:
Martino Ferrari committed 2026-10-08 18:18:12 +02:00
1 parent 85ba835d61
commit 5915fc8421
4 files changed
+183 -9

No files matched your search

+114 -8
View File
@@ -14,6 +14,12 @@
namespace udpscope {
namespace {
/// How far a packet's stamps may run ahead before the jump is read as a clock fault rather
/// than a stream gap. Also the cap for hole reservation: 256 packet spans in one go is an
/// outage, not a loss, and reserving one would push the entire retained history out of the
/// ring to make room for samples that are not coming.
constexpr double kGlitchPackets = 256.0;
SignalInfo to_info(const udps_signal_t& s) {
SignalInfo info;
info.name = s.name;
@@ -165,6 +171,13 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign
}
double dt = model_.partition().base_dt;
if (!(dt > 0.0)) dt = 1.0e-3;
// The reserved hole for the missing packets advances `slot_base_` so this frame's
// own stamps land *after* the hole rather than on top of it. The same conditions
// the reservation call uses: not late, gap present, batch known, within the cap.
if (!frame->late && frame->lost > 0 && batch_ > 0 &&
double(frame->lost) <= kGlitchPackets) {
slot_base_ += u64(frame->lost) * batch_;
}
for (std::uint32_t k = 0; k < n; ++k) {
stamps_[k] = double(slot_base_ + k) * dt;
}
@@ -276,13 +289,6 @@ std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_sign
// One sample's worth of the time signal's *own* unit, which is what a synthesised
// continuation has to be spaced by and what a jump is measured against.
const double step = cadence / unit_scale;
// How far a packet's stamps may run ahead of the previous packet's last stamp
// before the jump is read as a fault in the producer's clock rather than a gap in
// the stream. Counted in packets, because that is what a loss burst costs: 256
// packet spans in one go is an outage, not loss. Letting a real glitch through is
// unrecoverable -- the monotonic clamp below then pins every later packet flat at
// the glitched value, so the whole stream freezes at one x.
constexpr double kGlitchPackets = 256.0;
// How many packets in a row must fail before the declared time signal is abandoned
// for the arrival clock. A glitch is one packet; a signal that is not a time base
// at all fails every one of them.
@@ -441,6 +447,14 @@ void Acquisition::on_data(const udps_frame_t* frame) {
std::lock_guard<std::mutex> lock(model_.mutex());
if (!running_.load(std::memory_order_relaxed)) return;
// The counter is this frame's identity and the hole's address. A *late* frame's counter is
// behind the sequence, so it must not move the mark -- the hole it belongs in was addressed
// from the counter that revealed it.
if (!have_counter_ || static_cast<std::int32_t>(frame->counter - last_counter_) > 0) {
last_counter_ = frame->counter;
have_counter_ = true;
}
const Partition& part = model_.partition();
const std::vector<ChannelInfo>& channels = model_.channels();
if (part.time_capacity == 0 || channels.empty()) return;
@@ -487,6 +501,30 @@ void Acquisition::on_data(const udps_frame_t* frame) {
BufferStore& store = model_.live();
const u64 g = part.time_stride;
// The gap the counter reports, reserved as NaN before this frame's own samples are
// stamped: the hole belongs *ahead* of the data that revealed it. Running the missing
// samples through the same per-sample loop below is the whole trick -- `tick_` advances
// so the decimation phase does not shift, the group mean lands on the mean stamp, and
// `MinMaxPyramid::append` sees NaN, which its non-finite contract already stores as an
// empty aggregate every reader treats as absent.
//
// Capped at `kGlitchPackets`: 256 packet spans in one go is an outage, not loss, and
// reserving one would push the entire retained history out of the ring to make room for
// samples that are not coming.
const std::uint32_t missing = frame->late ? 0u : frame->lost;
if (missing > 0 && batch_ > 0 && double(missing) <= kGlitchPackets) {
reserve_hole(missing, part, channels, store, g);
} else if (missing > 0 && !warned_outage_) {
warned_outage_ = true;
std::fprintf(stderr,
"[udpscope] warning: %u packets missing in one go, past the %g-packet "
"reservation cap; treating it as an outage and carrying on rather than "
"reserving %llu samples of hole.\n",
missing, kGlitchPackets,
static_cast<unsigned long long>(std::uint64_t(missing) * batch_));
std::fflush(stderr);
}
// The filter's groups are `g` long, so a re-plan has to start a fresh group rather than
// finish the one it was in with the wrong length.
if (acc_stride_ != g || acc_channels_ != channels.size()) {
@@ -616,7 +654,8 @@ void Acquisition::on_data(const udps_frame_t* frame) {
// boxcar's group delay puts the averaged value, and stamping it at the end of the
// group would shift every trace half a stored sample late.
time_acc_ += rel;
abs_time_acc_ += abs_stamps_[k] - epoch;
last_abs_stamp_ = abs_stamps_[k] - epoch;
abs_time_acc_ += last_abs_stamp_;
++time_acc_n_;
if (tick % g == g - 1) {
store.time().push(time_acc_ / static_cast<double>(time_acc_n_));
@@ -721,6 +760,73 @@ void Acquisition::on_data(const udps_frame_t* frame) {
}
}
void Acquisition::reserve_hole(std::uint32_t missing, const Partition& part,
const std::vector<ChannelInfo>& channels, BufferStore& store,
u64 g) {
const u64 raw = u64(missing) * u64(batch_);
Hole h;
h.tick_begin = tick_;
h.slot_begin = store.time().produced();
h.counter_first = last_counter_ + 1u;
h.packets = missing;
double dt = part.base_dt;
if (!(dt > 0.0)) dt = 1.0e-3;
for (u64 j = 0; j < raw; ++j) {
const u64 tick = tick_++;
// The hole is stamped at the declared cadence from the last good stamp -- exact in
// relative mode, and in stream mode the same continuation the glitch bridge makes.
// It must be *stamped* rather than left short: the time ring is binary-searched by
// `ChannelBuffer::neighbours()`, so it has to stay monotone and dense, and backfill
// then changes values only and never a stamp.
const double rel = last_stamp_ + dt;
last_stamp_ = rel;
last_abs_stamp_ += dt;
if (j == 0) h.x_begin = rel;
h.x_end = rel;
time_acc_ += rel;
abs_time_acc_ += last_abs_stamp_;
++time_acc_n_;
if (tick % g == g - 1) {
store.time().push(time_acc_ / double(time_acc_n_));
// Null in stream mode, where `time()` already is the absolute clock.
if (plot::data::RingBuffer<f64>* abs = store.abs_time(); abs != nullptr) {
abs->push(abs_time_acc_ / double(time_acc_n_));
}
time_acc_ = 0.0;
abs_time_acc_ = 0.0;
time_acc_n_ = 0;
}
for (std::size_t i = 0; i < channels.size(); ++i) {
if (channels[i].math) continue;
// A NaN in the accumulator makes the whole group NaN, which is why a slot straddling
// a hole edge is NaN too: its real members were averaged into a sum that is now
// non-finite and were never retained, so the backfill cannot recover it and must not
// pretend to. One slot per edge, against a hole thousands of slots wide.
value_acc_[i] += std::numeric_limits<double>::quiet_NaN();
++value_acc_n_[i];
const u64 period = g * part.stride[i];
if (tick % period != period - 1) continue;
store.values(i).push(static_cast<f32>(value_acc_[i] / double(value_acc_n_[i])));
value_acc_[i] = 0.0;
value_acc_n_[i] = 0;
}
// The math channels are fed NaN like everything else, so a derived trace keeps the gap
// its source had. The trigger is *not* fed: a NaN comparison is false, so the engine
// could not latch here anyway, and feeding it would advance its hold-off clock over
// time nothing was sampled in.
if (tick % g == g - 1) {
std::fill(math_sample_.begin(), math_sample_.end(), std::nanf(""));
model_.push_math_sample(math_sample_);
}
}
h.slot_count = store.time().produced() - h.slot_begin;
model_.note_hole(h);
model_.stats().holes.fetch_add(1, std::memory_order_relaxed);
model_.stats().hole_slots.fetch_add(h.slot_count, std::memory_order_relaxed);
}
// Static trampolines: the client's C ABI cannot call a member function directly.
void Acquisition::config_trampoline(const udps_signal_t* signals, std::uint32_t count,
std::uint8_t publish_mode, void* user) {
+20
View File
@@ -12,6 +12,7 @@
#include <atomic>
#include <chrono>
#include <cstdint>
#include <limits>
#include <thread>
#include <vector>
@@ -53,6 +54,13 @@ class Acquisition {
/// malformed).
std::uint32_t build_stamps(const udps_frame_t* frame, int time_signal);
/// Pushes NaN for the `missing` packets' worth of samples that were never delivered,
/// advances the decimation accumulators and tick_, records the hole in the model, and
/// bumps the hole counters. Called before the frame's own samples so the gap sits
/// *ahead* of the data that revealed it.
void reserve_hole(std::uint32_t missing, const Partition& part,
const std::vector<ChannelInfo>& channels, BufferStore& store, u64 g);
ScopeModel& model_;
Config cfg_;
udps_client* client_ = nullptr;
@@ -101,6 +109,18 @@ class Acquisition {
std::uint32_t batch_ = 0;
/// Latches the one warning about a frame whose length disagrees with the stream's first.
bool warned_batch_ = false;
/// The newest counter seen, and the hole's address. Tracked here rather than inferred from
/// `frame->lost` because a *late* frame's counter is behind the sequence and must not move
/// the mark forward.
std::uint32_t last_counter_ = 0;
bool have_counter_ = false;
/// Last absolute stamp written, epoch-relative, so a reserved hole can continue the
/// absolute clock at the declared cadence. The packets that would have carried the real
/// stamps never arrived, so a cadence continuation is the only honest value available.
double last_abs_stamp_ = 0.0;
/// Warn-once latch for a loss past the reservation cap. A member, not a function-local
/// `static`, matching the latch Task 2 added for the same reason.
bool warned_outage_ = false;
std::vector<double> stamps_;
std::vector<double> abs_stamps_;
/// The decimation filter's state: a running sum per channel (and one for the shared
+19 -1
View File
@@ -654,7 +654,25 @@ void ScopeModel::set_display_epoch(f64 e) noexcept {
}
void ScopeModel::acquire_all() {
if (store_) store_->acquire_all();
if (store_) {
store_->acquire_all();
// Drop holes whose slots the time ring has already overwritten: their slot addresses
// would alias live slots once the ring wraps past them, and they can no longer be
// backfilled or drawn. `base()` is the absolute index of the oldest slot the ring
// still retains -- anything earlier has already been discarded.
forget_holes_before(store_->time().base());
}
}
void ScopeModel::note_hole(const Hole& h) {
if (holes_.size() >= kMaxHoles) holes_.erase(holes_.begin());
holes_.push_back(h);
}
void ScopeModel::forget_holes_before(u64 slot) {
// A hole whose last slot the ring has dropped can no longer be backfilled or drawn, and
// its address would alias a live slot once the ring wraps past it.
std::erase_if(holes_, [slot](const Hole& h) { return h.slot_begin + h.slot_count <= slot; });
}
void ScopeModel::set_trigger_enabled(bool on) {
+30
View File
@@ -167,6 +167,24 @@ class BufferStore {
std::vector<std::shared_ptr<ChannelBuffer>> channels_;
};
/// One reserved, invalidated span: samples the counter says exist and the stream has not
/// delivered. Its stored slots hold NaN, so a reader needs no special case -- the table is
/// what lets a *late* packet find the slots it belongs to, and what the band draws.
struct Hole {
f64 x_begin = 0.0; ///< stream seconds of the hole's first slot
f64 x_end = 0.0; ///< stream seconds of the hole's last slot
/// Absolute **time-ring** slot of the first reserved slot, and the count of them. The
/// time ring is the one clock every channel shares, so this is the only slot address
/// that is unambiguous: per-channel value rings run at `time_stride * stride[i]`, which
/// differs between channels. A consumer that needs a *channel's* slots derives them from
/// `tick_begin` and that channel's stride rather than reusing these.
u64 slot_begin = 0;
u64 slot_count = 0;
u64 tick_begin = 0; ///< absolute base tick of the hole's first raw sample
std::uint32_t counter_first = 0; ///< counter of the first missing packet
std::uint32_t packets = 0; ///< packets the hole is worth
};
/// Live connection and counter state, readable from the UI thread.
struct StreamStats {
std::atomic<std::uint64_t> packets{0};
@@ -175,6 +193,8 @@ struct StreamStats {
std::atomic<std::uint64_t> reconnects{0};
std::atomic<bool> connected{false};
std::atomic<bool> have_config{false};
std::atomic<std::uint64_t> holes{0}; ///< reserved invalidated spans
std::atomic<std::uint64_t> hole_slots{0}; ///< stored slots those spans cover
};
/// What the plot is showing, which is one decision made once per frame: the bind, the
@@ -301,6 +321,15 @@ class ScopeModel {
/// is armed.
BufferStore& live();
/// Records a reserved hole. Caller holds `mutex()` (the producer already does). The table
/// is capped at `kMaxHoles`; past that the oldest is dropped, because an unbounded list of
/// gaps is a leak and the oldest gap is the one furthest from being backfilled.
static constexpr std::size_t kMaxHoles = 64;
void note_hole(const Hole& h);
const std::vector<Hole>& holes() const noexcept { return holes_; }
/// Drops holes whose time-ring slots the ring has already discarded.
void forget_holes_before(u64 slot);
/// Builds the captured event from the slice of the live buffer covering
/// `[edge - pre_s, edge + post_s]` in stream seconds, and publishes it atomically.
/// The stamps are re-based so the edge sits at x = 0. The plot binds to this
@@ -544,6 +573,7 @@ class ScopeModel {
Partition part_;
std::unique_ptr<BufferStore> store_;
std::vector<Hole> holes_;
TriggerConfig trigger_{};
StreamStats stats_{};
f64 epoch_ = 0.0;