Martino FerrariandClaude Opus 4.6
84c158ff21
`plan_partition()` only ever chooses powers of two for `time_stride`, and the comment there already claimed the test was cheap -- but the stride is a runtime value, so `tick % period` compiled to a real division that the hot loop paid once per channel per wire sample. The per-channel masks are built where the partition is already reacted to, and the shared clock's group end is computed once per sample instead of three times. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
897 lines
41 KiB
C++
897 lines
41 KiB
C++
// SPDX-License-Identifier: MIT
|
|
// SPDX-FileCopyrightText: 2026 Martino Ferrari
|
|
|
|
#include "model.hpp"
|
|
|
|
#include <udps_client.h>
|
|
|
|
#include <algorithm>
|
|
#include <chrono>
|
|
#include <cmath>
|
|
#include <cstdio>
|
|
#include <stdexcept>
|
|
#include <string>
|
|
#include <unordered_map>
|
|
|
|
namespace udpscope {
|
|
namespace {
|
|
|
|
u64 pow2_floor(u64 v) {
|
|
if (v == 0) return 0;
|
|
u64 p = 1;
|
|
while (p <= v / 2) p <<= 1;
|
|
return p;
|
|
}
|
|
|
|
std::uint64_t now_ns() {
|
|
return static_cast<std::uint64_t>(std::chrono::duration_cast<std::chrono::nanoseconds>(
|
|
std::chrono::steady_clock::now().time_since_epoch())
|
|
.count());
|
|
}
|
|
|
|
/// One stored slot in this many is timed: see `ScopeModel::push_math_sample()`.
|
|
///
|
|
/// A *single* slot, deliberately, even though one slot's work is the same order as one
|
|
/// `clock_gettime` -- so the figure carries a clock read's own cost and is an upper bound. Timing a
|
|
/// *run* of consecutive slots to amortise that away was tried and is wrong: the producer gets its
|
|
/// slots a packet at a time, so a run long enough to be worth timing straddles a packet boundary
|
|
/// now and then and charges the measured slots the **wait for the next packet**. At 2000 samples a
|
|
/// packet a 64-slot run crosses one about 3% of the time, and a ~1.5 ms gap spread over 64 slots
|
|
/// reported 799 ns/slot against the true 78. One slot is always inside one packet.
|
|
constexpr std::uint64_t kMathCostSample = 1024;
|
|
|
|
} // namespace
|
|
|
|
Partition plan_partition(const std::vector<double>& channel_dt, double buffer_bytes,
|
|
double target_span_s, bool with_abs_time) {
|
|
Partition p;
|
|
const std::size_t n = channel_dt.size();
|
|
p.stride.assign(n, 1);
|
|
p.value_capacity.assign(n, 0);
|
|
if (n == 0) return p;
|
|
|
|
// The fastest channel drives the shared time base: every other channel is
|
|
// decimated by the integer ratio of its own interval to that one.
|
|
double base_dt = 0.0;
|
|
for (double dt : channel_dt) {
|
|
if (dt > 0.0) base_dt = base_dt > 0.0 ? std::min(base_dt, dt) : dt;
|
|
}
|
|
// No declared rate: assume 1 kHz rather than one sample per second, so a
|
|
// stream that only says "here are samples" still gets a sane time scale and a
|
|
// sane first frame in `Acquisition::build_stamps()`.
|
|
if (!(base_dt > 0.0)) base_dt = 1.0e-3;
|
|
p.base_dt = base_dt;
|
|
|
|
// No *per-channel* decimation. A stride of `round(dt_i / dt_fastest)` keeps one sample
|
|
// in every stride, and when that stride equals the signal's own period -- a 2 kHz sine
|
|
// on a 1 MHz clock, stride 500 -- every kept sample sits at the same phase and the
|
|
// channel plots as a flat line. The shared clock is decimated below instead, which
|
|
// affects every channel equally and goes through the ingest filter.
|
|
for (std::size_t i = 0; i < n; ++i) p.stride[i] = 1;
|
|
|
|
p.abs_time = with_abs_time;
|
|
|
|
const u64 budget_bytes = static_cast<u64>(std::max(1.0, buffer_bytes));
|
|
// Weighted in *bytes*, not in slots: the shared clock is f64 and the values are f32
|
|
// (see `BufferStore::time()`), so a time slot costs twice a value sample and the two
|
|
// slices are not interchangeable. Counting both as one f32 is what would make
|
|
// `--buffer-mb` quietly understate the time ring by its own size.
|
|
double weight = static_cast<double>(sizeof(f64)) * (with_abs_time ? 2.0 : 1.0); // the time slice(s)
|
|
for (u64 s : p.stride) weight += static_cast<double>(sizeof(f32)) / static_cast<double>(s);
|
|
|
|
const u64 slots = static_cast<u64>(static_cast<double>(budget_bytes) / weight);
|
|
u64 time_capacity = pow2_floor(std::max<u64>(slots, 128));
|
|
if (time_capacity == 0) {
|
|
throw std::runtime_error("buffer too small for one time-base sample per channel");
|
|
}
|
|
|
|
// The target span sets a floor, not a ceiling: the rest of the budget buys a
|
|
// longer ring, which is strictly more history to scroll back through. Growth
|
|
// stops before the total would exceed the budget. A channel's slots are rounded
|
|
// *up* on the power of two `RingBuffer` will round them to anyway, so every
|
|
// channel covers at least the time ring's span.
|
|
// Floor, not ceiling: a channel's window must not be *longer* than the time ring
|
|
// it reads its stamps from. Rounding up (with a stride that is not a power of two)
|
|
// gave a channel more samples than the shared timeline covers, and the oldest of
|
|
// them referenced time slots the ring had already overwritten -- they were then
|
|
// stamped with the epoch, a plausible-looking coordinate that collapsed the trace
|
|
// against the axis. Flooring can leave a channel holding as little as ~62% of the
|
|
// time ring's span, which is a shorter history, not a wrong one.
|
|
const auto capacity_for = [&](u64 ticks, u64 stride) {
|
|
return std::max<u64>(pow2_floor(ticks / stride), 1);
|
|
};
|
|
const auto bytes_for = [&](u64 ticks) {
|
|
u64 total = ticks * sizeof(f64) * (with_abs_time ? 2u : 1u);
|
|
for (u64 s : p.stride) total += capacity_for(ticks, s) * sizeof(f32);
|
|
return total;
|
|
};
|
|
// `bytes_for` bounds the doubling on its own: it is at least `2 * ticks * sizeof(f64)`,
|
|
// so a pass means `2 * ticks` is well inside the budget and cannot overflow.
|
|
while (bytes_for(time_capacity * 2) <= budget_bytes) {
|
|
time_capacity *= 2;
|
|
}
|
|
p.time_capacity = time_capacity;
|
|
for (std::size_t i = 0; i < n; ++i) {
|
|
p.value_capacity[i] = capacity_for(p.time_capacity, p.stride[i]);
|
|
}
|
|
|
|
// Only now, with the ring as long as the budget can buy, decimate the shared clock
|
|
// enough for the requested window to fit in it: deciding this earlier would over-
|
|
// decimate by whatever factor the growth loop was about to add. A power of two keeps
|
|
// the ring capacities (themselves powers of two) in step and lets the producer's group
|
|
// test be a mask (`Acquisition::acc_mask_`) rather than a modulo -- a modulo on a stride
|
|
// this function chose at runtime is a real 64-bit division, which the test *was* paying
|
|
// once per channel per wire sample.
|
|
// The cap stops a nonsense request -- a window far past `--max-window` -- from
|
|
// spinning here; at 2^20 the stored rate is six decades below the wire rate and the
|
|
// picture is averages of averages anyway.
|
|
u64 m = 1;
|
|
if (target_span_s > 0.0) {
|
|
while (m < (u64(1) << 20) &&
|
|
static_cast<double>(time_capacity) * static_cast<double>(m) * base_dt <
|
|
target_span_s) {
|
|
m *= 2;
|
|
}
|
|
}
|
|
p.time_stride = m;
|
|
p.span_s = static_cast<double>(p.time_capacity) * static_cast<double>(p.time_stride) * base_dt;
|
|
return p;
|
|
}
|
|
|
|
// --- BufferStore ------------------------------------------------------------
|
|
|
|
BufferStore::BufferStore(const Partition& part, f64 span_s)
|
|
: part_(part),
|
|
time_(std::make_shared<plot::data::RingBuffer<f64>>(part.time_capacity)) {
|
|
(void)span_s;
|
|
// Same capacity as the clock it parallels: the two are written on the same slots, so a
|
|
// smaller ring would make an export's stamps run out before its values do.
|
|
if (part.abs_time) {
|
|
abs_time_ = std::make_shared<plot::data::RingBuffer<f64>>(part.time_capacity);
|
|
}
|
|
const std::size_t n = part.value_capacity.size();
|
|
values_.reserve(n);
|
|
channels_.reserve(n);
|
|
for (std::size_t i = 0; i < n; ++i) {
|
|
values_.push_back(std::make_shared<plot::data::RingBuffer<f32>>(part.value_capacity[i]));
|
|
// `ChannelBuffer::dx()` multiplies this by the channel's stride, so the
|
|
// value here is the *slot* interval: stride * slot_dt is the real sample
|
|
// interval at the decimation the partition chose.
|
|
const double dt = part.slot_dt();
|
|
channels_.push_back(std::make_shared<ChannelBuffer>(values_[i], time_, part.stride[i], 0.0,
|
|
dt));
|
|
}
|
|
}
|
|
|
|
void BufferStore::acquire_all() {
|
|
for (auto& c : channels_) c->acquire();
|
|
}
|
|
|
|
// --- ScopeModel -------------------------------------------------------------
|
|
|
|
ScopeModel::ScopeModel(const Config& cfg) : cfg_(cfg) {
|
|
trigger_.holdoff_s = cfg.trigger_holdoff_s;
|
|
}
|
|
|
|
namespace {
|
|
/// True when the incoming CONFIG names exactly the stream the model already has: same
|
|
/// signals, same shapes, rates, ranges and time modes. A reconnect re-sends the CONFIG,
|
|
/// and EMI makes that happen at the worst possible moment; the buffer, its history, the
|
|
/// holes, the trigger's held event and every measurement are all still valid and must
|
|
/// survive the reconnect untouched. Only a *different* stream is worth a re-plan, and a
|
|
/// re-plan is what drops history (see `rebuild_locked`).
|
|
bool same_stream_signals(const std::vector<SignalInfo>& incoming,
|
|
const std::vector<SignalInfo>& current) {
|
|
// `current` is the stream's signals with the math channels' synthetic entries appended,
|
|
// so compare the incoming list against current's prefix, skipping nothing on the left.
|
|
std::size_t n_stream = 0;
|
|
while (n_stream < current.size() && !current[n_stream].math) ++n_stream;
|
|
if (incoming.size() != n_stream) return false;
|
|
for (std::size_t i = 0; i < incoming.size(); ++i) {
|
|
const SignalInfo& a = incoming[i];
|
|
const SignalInfo& b = current[i];
|
|
if (a.name != b.name || a.unit != b.unit || a.type_code != b.type_code ||
|
|
a.quant_type != b.quant_type || a.rows != b.rows || a.cols != b.cols ||
|
|
a.range_min != b.range_min || a.range_max != b.range_max ||
|
|
a.time_mode != b.time_mode || a.sampling_rate != b.sampling_rate ||
|
|
a.time_signal_idx != b.time_signal_idx) {
|
|
return false;
|
|
}
|
|
}
|
|
return true;
|
|
}
|
|
} // namespace
|
|
|
|
void ScopeModel::set_signals(std::vector<SignalInfo> signals) {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
// Keep the enable state across CONFIG updates for signals that kept their name. The math
|
|
// channels' synthetic entries carry theirs in `math_`, so they are skipped here -- a CONFIG must
|
|
// not be able to turn a derived channel off, and must not lose it either (it is re-appended
|
|
// below).
|
|
std::unordered_map<std::string, bool> was_enabled;
|
|
for (std::size_t i = 0; i < signals_.size(); ++i) {
|
|
if (signals_[i].math) continue;
|
|
was_enabled.emplace(signals_[i].name, i < enabled_.size() ? enabled_[i] : false);
|
|
}
|
|
signals.erase(std::remove_if(signals.begin(), signals.end(),
|
|
[](const SignalInfo& s) { return s.math; }),
|
|
signals.end());
|
|
if (!signals_.empty() && same_stream_signals(signals, signals_)) {
|
|
// A reconnect's identical CONFIG: the honest recovery is *nothing* -- the ring keeps
|
|
// its history, the trigger keeps its state and its held event, and the outage shows
|
|
// as the gap in the data that it was.
|
|
std::fprintf(stderr, "[udpscope] CONFIG unchanged: keeping the buffer and its history\n");
|
|
std::fflush(stderr);
|
|
return;
|
|
}
|
|
signals_ = std::move(signals);
|
|
enabled_.assign(signals_.size(), false);
|
|
for (std::size_t i = 0; i < signals_.size(); ++i) {
|
|
auto it = was_enabled.find(signals_[i].name);
|
|
// Off until the user enables it (spec: the signal panel's switch is what
|
|
// starts acquisition of a signal). A later CONFIG keeps whatever the user
|
|
// chose for a signal that kept its name.
|
|
enabled_[i] = it != was_enabled.end() ? it->second : false;
|
|
}
|
|
append_math_signals_locked();
|
|
rebuild_channels();
|
|
rebuild_locked();
|
|
generation_.fetch_add(1, std::memory_order_release);
|
|
}
|
|
|
|
void ScopeModel::append_math_signals_locked() {
|
|
// One synthetic signal per math channel, appended after the stream's own. They go through the
|
|
// ordinary signal path on purpose: a derived channel is a channel, so the panel's switch, the
|
|
// plot binding, the measurements and the export all work on it without knowing it is derived.
|
|
// A `Source` in a graph names a *signal*, so another math channel is a valid source too.
|
|
math_signal_names_.clear();
|
|
for (const MathChannel& m : math_) {
|
|
if (m.name.empty()) continue;
|
|
SignalInfo sig;
|
|
sig.name = m.name;
|
|
sig.unit = m.unit;
|
|
sig.math = true;
|
|
sig.sampling_rate = 0.0; // not a streamed signal: the model's own dt applies
|
|
math_signal_names_.push_back(m.name);
|
|
signals_.push_back(std::move(sig));
|
|
enabled_.push_back(m.enabled);
|
|
}
|
|
}
|
|
|
|
void ScopeModel::set_math_channels(std::vector<MathChannel> channels) {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
// An identical list is not a change. The rebuild below replaces the buffer and drops its history --
|
|
// which a file view then has to re-read from disk -- and `apply_math_channels()` runs this on every
|
|
// session start, where a session with no graphs hands over an empty list against a model that has
|
|
// none either. Comparing the graphs is what makes that call free.
|
|
if (channels == math_) return;
|
|
// An edit of an existing graph -- the same channels, names and enables -- does not move the *plan*, which
|
|
// is a partition over the channels. Nothing has to be rebuilt, and every other channel keeps its history:
|
|
// only the edited graphs, and the channels that read them, are recomputed.
|
|
bool same_set = channels.size() == math_.size();
|
|
for (std::size_t i = 0; same_set && i < channels.size(); ++i) {
|
|
same_set = channels[i].name == math_[i].name && channels[i].enabled == math_[i].enabled;
|
|
}
|
|
if (same_set) {
|
|
const auto math_index_of = [this](const std::string& name) -> int {
|
|
for (std::size_t k = 0; k < math_.size(); ++k) {
|
|
if (math_[k].name == name) return static_cast<int>(k);
|
|
}
|
|
return -1;
|
|
};
|
|
std::vector<bool> affected(math_.size(), false);
|
|
std::vector<bool> edited(math_.size(), false);
|
|
for (std::size_t i = 0; i < channels.size(); ++i) {
|
|
if (!(channels[i] == math_[i])) {
|
|
affected[i] = true;
|
|
edited[i] = true;
|
|
}
|
|
}
|
|
// A channel's output is another's *input*, so whatever names a changed channel is recomputed too -- and
|
|
// nothing else is touched at all.
|
|
for (bool moved = true; moved;) {
|
|
moved = false;
|
|
for (std::size_t i = 0; i < channels.size(); ++i) {
|
|
if (!affected[i]) continue;
|
|
for (const MathNode& node : channels[i].nodes) {
|
|
if (node.op != MathOp::Source) continue;
|
|
const int k = math_index_of(node.source);
|
|
if (k >= 0 && !affected[static_cast<std::size_t>(k)]) {
|
|
affected[static_cast<std::size_t>(k)] = true;
|
|
moved = true;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
math_ = std::move(channels);
|
|
const auto index_of = [this](const std::string& name) -> int {
|
|
for (std::size_t c = 0; c < channels_.size(); ++c) {
|
|
if (channels_[c].name == name) return static_cast<int>(c);
|
|
}
|
|
return -1;
|
|
};
|
|
// Every affected channel is prepared again: a runtime carried over from the old graph is history for a
|
|
// filter nobody asked for, and one whose *inputs* changed is history for inputs it no longer has.
|
|
for (std::size_t i = 0; i < math_.size(); ++i) {
|
|
if (affected[i]) math_prepare(math_[i], math_rt_[i], part_.slot_dt(), index_of);
|
|
}
|
|
recompute_math_locked(affected);
|
|
generation_.fetch_add(1, std::memory_order_release);
|
|
return;
|
|
}
|
|
math_ = std::move(channels);
|
|
// Lift the previous synthetic entries out of the signal list; `append_math_signals_locked()`
|
|
// puts the current ones back, so an edit is the same kind of structural rebuild as a CONFIG.
|
|
signals_.erase(std::remove_if(signals_.begin(), signals_.end(),
|
|
[](const SignalInfo& s) { return s.math; }),
|
|
signals_.end());
|
|
enabled_.resize(signals_.size());
|
|
append_math_signals_locked();
|
|
rebuild_channels();
|
|
rebuild_locked();
|
|
// Prepare the runtimes for the rate this plan chose: a cutoff the user typed in Hz is only
|
|
// meaningful against the rate the filter will actually run at, and decimation can have changed
|
|
// it under them.
|
|
prepare_math_locked();
|
|
generation_.fetch_add(1, std::memory_order_release);
|
|
}
|
|
|
|
void ScopeModel::prepare_math_locked() {
|
|
// Called from `rebuild_locked()`, so it runs on **every** plan change -- a CONFIG, an
|
|
// enable/disable, and a window change that moves the decimation. A biquad's coefficients are a
|
|
// function of the sample interval, so one prepared for the old rate is a filter with the wrong
|
|
// cutoff the moment the plan moves; and its state is the old rate's history, which is not a
|
|
// starting point for the new one. Both are reset here, together, so the two cannot drift apart.
|
|
math_rt_.assign(math_.size(), MathRuntime{});
|
|
const double dt = part_.slot_dt();
|
|
// A `Source` node's channel index, resolved here so the per-sample step never looks a name up.
|
|
const auto index_of = [this](const std::string& name) -> int {
|
|
for (std::size_t c = 0; c < channels_.size(); ++c) {
|
|
if (channels_[c].name == name) return static_cast<int>(c);
|
|
}
|
|
return -1;
|
|
};
|
|
for (std::size_t i = 0; i < math_.size(); ++i) {
|
|
math_prepare(math_[i], math_rt_[i], dt, index_of);
|
|
}
|
|
}
|
|
|
|
void ScopeModel::push_loaded_slot(const std::vector<f32>& values, f64 t, f64 abs_t) {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
if (!store_) return;
|
|
BufferStore& store = *store_;
|
|
const std::size_t channels = channels_.size();
|
|
std::vector<f32> slot(channels, 0.0f);
|
|
for (std::size_t i = 0; i < channels; ++i) {
|
|
// A derived channel has no column of its own: `push_math_sample()` below fills it, and pushing the
|
|
// file's value here as well would put *two* samples in its ring per slot -- a trace at double rate.
|
|
// The same rule the packet path states in `ChannelInfo::math`: the ingest must skip it.
|
|
if (channels_[i].math) continue;
|
|
const f32 v = i < values.size() ? values[i] : 0.0f;
|
|
store.values(i).push(v);
|
|
slot[i] = v;
|
|
}
|
|
store.time().push(t);
|
|
if (auto* abs = store.abs_time()) abs->push(abs_t);
|
|
push_math_sample(slot);
|
|
}
|
|
|
|
std::uint64_t ScopeModel::stored_slots() {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
return store_ ? store_->time().size() : 0;
|
|
}
|
|
|
|
void ScopeModel::push_math_sample(const std::vector<f32>& sample) {
|
|
// The producer's entry: every derived channel, and timed (see the cost sampling below).
|
|
math_apply_locked(sample, nullptr, true);
|
|
}
|
|
|
|
void ScopeModel::math_apply_locked(const std::vector<f32>& sample, const std::vector<bool>* write_mask,
|
|
bool timed) {
|
|
// The common case is no derived channels at all, and this runs once per *stored slot* -- so the
|
|
// early-out is what keeps a stream with no math from paying for the feature.
|
|
if (math_.empty()) return;
|
|
// The cost is *sampled*: a run of `kMathCostRun` consecutive slots every `kMathCostPeriod`. This
|
|
// runs up to the wire rate, where a slot's own work is tens of nanoseconds -- so timing every
|
|
// slot is itself able to drop packets, and timing one slot at a time charges it a whole clock
|
|
// read. A run amortises both: one pair of reads per 64 measured slots, one pair per 65536 slots
|
|
// overall, and the figure converges over the millions this counter sees.
|
|
// The sampling counts *producer* slots: a recompute must not inflate `math_slots_`, which the dump
|
|
// reports as the run's own cadence.
|
|
std::uint64_t slot = 0;
|
|
if (timed) {
|
|
slot = math_slot_count_++;
|
|
math_slots_.store(math_slot_count_, std::memory_order_relaxed);
|
|
timed = slot % kMathCostSample == 0;
|
|
}
|
|
if (timed) math_cost_t0_ = now_ns();
|
|
// Values for this slot, extended as the derived ones are computed: a graph may name another math
|
|
// channel, and nodes are evaluated in list order, so by the time one is asked for it is there.
|
|
// `assign()` into a member reuses the capacity -- a local vector here was a heap allocation and
|
|
// free per stored slot, i.e. up to a million pairs a second on a fast stream.
|
|
math_values_.assign(sample.begin(), sample.end());
|
|
if (math_values_.size() < channels_.size()) math_values_.resize(channels_.size(), 0.0f);
|
|
for (std::size_t m = 0; m < math_.size(); ++m) {
|
|
const int channel = m < math_channel_.size() ? math_channel_[m] : -1;
|
|
if (channel < 0 || m >= math_rt_.size()) continue;
|
|
const double out =
|
|
math_step(math_[m], math_rt_[m], math_values_.data(), math_values_.size());
|
|
// A recompute computes every channel -- the graphs are a list walked in order, so a channel's inputs
|
|
// have to be there -- but only the affected ones take the sample.
|
|
if (write_mask == nullptr || (*write_mask)[m]) {
|
|
live().values(static_cast<std::size_t>(channel)).push(static_cast<f32>(out));
|
|
}
|
|
if (static_cast<std::size_t>(channel) < math_values_.size()) {
|
|
math_values_[static_cast<std::size_t>(channel)] = static_cast<f32>(out);
|
|
}
|
|
}
|
|
if (timed) {
|
|
math_cost_count_ += now_ns() - math_cost_t0_;
|
|
++math_timed_count_;
|
|
math_cost_ns_.store(math_cost_count_, std::memory_order_relaxed);
|
|
math_timed_.store(math_timed_count_, std::memory_order_relaxed);
|
|
}
|
|
}
|
|
|
|
void ScopeModel::recompute_math_locked(const std::vector<bool>& affected) {
|
|
if (!store_ || math_.empty()) return;
|
|
BufferStore& store = *store_;
|
|
const std::size_t slots = static_cast<std::size_t>(store.time().size());
|
|
if (slots == 0) return;
|
|
|
|
std::vector<bool> derived(channels_.size(), false);
|
|
for (std::size_t m = 0; m < math_.size(); ++m) {
|
|
const int channel = m < math_channel_.size() ? math_channel_[m] : -1;
|
|
if (channel < 0) continue;
|
|
derived[static_cast<std::size_t>(channel)] = true;
|
|
// Only the affected channels start again. The rest keep their samples *and* their runtimes: their graphs
|
|
// are untouched, so the state they hold is the answer to the inputs they have always had.
|
|
if (affected[m]) store.values(static_cast<std::size_t>(channel)).clear();
|
|
}
|
|
|
|
// One bulk copy per channel per chunk, through the ring's own reader. A derived channel is fed a zero:
|
|
// `math_apply_locked()` overwrites its value as it walks the list in order, and reading its half-rebuilt
|
|
// ring here would read out of range.
|
|
constexpr std::size_t kChunk = 1u << 14;
|
|
const std::size_t base = static_cast<std::size_t>(store.time().base());
|
|
std::vector<std::vector<f32>> inputs(channels_.size());
|
|
for (std::size_t c = 0; c < channels_.size(); ++c) {
|
|
if (!derived[c]) inputs[c].resize(kChunk);
|
|
}
|
|
std::vector<f32> slot(channels_.size(), 0.0f);
|
|
for (std::size_t start = 0; start < slots; start += kChunk) {
|
|
const std::size_t n = std::min(kChunk, slots - start);
|
|
for (std::size_t c = 0; c < channels_.size(); ++c) {
|
|
if (derived[c]) continue;
|
|
store.values(c).copy_from(base + start, n, inputs[c].data());
|
|
}
|
|
for (std::size_t i = 0; i < n; ++i) {
|
|
for (std::size_t c = 0; c < channels_.size(); ++c) {
|
|
slot[c] = derived[c] ? 0.0f : inputs[c][i];
|
|
}
|
|
math_apply_locked(slot, &affected, /*timed=*/false);
|
|
}
|
|
}
|
|
// The pyramid is the consumer's and the tick publishes it; the samples are written here.
|
|
}
|
|
|
|
void ScopeModel::rebuild_channels() {
|
|
channels_.clear();
|
|
// Which channel each math channel ended up with, by name: a disabled one has none, and the
|
|
// producer must not push a derived sample into another channel's ring.
|
|
math_channel_.assign(math_.size(), -1);
|
|
const auto math_index_of = [&](const std::string& name) -> int {
|
|
for (std::size_t k = 0; k < math_.size(); ++k) {
|
|
if (math_[k].name == name) return static_cast<int>(k);
|
|
}
|
|
return -1;
|
|
};
|
|
for (std::size_t s = 0; s < signals_.size(); ++s) {
|
|
const SignalInfo& sig = signals_[s];
|
|
if (s >= enabled_.size() || !enabled_[s]) continue;
|
|
if (is_time_signal(signals_, s)) continue;
|
|
if (sig.math) {
|
|
const int k = math_index_of(sig.name);
|
|
if (k >= 0) math_channel_[static_cast<std::size_t>(k)] = static_cast<int>(channels_.size());
|
|
}
|
|
ChannelInfo ch;
|
|
ch.signal = s;
|
|
ch.enabled = true;
|
|
// A UDPStreamer signal is one channel: its columns are consecutive samples of
|
|
// that channel within a DATA packet (1x1000 is 1000 samples of the same
|
|
// signal, not 1000 signals). The worker client hands a multi-element signal
|
|
// exactly `elements` values per packet, which is the block of samples.
|
|
ch.name = sig.name;
|
|
ch.unit = sig.unit;
|
|
ch.math = sig.math;
|
|
ch.elements = sig.elements();
|
|
ch.dt = sig.sampling_rate > 0.0 ? 1.0 / sig.sampling_rate : 0.0;
|
|
// A calibration the user already set for this name (before it was enabled, or
|
|
// before a CONFIG renumbered the channels) still applies.
|
|
if (auto it = calibration_.find(sig.name); it != calibration_.end()) {
|
|
ch.gain = it->second.first;
|
|
ch.offset = it->second.second;
|
|
}
|
|
channels_.push_back(std::move(ch));
|
|
}
|
|
}
|
|
|
|
void ScopeModel::rebuild_locked() {
|
|
// The store is replaced below, so every hole's `slot_begin`/`tick_begin`/`x_begin` describes a
|
|
// buffer that no longer exists: the new store's base is 0, so `forget_holes_before()` cannot
|
|
// retire them and they would shade a band over the wrong x until the table pushed them out.
|
|
// Dropping them is the honest answer -- the samples they reported absent are gone with the
|
|
// history. `++plan_generation_` is what tells the producer the same thing about `tick_`, whose
|
|
// relation to `produced()` is what addresses a backfill (see `Acquisition::on_data`).
|
|
holes_.clear();
|
|
++plan_generation_;
|
|
|
|
std::vector<double> dts;
|
|
dts.reserve(channels_.size());
|
|
for (const ChannelInfo& c : channels_) dts.push_back(c.dt);
|
|
|
|
if (dts.empty()) {
|
|
part_ = Partition{};
|
|
store_.reset();
|
|
return;
|
|
}
|
|
|
|
part_ = plan_partition(dts, cfg_.buffer_mb * 1024.0 * 1024.0,
|
|
target_span_.load(std::memory_order_relaxed), cfg_.relative_time);
|
|
store_ = std::make_unique<BufferStore>(part_, part_.span_s);
|
|
|
|
if (trigger_.channel >= static_cast<int>(channels_.size())) trigger_.channel = -1;
|
|
prepare_math_locked();
|
|
}
|
|
|
|
bool ScopeModel::set_target_span(double seconds) {
|
|
if (!(seconds > 0.0)) return false;
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
const double previous = target_span_.exchange(seconds, std::memory_order_relaxed);
|
|
if (channels_.empty()) return false;
|
|
// Re-plan only if the factor moves. The planner is cheap, so ask it rather than
|
|
// reimplementing its arithmetic here -- two expressions for the same decision is how
|
|
// the buffer and the view end up disagreeing about the stored rate.
|
|
std::vector<double> dts;
|
|
dts.reserve(channels_.size());
|
|
for (const ChannelInfo& c : channels_) dts.push_back(c.dt);
|
|
const Partition want = plan_partition(dts, cfg_.buffer_mb * 1024.0 * 1024.0, seconds, cfg_.relative_time);
|
|
if (want.time_stride == part_.time_stride) return false;
|
|
std::fprintf(stderr,
|
|
"[udpscope] window %.4f s -> %.4f s: decimation %llu -> %llu, stored rate "
|
|
"%.4g -> %.4g S/s (anti-alias %.4g Hz); buffer rebuilt\n",
|
|
previous, seconds, static_cast<unsigned long long>(part_.time_stride),
|
|
static_cast<unsigned long long>(want.time_stride), part_.stored_rate(),
|
|
want.stored_rate(), want.antialias_hz());
|
|
std::fflush(stderr);
|
|
rebuild_locked();
|
|
return true;
|
|
}
|
|
|
|
bool ScopeModel::is_time_signal(const std::vector<SignalInfo>& signals,
|
|
std::size_t signal) {
|
|
if (signal >= signals.size()) return false;
|
|
const SignalInfo& s = signals[signal];
|
|
// Referenced by name: another signal declares this one as its time base.
|
|
for (const SignalInfo& other : signals) {
|
|
if (other.time_signal_idx == signal) return true;
|
|
}
|
|
// A full-array signal that declares **no sample rate** is the stream's timestamp
|
|
// signal: it has stamps but no cadence of its own, because it is not data.
|
|
// Data channels can also be declared full-array (per-element stamps), and they
|
|
// keep a declared rate -- treating them as time signals hid every real channel.
|
|
if (s.time_mode == UDPS_TIME_FULL_ARRAY && s.elements() > 1 && s.sampling_rate == 0.0) {
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
bool ScopeModel::signal_enabled(std::size_t signal) const {
|
|
return signal < enabled_.size() && enabled_[signal];
|
|
}
|
|
|
|
void ScopeModel::set_signal_enabled(std::size_t signal, bool on) {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
if (signal >= enabled_.size() || enabled_[signal] == on) return;
|
|
enabled_[signal] = on;
|
|
// A math channel's synthetic signal is enabled *here* -- it is the panel's switch -- while
|
|
// `append_math_signals_locked()` re-states `enabled_` from `math_` on the next edit. The two have
|
|
// to agree, or an edit would quietly turn a derived channel off (and the rebuild below reads
|
|
// `enabled_`, so the channel itself would go with it).
|
|
if (signal < signals_.size() && signals_[signal].math) {
|
|
for (MathChannel& m : math_) {
|
|
if (m.name == signals_[signal].name) m.enabled = on;
|
|
}
|
|
}
|
|
// The trigger names a channel by its display name, which is stable across
|
|
// renumbering: when its own signal is disabled the channel vanishes (the trigger
|
|
// then waits, and comes back when the signal is re-enabled).
|
|
const int old_trigger = trigger_.channel;
|
|
const std::string trigger_name =
|
|
(old_trigger >= 0 && old_trigger < static_cast<int>(channels_.size()))
|
|
? channels_[static_cast<std::size_t>(old_trigger)].name
|
|
: std::string();
|
|
rebuild_channels();
|
|
trigger_.channel = -1;
|
|
if (!trigger_name.empty()) {
|
|
for (std::size_t i = 0; i < channels_.size(); ++i) {
|
|
if (channels_[i].name == trigger_name) trigger_.channel = static_cast<int>(i);
|
|
}
|
|
}
|
|
rebuild_locked();
|
|
generation_.fetch_add(1, std::memory_order_release);
|
|
}
|
|
|
|
std::pair<double, double> ScopeModel::calibration(std::size_t signal) {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
if (signal >= signals_.size()) return {1.0, 0.0};
|
|
const auto it = calibration_.find(signals_[signal].name);
|
|
return it != calibration_.end() ? it->second : std::pair<double, double>{1.0, 0.0};
|
|
}
|
|
|
|
void ScopeModel::set_calibration(std::size_t signal, double gain, double offset) {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
if (signal < signals_.size()) calibration_[signals_[signal].name] = {gain, offset};
|
|
bool any = false;
|
|
for (ChannelInfo& c : channels_) {
|
|
if (c.signal != signal) continue;
|
|
c.gain = gain;
|
|
c.offset = offset;
|
|
any = true;
|
|
}
|
|
// Calibration is applied at ingest, so the buffers are rebuilt to avoid mixing
|
|
// calibrated and uncalibrated samples (and stale LOD aggregates).
|
|
if (any) {
|
|
rebuild_locked();
|
|
generation_.fetch_add(1, std::memory_order_release);
|
|
}
|
|
}
|
|
|
|
double ScopeModel::channel_dt(std::size_t channel) const {
|
|
return channel < channels_.size() ? channels_[channel].dt : 0.0;
|
|
}
|
|
|
|
BufferStore& ScopeModel::live() { return *store_; }
|
|
|
|
void ScopeModel::live_window_locked(double lo_s, double hi_s, u64& first, u64& last) {
|
|
// There may be **no store at all**: `rebuild_locked()` resets it for a stream with no channels, which
|
|
// is what the model is in until a CONFIG arrives. `live()` dereferences the pointer, so every path
|
|
// that can be reached without a stream has to ask first -- pausing a scope that is not connected did
|
|
// not, and walked into this function through `publish_pause_view()`, a null dereference.
|
|
first = 0;
|
|
last = 0;
|
|
if (store_ == nullptr) return;
|
|
BufferStore& src = live();
|
|
plot::data::RingBuffer<f64>& tsrc = src.time();
|
|
|
|
// The slice is one absolute index range, found in the time ring and then shared by
|
|
// every value ring. Sharing it is what makes a cut's stamps and each of its channels
|
|
// the same length and index-aligned *by construction*, instead of by two independent
|
|
// truncations that could disagree.
|
|
const u64 tbase = tsrc.produced() - tsrc.size();
|
|
const u64 tend = tsrc.produced();
|
|
// The bounds are compared in the ring's own type, which is the `f64` they arrive in,
|
|
// so a bound that lands exactly on a sample includes it. This used to need rounding
|
|
// *down* to the stored f32 first: an f64 comparison against an f32 stamp decided such
|
|
// a sample by which way its stamp happened to round, and the same nominal window kept
|
|
// or dropped its first sample depending on the magnitude of the stream time
|
|
// (`f32(38.001) < 38.001` but `f32(18.001) > 18.001`). The ring is f64 now and the
|
|
// question does not arise.
|
|
const f64 t_lo = lo_s;
|
|
const f64 t_hi = hi_s;
|
|
|
|
// Stream time is non-decreasing, so the bounds are a binary search. Searching the
|
|
// stamps is the whole point: the window is in seconds and this ring is the only
|
|
// thing that knows how many samples a second of this stream holds.
|
|
u64 a = tbase, b = tend;
|
|
while (a < b) {
|
|
const u64 m = a + (b - a) / 2;
|
|
if (tsrc[m - tbase] < t_lo) {
|
|
a = m + 1;
|
|
} else {
|
|
b = m;
|
|
}
|
|
}
|
|
first = a;
|
|
b = tend;
|
|
while (a < b) {
|
|
const u64 m = a + (b - a) / 2;
|
|
if (tsrc[m - tbase] <= t_hi) {
|
|
a = m + 1;
|
|
} else {
|
|
b = m;
|
|
}
|
|
}
|
|
last = a;
|
|
|
|
for (std::size_t c = 0; c < channels_.size(); ++c) {
|
|
plot::data::RingBuffer<f32>& from = src.values(c); // values stay f32
|
|
first = std::max(first, from.produced() - from.size());
|
|
last = std::min(last, from.produced());
|
|
}
|
|
if (last < first) last = first;
|
|
}
|
|
|
|
void ScopeModel::publish_pause_view(double lo_s, double hi_s, double x_offset) {
|
|
// Nothing has arrived yet, so there is nothing to freeze: pausing a scope that is not connected is a
|
|
// legitimate gesture (the button works), and it must not take a window from a store that does not
|
|
// exist.
|
|
if (store_ == nullptr) return;
|
|
// Timed from before the lock: the wait for it is part of what freezing a view costs the thread
|
|
// doing it, and this one is the UI's.
|
|
const std::uint64_t t0 = now_ns();
|
|
auto snap = std::make_shared<EventSnapshot>();
|
|
snap->x_offset = x_offset;
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
BufferStore& src = live();
|
|
u64 first = 0;
|
|
u64 last = 0;
|
|
live_window_locked(lo_s, hi_s, first, last);
|
|
const u64 n = last - first;
|
|
snap->stamps.resize(static_cast<std::size_t>(n));
|
|
if (n > 0) {
|
|
// The stamps come out of the ring on the stream's own clock and are rebased onto
|
|
// `x_offset`, which is the coordinate the axis was *already* reading: a live view is
|
|
// newest-relative, so leaving them absolute would throw the frozen picture a whole uptime
|
|
// to the right of the axis at the instant the user pressed pause. Nothing is lost by the
|
|
// shift -- `x_offset` is exactly what makes the stream's time recoverable, which is what
|
|
// `stream_at_x0()` reads it for.
|
|
src.time().copy_from(first, n, snap->stamps.data());
|
|
for (f64& t : snap->stamps) t -= x_offset;
|
|
}
|
|
if (const plot::data::RingBuffer<f64>* abs = src.abs_time(); abs != nullptr) {
|
|
snap->abs_stamps.resize(static_cast<std::size_t>(n));
|
|
abs->copy_from(first, n, snap->abs_stamps.data());
|
|
// Absolute seconds outright -- deliberately not rebased like `stamps`.
|
|
for (f64& t : snap->abs_stamps) t += epoch_;
|
|
}
|
|
snap->values.resize(channels_.size());
|
|
snap->y.resize(channels_.size(), plot::Range{0.0, 0.0});
|
|
snap->pyramids.resize(channels_.size());
|
|
for (std::size_t c = 0; c < channels_.size(); ++c) {
|
|
const u64 stride = std::max<u64>(1, part_.stride[c]);
|
|
auto& vals = snap->values[c];
|
|
const u64 take = std::min(n / stride, src.values(c).size());
|
|
vals.resize(static_cast<std::size_t>(take));
|
|
if (take > 0) src.values(c).copy_from(first + n - take, take, vals.data());
|
|
f32 lo = vals.empty() ? 0.0f : vals.front();
|
|
f32 hi = lo;
|
|
// The LOD index, folded into the bounds pass exactly as `publish_event()` folds it, and for
|
|
// the same reason: a snapshot that answers `pyramid()` with an empty one has `level_for()`
|
|
// descend to level 0, so the frozen view is re-gathered from every raw sample it holds on
|
|
// every frame it is on screen. A pause is not a cheaper case than a capture -- it freezes the
|
|
// *same* rings -- and leaving it out was visible twice over. The cost: `frame path:` read
|
|
// `raw=21 lod=118` over a 1.6 s pause against `raw=1 lod=170` for the same run live, with the
|
|
// draw mean 12.34 -> 18.18 ms. And the *look* changed, which is the half a cost figure does
|
|
// not show: `stage_trace_density()` grades a column against its own peak from a `1/height`
|
|
// deposit per segment, so a column fed ~660 raw inter-sample segments accumulates a far
|
|
// spikier distribution -- and a far larger peak to normalise against -- than one fed the one
|
|
// or two bucket envelopes the pyramid gives. The paused trace read as a hard bright core with
|
|
// its dwell tail dimmed out, i.e. the trace changed appearance on pause.
|
|
//
|
|
// It is not free, and the trade is one-off against per-frame: building it took the pause
|
|
// publish from 6.15 ms to 22.31 ms (`publish cost:`), which blocks the *producer* -- against
|
|
// ~5 ms a frame of draw for as long as the pause lasts, which at ~60 fps a 1.6 s pause repays
|
|
// thirty times over. The producer absorbs the stall: `gaps=0` with and without a pause, the
|
|
// paused run having published in 86 ms.
|
|
auto& pyr = snap->pyramids[c];
|
|
pyr.reserve(take);
|
|
for (f32 v : vals) {
|
|
lo = std::min(lo, v);
|
|
hi = std::max(hi, v);
|
|
pyr.append(v);
|
|
}
|
|
snap->y[c] = plot::Range{lo, hi};
|
|
}
|
|
pause_view_.store(std::move(snap), std::memory_order_release);
|
|
note_publish(publish_pause_count_, publish_pause_us_total_, publish_pause_us_worst_,
|
|
(now_ns() - t0) / 1000u);
|
|
}
|
|
|
|
void ScopeModel::publish_event(double pre_s, double post_s, double edge) {
|
|
// The caller is the producer and already holds `mutex()`, so every microsecond counted here is a
|
|
// microsecond the UI tick is blocked for -- which is what the documented `tick` max is made of.
|
|
const std::uint64_t t0 = now_ns();
|
|
auto snap = std::make_shared<EventSnapshot>();
|
|
snap->x_offset = edge; // the stream time x = 0 means, for everything that has to cross back
|
|
BufferStore& src = live();
|
|
plot::data::RingBuffer<f64>& tsrc = src.time();
|
|
|
|
u64 first = 0, last = 0;
|
|
live_window_locked(edge - std::max(0.0, pre_s), edge + std::max(0.0, post_s), first, last);
|
|
const u64 n = last - first;
|
|
|
|
snap->stamps.resize(static_cast<std::size_t>(n));
|
|
if (n > 0) {
|
|
tsrc.copy_from(first, n, snap->stamps.data());
|
|
// Edge-relative from here on: the frozen event reads [-pre, +post] with the
|
|
// edge at exactly 0, with no axis-level shift any other code path could
|
|
// disagree with.
|
|
for (auto& t : snap->stamps) t -= edge;
|
|
}
|
|
if (const plot::data::RingBuffer<f64>* abs = src.abs_time(); abs != nullptr) {
|
|
snap->abs_stamps.resize(static_cast<std::size_t>(n));
|
|
abs->copy_from(first, n, snap->abs_stamps.data());
|
|
// Absolute seconds outright -- deliberately not rebased like `stamps`.
|
|
for (f64& t : snap->abs_stamps) t += epoch_;
|
|
}
|
|
|
|
snap->values.resize(channels_.size());
|
|
snap->y.resize(channels_.size(), plot::Range{0.0, 0.0});
|
|
snap->pyramids.resize(channels_.size());
|
|
for (std::size_t c = 0; c < channels_.size(); ++c) {
|
|
auto& vals = snap->values[c];
|
|
vals.resize(static_cast<std::size_t>(n));
|
|
if (n > 0) src.values(c).copy_from(first, n, vals.data());
|
|
f32 lo = vals.empty() ? 0.0f : vals.front();
|
|
f32 hi = lo;
|
|
// The LOD index is folded into the bounds pass that was already here, so the
|
|
// snapshot costs one O(n) walk per channel rather than two. Without a pyramid
|
|
// the renderer's `level_for()` descends to level 0 and the whole capture is
|
|
// drawn raw on every frame the event is held: see `EventSnapshot::pyramids`.
|
|
auto& pyr = snap->pyramids[c];
|
|
pyr.reserve(n);
|
|
for (f32 v : vals) {
|
|
lo = std::min(lo, v);
|
|
hi = std::max(hi, v);
|
|
pyr.append(v);
|
|
}
|
|
snap->y[c] = plot::Range{lo, hi};
|
|
}
|
|
|
|
// The span is read off the stamps that were actually captured, never off the
|
|
// request: the ring may not reach back a full `pre_s`, and the producer stops the
|
|
// capture at the first sample past `post_s` rather than exactly on it.
|
|
snap->pre_s = snap->stamps.empty() ? 0.0 : -static_cast<double>(snap->stamps.front());
|
|
snap->post_s = snap->stamps.empty() ? 0.0 : static_cast<double>(snap->stamps.back());
|
|
event_.store(std::move(snap), std::memory_order_release);
|
|
// Bumped after the snapshot is in place, so a consumer that sees the new sequence
|
|
// number sees the snapshot it counts.
|
|
event_seq_.fetch_add(1, std::memory_order_release);
|
|
note_publish(publish_event_count_, publish_event_us_total_, publish_event_us_worst_,
|
|
(now_ns() - t0) / 1000u);
|
|
}
|
|
|
|
void BufferStore::set_epoch(f64 epoch) noexcept {
|
|
for (auto& c : channels_) c->set_epoch(epoch);
|
|
}
|
|
|
|
void ScopeModel::set_display_epoch(f64 e) noexcept {
|
|
display_epoch_.store(e, std::memory_order_relaxed);
|
|
if (store_) store_->set_epoch(e);
|
|
}
|
|
|
|
void ScopeModel::acquire_all() {
|
|
if (store_) {
|
|
store_->acquire_all();
|
|
// Drop holes whose slots the time ring has already overwritten: their slot addresses
|
|
// would alias live slots once the ring wraps past them, and they can no longer be
|
|
// backfilled or drawn. `base()` is the absolute index of the oldest slot the ring
|
|
// still retains -- anything earlier has already been discarded.
|
|
forget_holes_before(store_->time().base());
|
|
}
|
|
}
|
|
|
|
void ScopeModel::note_hole(const Hole& h) {
|
|
if (holes_.size() >= kMaxHoles) holes_.erase(holes_.begin());
|
|
holes_.push_back(h);
|
|
}
|
|
|
|
void ScopeModel::forget_holes_before(u64 slot) {
|
|
// A hole whose last slot the ring has dropped can no longer be backfilled or drawn, and
|
|
// its address would alias a live slot once the ring wraps past it.
|
|
std::erase_if(holes_, [slot](const Hole& h) { return h.slot_begin + h.slot_count <= slot; });
|
|
}
|
|
|
|
void ScopeModel::set_trigger_enabled(bool on) {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
trigger_.enabled = on;
|
|
trigger_enabled_.store(on, std::memory_order_relaxed);
|
|
}
|
|
|
|
void ScopeModel::set_epoch_if_unset(f64 e) noexcept {
|
|
if (epoch_ == 0.0) epoch_ = e;
|
|
}
|
|
|
|
} // namespace udpscope
|