From 7c60d79b65c7cf123fab1dc3f71cb8b29c2796ea Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Thu, 8 Oct 2026 19:44:24 +0200 Subject: [PATCH] feat: a late packet backfills the hole it left, values only --- src/acquisition.cpp | 95 +++++++++++++++++++++++++++++++++++++++------ src/acquisition.hpp | 6 +++ src/channel.cpp | 61 ++++++++++++++++------------- src/channel.hpp | 18 +++++++++ 4 files changed, 143 insertions(+), 37 deletions(-) diff --git a/src/acquisition.cpp b/src/acquisition.cpp index 06d8532..860df80 100644 --- a/src/acquisition.cpp +++ b/src/acquisition.cpp @@ -465,20 +465,25 @@ void Acquisition::on_data(const udps_frame_t* frame) { have_counter_ = true; } - // A late frame's samples belong in the hole already reserved for them, not at the end of - // the timeline. Appending them would walk the time ring's clock backwards, which trips the - // non-monotonic guard and can throw the whole stream onto the arrival clock -- a far worse - // outcome than the dropped packet we used to have. Until the backfill lands (RingBuffer::amend - // and MinMaxPyramid::amend), late frames are dropped here, leaving the reserved hole holding - // the NaN it was reserved with -- a value every reader already treats as absent. This is - // exactly the behaviour the client had before it started delivering late packets, so nothing is - // lost relative to that. This return is the single place the backfill will later hook into. - if (frame->late) return; - const Partition& part = model_.partition(); const std::vector& channels = model_.channels(); if (part.time_capacity == 0 || channels.empty()) return; + // A frame delivered behind the sequence: the slots it owns were reserved as NaN by + // `reserve_hole()`, so it writes values into them and does nothing else. It is deliberately + // not stamped, does not advance `tick_`, and never reaches the trigger or the math channels: + // the time ring was pre-stamped at reservation and those consumers are stateful and have + // already run past this region. That is the trade the design takes -- the raw trace repairs, + // a derived trace keeps the gap, and nothing that was computed from the gap is retrofitted. + // + // Placed here and not earlier because `part`/`channels` are not in scope above, and not + // later because `build_stamps()` is stateful: a late frame's stamps would walk the clock + // backwards and can throw the whole stream onto the arrival clock. + if (frame->late) { + backfill_from(frame, part, channels, model_.live(), part.time_stride); + return; + } + // The time reference, in order of preference: the signal the first channel's // CONFIG entry names via `time_signal_idx`, or -- when the stream instead // carries its time base as an explicit full-array signal -- that signal. @@ -549,7 +554,7 @@ void Acquisition::on_data(const udps_frame_t* frame) { // through the same decimation accumulators (value_acc_, value_acc_n_, math_sample_), // which must already be sized to the current channel count and stride before any // per-sample write. - const std::uint32_t missing = frame->late ? 0u : frame->lost; + const std::uint32_t missing = frame->lost; if (missing > 0 && batch_ > 0 && double(missing) <= kGlitchPackets) { reserve_hole(missing, part, channels, store, g); } else if (missing > 0 && !warned_outage_) { @@ -852,6 +857,74 @@ void Acquisition::reserve_hole(std::uint32_t missing, const Partition& part, model_.stats().hole_slots.fetch_add(h.slot_count, std::memory_order_relaxed); } +void Acquisition::backfill_from(const udps_frame_t* frame, const Partition& part, + const std::vector& channels, BufferStore& store, + u64 g) { + // `batch_` is learned in `build_stamps()`, which a late frame never reaches. A late frame + // before any good one therefore has no packet length to size its span with, and no hole + // could have been reserved for it either. + if (batch_ == 0) return; + + // Which hole this packet belongs to, by counter: that is the whole reason relative timing + // and resequencing are one design. A stamp could not answer it -- the packet's own stamps + // are a claim about a timeline the ring has moved past -- but the counter indexes the + // reservation table directly. + const Hole* hole = nullptr; + for (const Hole& h : model_.holes()) { + if (frame->counter >= h.counter_first && frame->counter < h.counter_first + h.packets) { + hole = &h; + break; + } + } + if (hole == nullptr) return; // its hole has already slid out of the ring, or never existed + + // The packet's own span in absolute base ticks. `tick_begin` is the hole's first raw sample + // and the packets inside a hole are consecutive, so the counter's distance from + // `counter_first` places this one exactly. + const u64 lo = hole->tick_begin + u64(frame->counter - hole->counter_first) * u64(batch_); + const u64 hi = lo + u64(batch_); + + u64 filled = 0; + for (std::size_t i = 0; i < channels.size(); ++i) { + // A derived channel has no packet signal to read, and the design does not retrofit it: + // its gap stays. + if (channels[i].math) continue; + const u64 period = g * part.stride[i]; + // Only slots lying *wholly* inside this packet's span are repaired. A slot straddling the + // span's edge mixes this packet's samples with ones still missing, and its real members + // were never retained -- `reserve_hole()` averaged a NaN into them -- so it stays NaN, and + // a displayed hole is at most one slot wider per edge than the true loss. + const u64 first_slot = (lo + period - 1) / period; + const u64 last_slot = hi / period; + for (u64 s = first_slot; s < last_slot; ++s) { + // The same boxcar mean the live path stores, over the same group of raw samples: the + // whole group is inside this packet, so the average is exact and not an approximation + // of what the live path would have written. + double acc = 0.0; + for (u64 j = 0; j < period; ++j) { + const u64 k = s * period + j - lo; // this packet's own sample index, 0..batch_-1 + const double raw = + channels[i].elements > 1 + ? udps_frame_value(frame, static_cast(channels[i].signal), 0, + std::min(static_cast(k), + channels[i].elements - 1)) + : udps_frame_value(frame, static_cast(channels[i].signal), + static_cast(k), 0); + acc += raw * channels[i].gain + channels[i].offset; + } + const f32 value = static_cast(acc / double(period)); + // The ring first, by the producer that owns its writes; the pyramid is queued for the + // consumer (see `ChannelBuffer::queue_amend`). A slot the ring has already discarded + // returns false, and then there is no aggregate to amend either. + if (store.values(i).amend(s, value)) { + store.channel(i).queue_amend(s, value); + ++filled; + } + } + } + model_.stats().backfilled.fetch_add(filled, 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) { diff --git a/src/acquisition.hpp b/src/acquisition.hpp index 7235376..0e71cae 100644 --- a/src/acquisition.hpp +++ b/src/acquisition.hpp @@ -61,6 +61,12 @@ class Acquisition { void reserve_hole(std::uint32_t missing, const Partition& part, const std::vector& channels, BufferStore& store, u64 g); + /// Writes a late frame's values into the NaN-filled holes their counter addresses, + /// amending the value ring directly (producer path) and queuing pyramid amends for + /// `ChannelBuffer::acquire()` to apply on the consumer thread. + void backfill_from(const udps_frame_t* frame, const Partition& part, + const std::vector& channels, BufferStore& store, u64 g); + ScopeModel& model_; Config cfg_; udps_client* client_ = nullptr; diff --git a/src/channel.cpp b/src/channel.cpp index 7793e2b..e486d07 100644 --- a/src/channel.cpp +++ b/src/channel.cpp @@ -31,41 +31,50 @@ ChannelBuffer::ChannelBuffer(std::shared_ptr> void ChannelBuffer::acquire() noexcept { const u64 produced = values_->produced(); - if (produced <= published_) { + if (produced > published_) { + // A producer that lapped this ring -- a long trigger wait, or a store that was + // not being published for a while -- makes the incremental append meaningless: + // the samples between `published_` and `produced_` are no longer in the ring. + // Rebuilding from the retained window is O(capacity) once, and it is the only + // way the published window stays coherent (an empty or garbled window is what + // "the trace disappeared" looks like from outside). // Nothing new; still allow the window to slide if another reader dropped it. - return; + if (produced - published_ > values_->capacity()) { + published_ = 0; + pyramid_.clear(); + } + f32 chunk[kChunk]; + for (u64 i = published_; i < produced;) { + const u64 take = std::min(kChunk, produced - i); + values_->copy_from(i, take, chunk); + for (u64 k = 0; k < take; ++k) pyramid_.append(chunk[k]); + i += take; + } + published_ = produced; + // The pyramid's window follows the ring's: sample index `base` is the oldest + // one still retained. drop_before() is monotonic and cheap. + pyramid_.drop_before(values_->base()); + // And the window's first stamp with it, here and nowhere else (see the member's note): the + // window is pinned by the line above, so its origin is a fixed sample, and this is the one + // place in the object's life where that sample's slot is safe to read. + if (pyramid_.size() > 0) window_x0_ = time_of(absolute(0)); } - // A producer that lapped this ring -- a long trigger wait, or a store that was - // not being published for a while -- makes the incremental append meaningless: - // the samples between `published_` and `produced_` are no longer in the ring. - // Rebuilding from the retained window is O(capacity) once, and it is the only - // way the published window stays coherent (an empty or garbled window is what - // "the trace disappeared" looks like from outside). - if (produced - published_ > values_->capacity()) { - published_ = 0; - pyramid_.clear(); + // Backfills, applied after the append pass above: an amend is only ever queued for a slot + // the value ring already produced, so the append has reached it and the pyramid's bounds + // test will accept it. Unconditional, outside the `produced > published_` guard -- a late + // packet can land in a frame interval that pushed nothing new, and the plan's placement + // behind the old early return would have left the pyramid stale until the next push. + if (!pending_amends_.empty()) { + for (const Amend& a : pending_amends_) pyramid_.amend(a.absolute, a.value); + pending_amends_.clear(); } - f32 chunk[kChunk]; - for (u64 i = published_; i < produced;) { - const u64 take = std::min(kChunk, produced - i); - values_->copy_from(i, take, chunk); - for (u64 k = 0; k < take; ++k) pyramid_.append(chunk[k]); - i += take; - } - published_ = produced; - // The pyramid's window follows the ring's: sample index `base` is the oldest - // one still retained. drop_before() is monotonic and cheap. - pyramid_.drop_before(values_->base()); - // And the window's first stamp with it, here and nowhere else (see the member's note): the - // window is pinned by the line above, so its origin is a fixed sample, and this is the one - // place in the object's life where that sample's slot is safe to read. - if (pyramid_.size() > 0) window_x0_ = time_of(absolute(0)); } void ChannelBuffer::invalidate() noexcept { published_ = 0; window_x0_ = 0.0; pyramid_.clear(); + pending_amends_.clear(); } f64 ChannelBuffer::time_of(u64 absolute_index) const noexcept { diff --git a/src/channel.hpp b/src/channel.hpp index 8ce424f..1d741c0 100644 --- a/src/channel.hpp +++ b/src/channel.hpp @@ -49,6 +49,17 @@ class ChannelBuffer final : public plot::data::Series { /// Consumer thread: forgets the published window (after a rebind or a reset). void invalidate() noexcept; + /// Producer thread, under `ScopeModel::mutex()`: records that absolute sample `absolute` + /// has been backfilled, for `acquire()` to fold into the pyramid. + /// + /// The pyramid is mutated *only* on the consumer thread, because a bucket is a 32-byte + /// non-atomic aggregate the renderer reads unlocked at paint time: amending it from the + /// producer would let a reader see a half-updated bucket whose `sum` did not match its + /// `count`, which is a wrong mean out of a structure whose readers trust `count == 0` to + /// mean absent. The value ring is amended by the producer, which owns its writes; the + /// pyramid waits for here. + void queue_amend(u64 absolute, f32 v) { pending_amends_.push_back({absolute, v}); } + void set_nominal_dt(f64 dt) noexcept { dt_ = dt; } /// Re-bases the x axis: `x0()`/`x_at()` become `epoch + timestamp`. Used to make /// the axis read seconds from the trigger event instead of from the stream. @@ -100,6 +111,13 @@ class ChannelBuffer final : public plot::data::Series { /// Timestamp of channel sample `absolute_index`, in seconds since `epoch_`. f64 time_of(u64 absolute_index) const noexcept; + struct Amend { + u64 absolute; + f32 value; + }; + /// Backfills the producer has written into the value ring and the pyramid has not seen yet. + std::vector pending_amends_; + std::shared_ptr> values_; std::shared_ptr> times_; u64 stride_ = 1;