// SPDX-License-Identifier: MIT // SPDX-FileCopyrightText: 2026 Martino Ferrari #include "acquisition.hpp" #include #include "layout_store.hpp" #include #include #include #include #include #include #include #include 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; info.unit = s.unit; info.type_code = s.type_code; info.quant_type = s.quant_type; info.rows = s.num_rows; info.cols = s.num_cols; info.range_min = s.range_min; info.range_max = s.range_max; info.time_mode = s.time_mode; info.sampling_rate = s.sampling_rate; info.time_signal_idx = s.time_signal_idx; return info; } } // namespace Acquisition::Acquisition(ScopeModel& model, const Config& cfg) : model_(model), cfg_(cfg) {} Acquisition::~Acquisition() { stop(); } void Acquisition::start() { udps_client_config_t c; udps_client_config_init(&c); c.server_addr = cfg_.server.c_str(); c.server_port = cfg_.port; c.multicast_group = cfg_.multicast.empty() ? nullptr : cfg_.multicast.c_str(); c.interface_addr = cfg_.interface_addr.empty() ? nullptr : cfg_.interface_addr.c_str(); c.recv_buffer_bytes = static_cast(cfg_.recv_buffer_bytes); c.silence_timeout_s = 2.0; c.reconnect_delay_s = 0.5; c.keepalive_interval_s = 1.0; client_ = udps_client_create(&c); if (client_ == nullptr) { throw std::runtime_error("could not create the UDPS client for " + cfg_.server + ":" + std::to_string(cfg_.port)); } udps_client_set_callbacks(client_, &Acquisition::config_trampoline, &Acquisition::data_trampoline, &Acquisition::event_trampoline, this); udps_client_set_datagram_cb(client_, &Acquisition::datagram_trampoline, this); // `--capture-packets`: armed before the first datagram, so the first CONNECT and // its CONFIG -- the hardest strange behaviours to reproduce -- are in the ring. // The open itself still happens on the producer thread, at the first datagram. if (cfg_.capture_packets > 0) { set_capture_want(true, cfg_.capture_packets); } started_ = std::chrono::steady_clock::now(); running_.store(true, std::memory_order_release); thread_ = std::thread([this] { run(); }); } void Acquisition::stop() { if (!running_.exchange(false, std::memory_order_acq_rel)) { if (client_ != nullptr) { udps_client_destroy(client_); client_ = nullptr; } return; } if (thread_.joinable()) thread_.join(); if (client_ != nullptr) { udps_client_destroy(client_); client_ = nullptr; } } void Acquisition::run() { while (running_.load(std::memory_order_acquire)) { udps_client_poll(client_, 50); udps_stats_t s; udps_client_stats(client_, &s); StreamStats& out = model_.stats(); out.packets.store(s.packets_received, std::memory_order_relaxed); out.frames.store(s.frames_delivered, std::memory_order_relaxed); out.gaps.store(s.counter_gaps, std::memory_order_relaxed); out.resequenced.store(s.late_resequenced, std::memory_order_relaxed); out.late.store(s.late_delivered, std::memory_order_relaxed); out.stale.store(s.stale_packets, std::memory_order_relaxed); out.overflows.store(s.reorder_overflows, std::memory_order_relaxed); out.reconnects.store(s.reconnects, std::memory_order_relaxed); out.connected.store(udps_client_is_connected(client_) != 0, std::memory_order_relaxed); } } void Acquisition::set_capture_want(bool want, std::uint32_t slots) { if (want) { if (slots > 0) capture_slots_ = slots; if (capture_path_.empty()) { // Keyed on the source: one path per source, the previous capture kept one deep. std::string host = cfg_.server; for (char& ch : host) { if (ch == ':' || ch == '/' || ch == ' ') ch = '_'; } std::string dir = cache_dir() + "/" + host + "-" + std::to_string(cfg_.port); g_mkdir_with_parents(dir.c_str(), 0700); capture_path_ = dir + "/capture.bin"; std::uint32_t addr = 0; struct ::in_addr a; if (::inet_pton(AF_INET, cfg_.server.c_str(), &a) == 1) { addr = a.s_addr; // already network order, which is what the header wants } capture_.set_peer(addr, cfg_.port); } } capture_want_.store(want, std::memory_order_release); } void Acquisition::on_datagram(const std::uint8_t* bytes, std::size_t len, double recv_time) { const bool want = capture_want_.load(std::memory_order_acquire); if (want && !capture_.enabled()) { // A user-initiated act that happens once per toggle, on this thread: a rename, a // create, a 62.6 MiB reserve and an mmap. It can cost the packets in flight, and // the measured cost is `capture_.open_ms()`. if (!capture_.open(capture_path_, capture_slots_)) { char err[128]; capture_.error(err, sizeof(err)); std::fprintf(stderr, "[udpscope] packet capture failed to open: %s\n", err); // Stay off rather than retrying every datagram against a full disk; the GTK // thread reads the flag back and returns the button to off. capture_want_.store(false, std::memory_order_release); return; } std::fprintf(stderr, "[udpscope] packet capture on: %s (%u slots, %.1f ms to open)\n", capture_.path().c_str(), capture_.slot_count(), capture_.open_ms()); } else if (!want && capture_.enabled()) { capture_.close(); std::fprintf(stderr, "[udpscope] packet capture off\n"); } if (capture_.enabled()) { capture_.store(bytes, len, recv_time); } } void Acquisition::datagram_trampoline(const std::uint8_t* bytes, std::size_t len, double recv_time, void* user) { static_cast(user)->on_datagram(bytes, len, recv_time); } void Acquisition::on_config(const udps_signal_t* signals, std::uint32_t count, std::uint8_t publish_mode) { std::vector info; info.reserve(count); for (std::uint32_t i = 0; i < count; ++i) info.push_back(to_info(signals[i])); // One line per CONFIG on stderr: what the stream says it is. This is the first // thing to look at when a signal refuses to plot. static const char* const kModes[] = {"strict", "accumulate", "decimate"}; std::fprintf(stderr, "[udpscope] CONFIG: %u signals, publish=%s\n", count, publish_mode < 3 ? kModes[publish_mode] : "?"); for (const SignalInfo& sig : info) { std::fprintf(stderr, "[udpscope] '%s' %ux%u (%u elem) rate=%g Hz time=%s idx=%u unit='%s'\n", sig.name.c_str(), sig.rows, sig.cols, sig.elements(), sig.sampling_rate, sig.time_mode == UDPS_TIME_PACKET ? "packet" : sig.time_mode == UDPS_TIME_FULL_ARRAY ? "full-array" : sig.time_mode == UDPS_TIME_FIRST_SAMPLE ? "first-sample" : "last-sample", sig.time_signal_idx, sig.unit.c_str()); } std::fflush(stderr); // The warn-once latches are per *connection*, not per process. A reconnect re-enters here, and // a diagnostic that goes quiet after the first occurrence of a condition that recurs on the // next session is the failure mode worth avoiding in a feature whose whole job is diagnosis. warned_batch_ = false; warned_outage_ = false; warned_bridge_ = false; warned_time_source_ = false; // The clock choice is per connection too: a reconnect re-derives it from the frames // the new session actually delivers, rather than trusting a decision made about a // session that dropped. time_source_ = 0; reset_session_state(); model_.stats().have_config.store(true, std::memory_order_relaxed); model_.set_signals(std::move(info)); } void Acquisition::on_event(int event, const char* detail) { (void)detail; if (event == UDPS_EVENT_CONNECTED) { model_.stats().connected.store(true, std::memory_order_relaxed); } else if (event == UDPS_EVENT_DISCONNECTED) { model_.stats().connected.store(false, std::memory_order_relaxed); // The session dropped mid-flight: everything the producer knows about *that* // session's numbering and clocks is now a lie about the next one. Clearing it here // is what makes the next frame re-anchor instead of reading as late, as a glitch or // as an outage -- and nothing here touches the buffer, whose history survives the // drop (see `ScopeModel::set_signals`'s unchanged-CONFIG path). reset_session_state(); } } void Acquisition::reset_session_state() noexcept { have_counter_ = false; last_counter_ = 0; have_stream_anchor_ = false; last_stream_anchor_ = 0.0; stamp_failures_ = 0; // The arrival clock's *anchor* stays -- stamps are small seconds the ring's timeline // is already scaled against, and re-anchoring mid-stream would move x -- but the // smoothed interval and the previous arrival are per-connection measurements, and // carrying them across a multi-second outage makes the first new frame read as a // clock fault rather than as the resumption it is. have_arrival_interval_ = false; arrival_interval_ = 0.0; arrival_prev_ = 0.0; } std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_signal) { // The frame's span in base ticks: a scalar signal carries `num_samples` values // (one per batched cycle), a block signal (rows*cols > 1) carries its whole // element block. The largest of those is how many base ticks this frame spans -- // in `strict` mode `num_samples` is 1 while a 1x1000 block carries 1000 samples. std::uint32_t n = frame->num_samples; for (std::uint32_t i = 0; i < frame->num_signals; ++i) { n = std::max(n, udps_signal_num_elements(&frame->signals[i])); } stamps_.assign(n, 0.0); abs_stamps_.assign(n, 0.0); // Samples per packet, needed in both modes: relative stamping must advance `slot_base_` // past the reserved hole before placing this frame's own stamps, and `on_data` must size // the hole before calling this function -- both operations require knowing how many samples // a missing packet represents, which is only knowable from the first frame that arrives. if (batch_ == 0) { batch_ = n; if (cfg_.relative_time) { std::fprintf(stderr, "[udpscope] relative time: batch=%u samples/packet, rate=%.6g S/s declared\n", batch_, model_.partition().base_dt > 0.0 ? 1.0 / model_.partition().base_dt : 0.0); std::fflush(stderr); } } else if (n != batch_) { // A frame whose length disagrees with the first one is an inconsistency in the // producer, not a loss: the counter still names one packet. Warn once and keep the // established batch, which is only used to size holes -- the timeline itself is built // from the lengths actually seen, so it stays exact for this frame either way. if (!warned_batch_) { warned_batch_ = true; std::fprintf(stderr, "[udpscope] warning: packet carries %u samples but the stream's first " "carried %u; keeping %u as the packet length a lost packet is worth.\n", n, batch_, batch_); std::fflush(stderr); } } // Relative mode: the stamp is a pure function of position and the *declared* rate. Nothing // below this block runs -- not the arrival-cadence override, not the degeneracy guard, not // the glitch bridge, not the monotonic clamp's reason for existing -- because a position // cannot jitter, cannot fail to advance and cannot jump. if (cfg_.relative_time) { 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; } slot_base_ += n; // The absolute clock is still read, for export only. When the stream carries no usable // time signal there is no absolute clock to record and the relative stamp is the honest // answer -- an export then says what the counter says, which is all anyone had. if (time_signal >= 0 && time_signal < static_cast(frame->num_signals)) { const udps_signal_t& ts = frame->signals[time_signal]; const bool scalar_time = udps_signal_num_elements(&ts) == 1; for (std::uint32_t k = 0; k < n; ++k) { abs_stamps_[k] = scalar_time ? udps_frame_value(frame, static_cast(time_signal), k, 0) : udps_frame_value(frame, static_cast(time_signal), 0, k); } } else { for (std::uint32_t k = 0; k < n; ++k) abs_stamps_[k] = stamps_[k]; } return n; } double period = model_.partition().base_dt; if (!(period > 0.0)) period = 1.0e-3; // The cadence of the samples actually stored: the partition's slot interval (base // tick times the clock's decimation), as *declared* by the CONFIG. const double slot = model_.partition().slot_dt(); double cadence = slot > 0.0 ? slot : period; // A declared rate is a claim the stream may not keep (a CONFIG declaring 1 MS/s // while delivering a hundred samples every fifth of a second is the case that // exposed this): stamps spaced by the declaration then span microseconds inside a // frame and jump by the frame interval between frames, which reads as a staircase // in the x column and, in an export, as data that is "downsampled" while the // values are fine. When the arrivals disagree with the declaration, the arrivals // are the truth -- the mean spacing of the samples that actually arrived. if (have_arrival_interval_ && n > 0) { const double measured = arrival_interval_ / double(n); const bool agrees = cadence > 0.0 && measured >= 0.75 * cadence && measured <= 1.25 * cadence; if (measured > 0.0 && std::isfinite(measured) && !agrees) { cadence = measured; } } bool used_stream_time = false; // Once the stream has been judged to have no usable time base, it keeps the // arrival clock for the rest of its life (until a CONFIG says otherwise): letting // the decision flip per frame is what interleaved two timelines in one ring. if (time_source_ != 2 && time_signal >= 0 && time_signal < static_cast(frame->num_signals)) { const udps_signal_t& ts = frame->signals[time_signal]; // How the client indexes a signal's values depends on its shape // (`udps_frame_value`): a scalar carries one value per batched cycle, so the // stamp of sample k is (sample=k, elem=0); a multi-element signal carries one // *sample* of `elements` values per packet, so the stamp of sample k is // (sample=0, elem=k). Passing the scalar form for both is why FULL_ARRAY time // signals -- the only mode that carries a stamp per element -- read as // element 0 for every sample and were then judged degenerate. const bool scalar_time = udps_signal_num_elements(&ts) == 1; auto value_of = [&](std::uint32_t k) { return scalar_time ? udps_frame_value(frame, static_cast(time_signal), k, 0) : udps_frame_value(frame, static_cast(time_signal), 0, k); }; switch (ts.time_mode) { case UDPS_TIME_FULL_ARRAY: for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = value_of(k); used_stream_time = true; break; case UDPS_TIME_FIRST_SAMPLE: { // Only element 0 is meaningful here: the producer stamps the batch's first // sample and says nothing about the rest, so the cadence has to space them -- // exactly as LAST_SAMPLE below dates its batch backwards from its anchor. // Deriving a step from `value_of(n - 1)` instead cannot work: a scalar signal // in Accumulate mode still occupies `num_samples` slots on the wire, and a // producer fills them by repeating the anchor, so the step came out as zero // and every sample of the frame landed on one x. The degeneracy guard below // then threw the whole stream onto the arrival clock. const double anchor = value_of(0); for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = anchor + double(k) * cadence; used_stream_time = true; break; } case UDPS_TIME_LAST_SAMPLE: { const double anchor = value_of(n > 0 ? n - 1 : 0); for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = anchor - double(n - 1 - k) * cadence; used_stream_time = true; break; } default: break; } } // A stream-time signal may be declared in a time unit rather than seconds ('ns' is // the common one). The scale has to be known *here*, before the checks below, because // they weigh the stamps' own spacing against the declared cadence, which is in // seconds: on a 'ns' stream an unscaled comparison is off by a factor of 1e9, so // every packet would read as an implausible jump. double unit_scale = 1.0; if (time_signal >= 0 && time_signal < static_cast(model_.signals().size())) { const std::string& unit = model_.signals()[static_cast(time_signal)].unit; if (unit == "ns") unit_scale = 1e-9; else if (unit == "us" || unit == "\u00b5s" || unit == "\xce\xbc" "s") unit_scale = 1e-6; else if (unit == "ms") unit_scale = 1e-3; } if (used_stream_time && n > 0) { // 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 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. constexpr unsigned kFailureLimit = 4; // A time signal that does not actually vary -- a vector time signal, whose // values the client reads one element at a time, or a producer that leaves it // zero -- would put every sample at one x and collapse the axis to a point. // The arrival estimate is a worse clock but a usable one; a degenerate one is // not a time base at all. const char* fault = nullptr; double lo = stamps_[0]; double hi = stamps_[0]; if (!std::isfinite(stamps_[0])) fault = "carries a stamp that is not finite"; for (std::uint32_t k = 1; fault == nullptr && k < n; ++k) { if (!std::isfinite(stamps_[k])) { fault = "carries a stamp that is not finite"; break; } lo = std::min(lo, stamps_[k]); hi = std::max(hi, stamps_[k]); } if (fault == nullptr && !(hi > lo)) fault = "carries no span (every stamp is equal)"; if (fault == nullptr && have_stream_anchor_) { // The stamps must also ADVANCE across packets, and by a plausible amount. A // producer that restarts its stamps every packet (or a signal mislabelled as // time) keeps handing back the same values; the max-clamp would then pin every // sample at one x, and the trigger's hold-off clock with it -- the engine waited // forever and no second event ever fired. if (!(stamps_[n - 1] > last_stream_anchor_)) { fault = "does not advance across packets"; } else if (stamps_[0] - last_stream_anchor_ > kGlitchPackets * double(n) * step) { fault = "jumped implausibly far ahead of the previous packet"; } } if (fault == nullptr) { stamp_failures_ = 0; last_stream_anchor_ = stamps_[n - 1]; have_stream_anchor_ = true; } else if (have_stream_anchor_ && ++stamp_failures_ < kFailureLimit) { // A short run of bad packets is bridged rather than believed: the stamps are // replaced by a continuation of the accepted timeline at the declared cadence, // so neither the monotonic mark nor the anchor is poisoned, and the next good // packet -- which carries on from where the true timeline was -- is accepted // normally. Believing one glitching packet instead is what condemned the *next*, // perfectly good one and threw the whole stream onto the arrival clock. const double from = last_stream_anchor_; for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = from + double(k + 1) * step; last_stream_anchor_ = stamps_[n - 1]; if (!warned_bridge_) { warned_bridge_ = true; std::fprintf(stderr, "[udpscope] warning: the stream's time signal %s (%g..%g against a " "previous %g); bridging that packet at the declared cadence and " "keeping the stream clock.\n", fault, lo, hi, from); } } else { used_stream_time = false; // not a time base; fall back to arrival if (!warned_time_source_) { warned_time_source_ = true; std::fprintf(stderr, "[udpscope] warning: the stream's time signal %s (%g..%g); timing " "samples by arrival instead. Check the time signal's mode and index " "in the CONFIG dump above.\n", fault, lo, hi); } } } if (!used_stream_time) { // Arrival-anchored: the packet is assumed to have arrived as its last element // was produced, and the earlier ones are dated backwards. A gap (`lost`) // shows up as extra elapsed time between arrivals, which is exactly the // spacing the samples had. const double end = frame->recv_time; if (!have_arrival_epoch_) { arrival_epoch_ = end; have_arrival_epoch_ = true; } else { const double elapsed = end - arrival_prev_; if (elapsed > 0.0 && std::isfinite(elapsed)) { // Smoothed: the socket clock jitters by a millisecond or so per frame, and // an unsmoothed step would make every frame's span disagree slightly. arrival_interval_ = have_arrival_interval_ ? 0.8 * arrival_interval_ + 0.2 * elapsed : elapsed; have_arrival_interval_ = true; } } arrival_prev_ = end; const double base = end - arrival_epoch_; // The spacing between two *stored* samples is the partition's own cadence -- // the exact interval the ingest filter decimates to, which is `slot_dt()` and // not the base tick. Deriving it from the arrival gap divided by // `n * (lost + 1)` collapsed the step to ~zero whenever the client reported // losses, and a whole frame then shared one f32 instant: the x column became a // staircase (many rows at one time, then a jump) while the values were fine -- // which is what an export of it looks like, and what the plot's crosshair and // trigger times were reading. The arrival clock still anchors the frame's LAST // sample, so a lost packet or a stalled producer still shows up as a jump // *between* frames. // The frame's stamps start at the later of two anchors: where the arrival clock // puts the frame's first sample, and one cadence past the previous frame's last. // The second is what keeps the timeline strictly increasing -- anchoring the // frame's *end* at the arrival let a cadence estimate even slightly long reach // back past the previous end, and the monotonic clamp then flattened the overlap // into equal timestamps (a consumer plotting against them draws a vertical line // there, and an export reads as if sampled twice at one instant). A gap still // shows: when the arrivals run ahead, the first anchor wins and the frame jumps. const double base_rel = base - model_.epoch(); const double start = std::max(base_rel - double(n - 1) * cadence, last_stamp_ + cadence); for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = start + double(k) * cadence + model_.epoch(); } // A change of time source invalidates the running mark: the stream-time branch // stamps in the time signal's own values and the arrival branch in seconds since // the first packet, and the two timelines share no zero. Leaving the mark at the // old source's last value made every arriving frame clamp flat onto it -- which is // the 2.5 s plateau of one timestamp that a run reports when its declared time // signal turns out unusable and the app falls back mid-stream. if (time_source_ == 0) { time_source_ = used_stream_time ? 1 : 2; } else if (time_source_ == 1 && !used_stream_time) { // The declared time signal failed its checks for good: keep the arrival clock from // here on, anchored so that it continues from the monotonic mark. The two timelines // then share a zero *by construction*, which is the whole point of deriving the // epoch from `last_stamp_` -- so the recorded history is still on one timeline and // is kept. It used to be wiped here, on the reasoning that the old stamps belonged // to a dead clock; they do not, and the wipe is what made a single glitching packet // look like the stream restarting from zero. time_source_ = 2; const double end = frame->recv_time; arrival_epoch_ = end - (last_stamp_ + model_.epoch()); arrival_prev_ = end; } // The declared unit, applied last so the x axis is in seconds either way. It applies // ONLY to stream time: the arrival fallback already produces seconds off the socket // clock -- scaling those by the time signal's unit shrank the whole time base by 1e9 // on 'ns' streams, which froze the trigger's hold-off clock and pinned the axis at ~0. if (used_stream_time && unit_scale != 1.0) { for (std::uint32_t k = 0; k < n; ++k) stamps_[k] *= unit_scale; } // Stream mode: `time()` already holds the absolute clock, so the parallel ring repeats it // rather than carrying a second timeline. One rule for the export either way. for (std::uint32_t k = 0; k < n; ++k) abs_stamps_[k] = stamps_[k]; return n; } 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; // 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. int time_signal = -1; { const SignalInfo& sig = model_.signals()[channels[0].signal]; if (sig.time_signal_idx != 0xFFFFFFFFu && sig.time_signal_idx < model_.signals().size()) { time_signal = static_cast(sig.time_signal_idx); } } if (time_signal < 0) { // No channel names a time signal: look for the standalone timestamp signal // (full-array, more than one element, no declared rate). for (std::size_t i = 0; i < model_.signals().size(); ++i) { const SignalInfo& sig = model_.signals()[i]; if (sig.time_mode == UDPS_TIME_FULL_ARRAY && sig.elements() > 1 && sig.sampling_rate == 0.0) { time_signal = static_cast(i); break; } } } const std::uint32_t n = build_stamps(frame, time_signal); if (n == 0) return; if (model_.epoch() == 0.0) { model_.set_epoch_if_unset(stamps_[0]); std::fprintf(stderr, "[udpscope] time base: epoch=%.3f, first frame recv=%.3f, n=%u\n", model_.epoch(), frame->recv_time, n); std::fflush(stderr); } const double epoch = model_.epoch(); TriggerConfig& tc = model_.trigger(); // One ring, always filling, whatever the trigger is doing: the pre-trigger // history has to already be there when an edge arrives, and the first latch after // arming would otherwise open on however little had accumulated since. What // freezes in trigger mode is the *view*, not acquisition -- the latch copies the // event window out into an immutable `EventSnapshot`. BufferStore& store = model_.live(); const u64 g = part.time_stride; // A re-plan replaced the store, so every ring's `produced()` restarted at 0 while `tick_` // kept counting. `backfill_from()` derives a slot index from a hole's `tick_begin`, which is // only an absolute slot index while `tick_ / period == produced()`, so the tick counter has // to restart with the buffer -- otherwise the amend lands outside the ring, repairs nothing, // and the hole is still marked repaired, which is the band saying the opposite of the truth. // The generation is the signal rather than the stride/channel-count comparison below: only // a CONFIG that *changes* the stream re-plans the store -- a reconnect's identical CONFIG // keeps it, history and all (see `ScopeModel::set_signals`). // `slot_base_` is deliberately *not* reset -- it is relative mode's clock origin, samples // since the stream began, and relative time must not jump backwards when the user retypes // the window. if (const unsigned gen = model_.plan_generation(); gen != seen_plan_generation_) { seen_plan_generation_ = gen; tick_ = 0; last_tick_ = ~std::uint64_t{0}; // Zeroing the stride makes the guard below fire too: the accumulators belong to the group // `tick_` was counting through, so they are stale for the same reason. acc_stride_ = 0; } // 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()) { acc_stride_ = g; acc_channels_ = channels.size(); value_acc_.assign(channels.size(), 0.0); value_acc_n_.assign(channels.size(), 0); math_sample_.assign(channels.size(), std::nanf("")); time_acc_ = 0.0; abs_time_acc_ = 0.0; time_acc_n_ = 0; } // 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. // // This block must follow the re-plan above: reserve_hole() runs the missing samples // 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->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); } // Calibration is read per frame; it only changes under the same lock. TriggerEngine& engine = engine_; const int trig_channel = tc.channel; // The post-trigger window straight through in seconds. There used to be a measured, // smoothed samples-per-second estimate here, converting the window into a sample // count for the engine; the estimate is what sized captures wrong. Its floor // (`inst >= 1.0`) turned any stream whose time signal advances in large steps into // 1 S/s, the count came out as 1, and the engine latched on the edge sample itself // -- an event with no post-trigger data, which is exactly the "post window is // empty" defect. The engine is fed stream time per sample anyway, so it can count // the window itself and no conversion has to exist. // // The oldest buffered stamp goes with it, so the engine can refuse an edge the // history does not reach back far enough behind. Read from the ring rather than // timed from the arm: the ring fills with the trigger off too, so the pre window is // usually already there and a timer would delay the first event for nothing. const double event_span = model_.event_span(); const double post_want = tc.post_span_s(event_span); const double oldest_s = store.time().size() > 0 ? double(store.time()[0]) : last_stamp_; // The channel's *name* goes with it, because that is what identifies it across a // rebuild of the channel list (see `TriggerEngine::configure`). static const std::string kNoTriggerChannel; const std::string& trig_name = trig_channel >= 0 && trig_channel < static_cast(channels.size()) ? channels[static_cast(trig_channel)].name : kNoTriggerChannel; engine.configure(tc, trig_name, post_want, event_span - post_want, oldest_s); // The engine's armed state follows the panel's, both ways: without the disarm // half, Disarm left the engine firing (its own `armed_` was only ever set), and // re-entering trigger mode kept the stale prev/edge/hold-off state. if (tc.enabled && tc.armed) { if (!engine.armed()) { engine.arm(last_stamp_); model_.trigger_armed().store(true, std::memory_order_relaxed); } } else if (engine.armed()) { // Disarm cuts a capture in progress where it got to and keeps it, instead of // discarding it. `engine.disarm()` clears `pending_`, which drops the display out of // `Capturing`; with nothing published the view fell back to the *previous* event, or // to blank if this was the first -- the capture the user was watching simply // vanished. Publishing first means the same snapshot path freezes it, so what is on // screen at the moment of the Disarm stays on screen. // // The full post window is requested and `live_window_locked()` clips it to what the // rings hold, so the snapshot ends at the newest sample and reports its own shorter // `post_s` -- the same way a first latch whose pre window was short reports its own // `pre_s`. No event fired, so `trigger_latches()` is deliberately *not* bumped; the // UI notices the new picture through `event_seq()`. if (engine.pending()) { const double edge = engine.last_trigger_time(); model_.publish_event(event_span - post_want, post_want, edge); model_.trigger_time().store(edge, std::memory_order_relaxed); model_.trigger_time_valid().store(true, std::memory_order_relaxed); auto snap = model_.event(); std::fprintf(stderr, "[udpscope] trigger: disarmed mid-capture, edge at %.3f s kept as " "[-%.4f..%.4f] n=%llu (post window asked %.4f s, filled %.4f s)\n", edge, snap ? snap->pre_s : 0.0, snap ? snap->post_s : 0.0, static_cast(snap ? snap->stamps.size() : 0), post_want, engine.post_elapsed()); std::fflush(stderr); } engine.disarm(); model_.trigger_armed().store(false, std::memory_order_relaxed); } double last_stamp_used = last_stamp_; // The latch is handled on the sample it happens on, and with the edge the engine // latched *on* -- after that sample's channels have all been pushed (see the loop). // // It used to set a flag and publish once after the loop, reading the edge back from // the engine. But the loop keeps feeding the engine after a latch, so any further // edge in the rest of the packet -- the hold-off is 10 ms and a packet can be a long // block of samples -- overwrote that edge before it was read. The capture was then // cut around an edge whose post window had not been collected yet, so the event came // out all pre-trigger with the marker and x = 0 at the right-hand end of the data. // Publishing here also means the ring provably holds the post window: the latch // condition is that stream time has passed `edge + post`, and every sample up to // this one has already been pushed. const double span = std::max(1.0e-9, model_.event_span()); const double post_s = tc.post_span_s(span); const auto latch = [&](double edge) { model_.publish_event(span - post_s, post_s, edge); model_.trigger_latches().fetch_add(1, std::memory_order_relaxed); model_.trigger_time().store(edge, std::memory_order_relaxed); model_.trigger_time_valid().store(true, std::memory_order_relaxed); // The captured window is reported from the snapshot itself, not from the request: // `want` versus `got` is what distinguishes "the engine did not wait" from "the // cut lost the data", which is otherwise guesswork from the picture alone. auto snap = model_.event(); std::fprintf(stderr, "[udpscope] trigger: edge at %.3f s, latched at %.3f s of stream time " "(want [-%.4f..%.4f] got [-%.4f..%.4f] n=%llu, frame recv=%.3f, " "ticks_so_far=%llu, wall=%.3f s)\n", edge, last_stamp_, span - post_s, post_s, snap ? snap->pre_s : 0.0, snap ? snap->post_s : 0.0, static_cast(snap ? snap->stamps.size() : 0), frame->recv_time, static_cast(tick_), std::chrono::duration(std::chrono::steady_clock::now() - started_) .count()); std::fflush(stderr); if (tc.mode == TriggerMode::Single) tc.armed = false; }; for (std::uint32_t k = 0; k < n; ++k) { const u64 tick = tick_++; const double rel = std::max(stamps_[k] - epoch, last_stamp_); last_stamp_used = rel; last_stamp_ = rel; // The stored stamp is the *mean* of the group, not its last member: that is where the // 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; 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_)); time_acc_ = 0.0; // The absolute clock on the same slot and under the same group mean, so an export can // pair value, relative stamp and absolute stamp by index with no interpolation. if (plot::data::RingBuffer* abs = store.abs_time(); abs != nullptr) { abs->push(abs_time_acc_ / static_cast(time_acc_n_)); } abs_time_acc_ = 0.0; time_acc_n_ = 0; } // The latch belongs to this sample, but it cannot be published from inside the // channel loop: `publish_event()` clamps the slice to every value ring's // `produced()`, and the channels *after* the trigger channel have not pushed this // sample yet, so the newest sample -- the one the post window was waiting for -- // would be clamped away. One sample is invisible in a 20 000-sample capture (it is // the `n=19999` and the `0.9999` instead of `1.0000` in the probe's output), and at // one sample per second it is the entire post side: `got [-0.9997..0.0000]`, which // reads exactly like the "post window is empty" defect. bool latch_here = false; double latch_edge = 0.0; for (std::size_t i = 0; i < channels.size(); ++i) { const ChannelInfo& ch = channels[i]; // A derived channel is filled by `push_math_sample()` below, once per stored slot, from the // values the streamed ones just pushed. Reading it from the packet would give zero (it has no // signal) and pushing that would advance its value ring twice per time slot -- which is an // alternation between the value and 0 while the buffer is filling, and a complete loss of the // pairing once both rings wrap. if (ch.math) continue; // A signal's columns are consecutive samples of the same channel: a scalar // (one element) carries one value per batched cycle, a block signal carries // the cycle's sample at element `k`. Either way element 0 is the channel. const double raw = ch.elements > 1 ? udps_frame_value(frame, static_cast(ch.signal), 0, std::min(k, ch.elements - 1)) : udps_frame_value(frame, static_cast(ch.signal), k, 0); // Every sample is read and accumulated even when only one in `period` is stored: // that is the whole point of the filter. Skipping the read -- which is what the // old stride did -- is what made decimation alias. value_acc_[i] += raw * ch.gain + ch.offset; ++value_acc_n_[i]; const u64 period = g * part.stride[i]; if (tick % period != period - 1) continue; const f32 value = static_cast(value_acc_[i] / static_cast(value_acc_n_[i])); value_acc_[i] = 0.0; value_acc_n_[i] = 0; store.values(i).push(value); // This slot's value, for the math channels below. Every channel is pushed on the same slot // (per-channel strides are 1 by design), so the vector is complete by the time the loop ends. if (i < math_sample_.size()) math_sample_[i] = value; // The trigger sees the *stored* signal, not the wire signal. An edge found in a // sample that is not in the buffer would be cut from data that does not contain it, // and with the filter in front of it a spike above the stored Nyquist is exactly // the thing that must not latch -- it is not in the record the user would get. if (static_cast(i) == trig_channel) { if (engine.process(value, rel)) { latch_here = true; latch_edge = engine.last_trigger_time(); } } } // The derived channels: once per *stored* slot, and **before** the latch publishes. // `publish_event()` clamps its snapshot to every value ring's `produced()`, so a derived // channel pushed after it would be missing its newest sample from the capture -- the same // one-sample clamp the deferred latch above exists for. if (tick % g == g - 1) model_.push_math_sample(math_sample_); if (latch_here) latch(latch_edge); } last_stamp_ = last_stamp_used; model_.trigger_armed().store(engine.armed(), std::memory_order_relaxed); // The edge goes out before the flag that makes the UI read it (release/acquire on // `trigger_pending`), so a tick that sees a capture open sees which edge it is for. A // capture that opens and latches inside this one frame never reports pending, and that // is right: the display goes straight to the frozen event. if (engine.pending()) model_.set_capture_edge(engine.last_trigger_time()); model_.set_trigger_pending(engine.pending()); static std::int64_t next_engine_log = 0; const std::int64_t now_e = static_cast(frame->recv_time * 1.0e6); if (now_e >= next_engine_log) { next_engine_log = now_e + 1000000; // `elapsed/post_want` is the capture's progress in the only unit that matters now // (stream seconds); `accum` is how many samples that window has collected, which // is what says whether a capture is thin because the window is short or because // the stream is slow. std::fprintf(stderr, "[udpscope] trig engine: ch=%d armed=%d pending=%d elapsed=%.4f/%.4f s " "accum=%llu history=%.4f/%.4f s pre_pct=%.0f v=%.4g now_t=%.3f " "holdoff_until=%.3f\n", trig_channel, int(engine.armed()), int(engine.pending()), engine.post_elapsed(), engine.post_seconds(), static_cast(engine.post_accum()), engine.history(), engine.pre_seconds(), tc.pretrigger_pct, engine.last_value(), engine.last_t(), engine.holdoff_until()); std::fflush(stderr); } } 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(); // `lost` counts the packets missing *immediately before* the frame that revealed them // (see `udps_frame_t::lost`), and the counter mark at the top of `on_data` has already // moved `last_counter_` to that frame's own counter -- so the hole spans // [counter - missing, counter). This read `last_counter_ + 1u`, which addressed a span // one past the revealing frame and `missing + 1` too high: no late packet's counter could // fall in it, so every backfill lookup missed and `backfilled` read 0 against 154 holes. h.counter_first = last_counter_ - missing; 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); } 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. // By signed delta, the idiom the client uses throughout: the counter is a wrapping uint32, so // a raw `>=`/`<` pair matches nothing for a hole straddling the 2^32 wrap and every one of its // late packets would be dropped on the next line. Hole* hole = nullptr; for (Hole& h : model_.holes()) { const std::int32_t into = static_cast(frame->counter - h.counter_first); if (into >= 0 && into < static_cast(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; } } } // This packet's share of the hole has been delivered, whatever the ring still retains of it. // Marked on *arrival* and not on `filled`: a decimation coarse enough that no whole slot lies // inside one packet repairs no slot at all, and keying off `filled` would leave such a hole // permanently unrepairable -- a solid band over data that is all there, which is the exact // defect this task closes. ++hole->filled_packets; 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) { static_cast(user)->on_config(signals, count, publish_mode); } void Acquisition::data_trampoline(const udps_frame_t* frame, void* user) { static_cast(user)->on_data(frame); } void Acquisition::event_trampoline(udps_event_t event, const char* detail, void* user) { static_cast(user)->on_event(static_cast(event), detail); } } // namespace udpscope