diff --git a/src/acquisition.cpp b/src/acquisition.cpp index 8f8b3a6..fb006db 100644 --- a/src/acquisition.cpp +++ b/src/acquisition.cpp @@ -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 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(frame->counter - last_counter_) > 0) { + last_counter_ = frame->counter; + have_counter_ = true; + } + const Partition& part = model_.partition(); const std::vector& 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(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(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& 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* 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::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(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) { diff --git a/src/acquisition.hpp b/src/acquisition.hpp index 7d044cf..7235376 100644 --- a/src/acquisition.hpp +++ b/src/acquisition.hpp @@ -12,6 +12,7 @@ #include #include #include +#include #include #include @@ -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& 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 stamps_; std::vector abs_stamps_; /// The decimation filter's state: a running sum per channel (and one for the shared diff --git a/src/model.cpp b/src/model.cpp index bd52978..9b7ad98 100644 --- a/src/model.cpp +++ b/src/model.cpp @@ -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) { diff --git a/src/model.hpp b/src/model.hpp index afeae87..dc38347 100644 --- a/src/model.hpp +++ b/src/model.hpp @@ -167,6 +167,24 @@ class BufferStore { std::vector> 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 packets{0}; @@ -175,6 +193,8 @@ struct StreamStats { std::atomic reconnects{0}; std::atomic connected{false}; std::atomic have_config{false}; + std::atomic holes{0}; ///< reserved invalidated spans + std::atomic 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& 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 store_; + std::vector holes_; TriggerConfig trigger_{}; StreamStats stats_{}; f64 epoch_ = 0.0;