Martino Ferrari
761615c2fe
Going live from a file view left the display frozen for seconds: the per-second line showed
`ring=[2.5200..2.5200] n=567000 -> n=1131000 -> n=1689000` — the sample count climbing at the stream's rate
while the retained span was one value, the file view's last stamp. The status bar was green because
everything about the *connection* was fine.
It is the poisoning `invariants-data-path.md` already describes, entered through a different door. The
new client's stream clock sits *behind* the stamps the old stream left in `Acquisition::last_stamp_` (the
file view's samples are still in the store), so the advance test rejects every packet of a perfectly good
stream, four rejections abandon the declared time signal, and the arrival-clock fallback derives its epoch
from the surviving mark — every sample stamped flat at the old value. The guard did exactly what it was
built to do; the *stream* was the thing that had changed and nothing had told it.
So `Acquisition::start()` — which is the one place a client (re)starts — now says so:
`ScopeModel::reset_stream()` drops the previous stream's history (`store_->reset()`, which the data-path
invariants require to leave nothing behind) and moves `plan_generation_`, which is what the producer
already watches to drop the state tied to the old store; and `last_stamp_` is re-based to `-infinity` so
the monotonic guard re-bases on the new clock instead of pinning it. The declaration is kept: the first
packet of the new stream passes, and monotonicity holds from there.
One self-inflicted crash on the way, fixed before this: `reset_stream()` dereferenced `store_`, which is
**null until the first CONFIG has planned anything** — at start-up there is none — so the app died with
SIGSEGV and no output at all. Caught by running it plainly rather than trusting the change.
Verified with the events probe (a replay and a group, then *Live*):
live restored (after replay): … pending=0
trig engine: … history=0.4377/0.4000 s now_t=0.298 ring=[-0.1392..0.2985] n=569000
trig engine: … history=0.8723/0.4000 s now_t=0.733 ring=[-0.1392..0.7331] n=1134000 (pending=1: it latched)
trig engine: … history=1.2954/0.4000 s now_t=1.156 ring=[-0.1392..1.1562] n=1684000
`ring` moves within a round-trip, `history` fills, and the trigger latches the frame after the restore;
`make test-ui` 14 files, zero `Gtk-CRITICAL`, and the plain run is clean. Docs: the restart's own door into
the poisoning in `invariants-data-path.md`, and the `ring=` reading's entry in `probe-readings.md` now says
what it must do after *Go live*.
1116 lines
56 KiB
C++
1116 lines
56 KiB
C++
// SPDX-License-Identifier: MIT
|
|
// SPDX-FileCopyrightText: 2026 Martino Ferrari
|
|
|
|
#include "acquisition.hpp"
|
|
|
|
#include <limits>
|
|
|
|
#include <udps_client.h>
|
|
|
|
#include "layout_store.hpp"
|
|
|
|
#include <glib/gstdio.h>
|
|
|
|
#include <arpa/inet.h>
|
|
#include <netinet/in.h>
|
|
|
|
#include <algorithm>
|
|
#include <cmath>
|
|
#include <cstring>
|
|
#include <stdexcept>
|
|
#include <string>
|
|
|
|
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() {
|
|
// A started -- or *restarted* -- client is a new stream. Going live from a file view is the restart that
|
|
// matters: the store still holds the event's samples, and the new stream's clock can sit behind their
|
|
// stamps, so the guard below has to re-base rather than clamp. `ring=[2.5200..2.5200]` with `n` climbing
|
|
// was the symptom -- every packet pinned flat at the file view's last stamp.
|
|
model_.reset_stream();
|
|
last_stamp_ = -std::numeric_limits<double>::infinity();
|
|
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<std::uint32_t>(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<Acquisition*>(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<SignalInfo> 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<int>(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<std::uint32_t>(time_signal), k, 0)
|
|
: udps_frame_value(frame, static_cast<std::uint32_t>(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<int>(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<std::uint32_t>(time_signal), k, 0)
|
|
: udps_frame_value(frame, static_cast<std::uint32_t>(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<int>(model_.signals().size())) {
|
|
const std::string& unit = model_.signals()[static_cast<std::size_t>(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<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;
|
|
|
|
// 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<int>(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<int>(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;
|
|
// `tick % g == g - 1` as a mask. `plan_partition()` doubles `time_stride` from 1, so `g` is
|
|
// a power of two -- but it is a *runtime* value, so the modulo compiled to a real 64-bit
|
|
// division rather than the cheap test its comment there claims. See `acc_mask_` for the
|
|
// per-channel form and for what that cost on the hot loop.
|
|
const u64 g_mask = g - 1;
|
|
|
|
// 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(""));
|
|
// The group boundaries, as masks, for the per-sample loops below. Here rather than in
|
|
// them: `g` and the strides are properties of the partition, and a partition change is
|
|
// exactly what this block already reacts to.
|
|
acc_mask_.resize(channels.size());
|
|
for (std::size_t i = 0; i < channels.size(); ++i) {
|
|
acc_mask_[i] = g * std::max<u64>(1, part.stride[i]) - 1;
|
|
}
|
|
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<unsigned long long>(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<int>(channels.size())
|
|
? channels[static_cast<std::size_t>(trig_channel)].name
|
|
: kNoTriggerChannel;
|
|
engine.configure(tc, trig_name, post_want, event_span - post_want, oldest_s,
|
|
model_.partition().slot_dt());
|
|
// 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<unsigned long long>(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<unsigned long long>(snap ? snap->stamps.size() : 0),
|
|
frame->recv_time, static_cast<unsigned long long>(tick_),
|
|
std::chrono::duration<double>(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_++;
|
|
// The shared clock's group boundary, read once and reused by the time ring, the channel
|
|
// loop's own per-channel test and the math push: three tests of the same thing.
|
|
const bool group_end = (tick & g_mask) == g_mask;
|
|
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 (group_end) {
|
|
store.time().push(time_acc_ / static_cast<double>(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<f64>* abs = store.abs_time(); abs != nullptr) {
|
|
abs->push(abs_time_acc_ / static_cast<double>(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<std::uint32_t>(ch.signal), 0,
|
|
std::min<std::uint32_t>(k, ch.elements - 1))
|
|
: udps_frame_value(frame, static_cast<std::uint32_t>(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 mask = acc_mask_[i];
|
|
if ((tick & mask) != mask) continue;
|
|
const f32 value =
|
|
static_cast<f32>(value_acc_[i] / static_cast<double>(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<int>(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 (group_end) 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<std::int64_t>(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 ring=[%.4f..%.4f] n=%zu\n",
|
|
trig_channel, int(engine.armed()), int(engine.pending()),
|
|
engine.post_elapsed(), engine.post_seconds(),
|
|
static_cast<unsigned long long>(engine.post_accum()), engine.history(),
|
|
engine.pre_seconds(), tc.pretrigger_pct, engine.last_value(),
|
|
engine.last_t(), engine.holdoff_until(),
|
|
// The store's own retained span, beside the engine's view of the clock. Going live
|
|
// restarts the client, and the question a stalled display raises is whether the *time
|
|
// ring* is moving at all: `n` climbing while `ring` stands still is a clock problem, and
|
|
// `ring` moving while `now_t` does not is a feed problem. Neither can be told from the
|
|
// engine's own fields.
|
|
store.time().size() > 0 ? double(store.time()[0]) : 0.0,
|
|
store.time().size() > 0 ? double(store.time()[store.time().size() - 1]) : 0.0,
|
|
store.time().size());
|
|
std::fflush(stderr);
|
|
}
|
|
}
|
|
|
|
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();
|
|
// `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;
|
|
// The same group tests as the live loop, written the same way: a hole's samples go through
|
|
// the same accumulators, so the two phases have to agree sample for sample.
|
|
const u64 g_mask = g - 1;
|
|
for (u64 j = 0; j < raw; ++j) {
|
|
const u64 tick = tick_++;
|
|
const bool group_end = (tick & g_mask) == g_mask;
|
|
// 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 (group_end) {
|
|
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 mask = acc_mask_[i];
|
|
if ((tick & mask) != mask) 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 (group_end) {
|
|
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<ChannelInfo>& 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<std::int32_t>(frame->counter - h.counter_first);
|
|
if (into >= 0 && into < static_cast<std::int32_t>(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<std::uint32_t>(channels[i].signal), 0,
|
|
std::min<std::uint32_t>(static_cast<std::uint32_t>(k),
|
|
channels[i].elements - 1))
|
|
: udps_frame_value(frame, static_cast<std::uint32_t>(channels[i].signal),
|
|
static_cast<std::uint32_t>(k), 0);
|
|
acc += raw * channels[i].gain + channels[i].offset;
|
|
}
|
|
const f32 value = static_cast<f32>(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<Acquisition*>(user)->on_config(signals, count, publish_mode);
|
|
}
|
|
|
|
void Acquisition::data_trampoline(const udps_frame_t* frame, void* user) {
|
|
static_cast<Acquisition*>(user)->on_data(frame);
|
|
}
|
|
|
|
void Acquisition::event_trampoline(udps_event_t event, const char* detail, void* user) {
|
|
static_cast<Acquisition*>(user)->on_event(static_cast<int>(event), detail);
|
|
}
|
|
|
|
} // namespace udpscope
|