udpscope/src/model.cpp

// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari

#include "model.hpp"

#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::reset() noexcept {
  time_->clear();
  if (abs_time_) abs_time_->clear();
  for (auto& v : values_) v->clear();
  for (auto& c : channels_) {
    c->invalidate();
    c->set_epoch(0.0);
  }
}

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;

  // The refill replays the inputs' *retained* window, which starts at this absolute index and not at
  // zero, so an emptied derived ring has to resume its count there: `clear()` would renumber it from 0
  // and leave its `produced()` short by `base` for a whole ring's worth of fresh data. `live_window_locked()`
  // clamps the window against every channel's `produced()`, so that one lagging channel collapses the
  // live view, every pause view and every trigger capture to nothing -- the "acquisition freezes after a
  // math edit" defect, invisible until the ring has wrapped once (`base == 0` before that).
  const std::size_t base = static_cast<std::size_t>(store.time().base());

  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]) continue;
    store.values(static_cast<std::size_t>(channel)).clear_to(base);
    // And the published window with them: this is a rewrite in place, so `produced()` lands back where it
    // started and the incremental `acquire()` would publish nothing at all -- leaving the pyramid, which is
    // what the renderer and every measurement read, describing the samples the edit just replaced.
    store.channel(static_cast<std::size_t>(channel)).republish_from(base);
  }

  // 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;
  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 (`republish_from()` above put its window
  // back at `base` so that publish is one window's worth of work); the samples are written here.
}

void ScopeModel::rebuild_channels() {
  // The trigger names a channel by its display name, which is stable across the renumbering this
  // function *is*; the index is not. Captured and restored here rather than at the call sites
  // because every caller renumbers and every caller needs it: adding a math channel appends a
  // signal and shifts nothing, but removing one, or a CONFIG that reorders the stream's own
  // signals, moves every index after it. `rebuild_locked()` then only clamps an out-of-range
  // index to -1, and `TriggerEngine::process()` returns immediately on a channel of -1 -- so the
  // panel went on naming the channel while nothing could ever fire again.
  const std::string trigger_name =
      (trigger_.channel >= 0 && trigger_.channel < static_cast<int>(channels_.size()))
          ? channels_[static_cast<std::size_t>(trigger_.channel)].name
          : std::string();
  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));
  }

  // A name that is no longer a channel -- its signal was disabled, or the stream stopped
  // declaring it -- leaves the trigger unassigned, which is a wait rather than a silent
  // mis-aim; re-enabling the signal brings it back here.
  trigger_.channel = -1;
  if (trigger_name.empty()) return;
  for (std::size_t i = 0; i < channels_.size(); ++i) {
    if (channels_[i].name == trigger_name) trigger_.channel = static_cast<int>(i);
  }
}

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;
  }

  const Partition want = plan_partition(dts, cfg_.buffer_mb * 1024.0 * 1024.0,
                                        target_span_.load(std::memory_order_relaxed),
                                        cfg_.relative_time);
  // A plan that describes the storage already allocated is cleared, not rebuilt. Allocating
  // a store is what costs -- its rings are value-initialised, so it touches the whole
  // `--buffer-mb` -- and most structural changes do not move the plan at all: a calibration
  // keeps the channel set, and so does leaving an event for the live view. Startup asked for
  // sixteen of these and eight of them planned the same buffer twice. Clearing has the same
  // meaning (`BufferStore::reset()`): the history is gone either way, which is why
  // `plan_generation_` bumps above regardless.
  if (store_ && part_.same_storage_as(want)) {
    part_ = want;
    store_->reset();
  } else {
    part_ = want;
    const std::uint64_t t0 = now_ns();
    store_ = std::make_unique<BufferStore>(part_, part_.span_s);
    // Named, with what it cost: this is the one operation in the application whose duration a
    // user can see with no instrument, because it blocks the main thread for the time it takes
    // to touch the whole buffer -- about 300 ms per GB. A startup or a mode change that reads
    // more than one of these lines is a re-plan that should have been coalesced, and that is
    // the defect the line exists to name (it was fifteen of them at startup).
    std::fprintf(stderr, "[udpscope] buffer allocated: %zu channels, %.3g s span, %.0f ms\n",
                 channels_.size(), part_.span_s, double(now_ns() - t0) / 1.0e6);
    std::fflush(stderr);
  }

  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 == kTimeFullArray && 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];
}

bool ScopeModel::enable_locked(std::size_t signal, bool on) {
  if (signal >= enabled_.size() || enabled_[signal] == on) return false;
  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;
    }
  }
  return true;
}

void ScopeModel::rebuild_enabled_locked() {
  // `rebuild_channels()` carries the trigger's channel across the renumbering by name.
  rebuild_channels();
  rebuild_locked();
  generation_.fetch_add(1, std::memory_order_release);
}

void ScopeModel::set_signal_enabled(std::size_t signal, bool on) {
  std::lock_guard<std::mutex> lock(mutex_);
  if (!enable_locked(signal, on)) return;
  rebuild_enabled_locked();
}

void ScopeModel::set_signals_enabled(const std::vector<std::size_t>& signals, bool on) {
  std::lock_guard<std::mutex> lock(mutex_);
  bool any = false;
  for (const std::size_t s : signals) {
    if (enable_locked(s, on)) any = true;
  }
  // Nothing moved: no re-plan, and in particular no `generation_` bump -- a restored session
  // re-enables what is already on, and a bump there would make the UI rebind for nothing.
  if (!any) return;
  rebuild_enabled_locked();
}

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->channel_names.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) {
    snap->channel_names[c] = channels_[c].name;
    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());
  // The persistence retention: what was on screen until now becomes the *newest of the older*
  // acquisitions, and the set is cut back to the depth the panel asks for and to the retention
  // budget. Nothing is retained at depth 1, so the default costs no memory and no copy at all.
  if (const int depth = persist_depth_locked(); depth > 1) {
    std::shared_ptr<const EventSnapshot> prev = event_.load(std::memory_order_relaxed);
    if (prev != nullptr) {
      // Only the slots the depth wants are shifted: the ones past it are the prune's business, and
      // moving an acquisition into a slot nothing can ever read is work for a snapshot about to be
      // dropped.
      const std::size_t top = std::min(static_cast<std::size_t>(depth - 1), persist_.size());
      for (std::size_t i = top; i-- > 1;) {
        persist_[i].store(persist_[i - 1].load(std::memory_order_relaxed),
                          std::memory_order_relaxed);
      }
      persist_[0].store(std::move(prev), std::memory_order_relaxed);
    }
  }
  prune_persist_locked(static_cast<std::size_t>(persist_depth_locked() - 1));
  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::prune_persist_locked(std::size_t keep) {
  const std::size_t limit = std::min(keep, persist_.size());
  const std::size_t budget = persist_budget_bytes();
  std::size_t bytes = 0;
  std::size_t kept = 0;
  std::uint64_t dropped = 0;
  bool full = false;
  for (std::size_t i = 0; i < persist_.size(); ++i) {
    const std::shared_ptr<const EventSnapshot> snap = persist_[i].load(std::memory_order_relaxed);
    if (snap == nullptr) continue;
    if (i < limit) {
      // Either bound can cut the set: the depth the panel asks for, and the budget, which a long
      // window on a fast stream reaches long before the seventh layer.
      const std::size_t b = snapshot_bytes(*snap);
      if (!full && bytes + b <= budget) {
        bytes += b;
        ++kept;
        continue;
      }
      full = true;  // does not fit, and neither does anything older
      // Only what the *budget* refused is counted: a slot past the depth is not a drop, it is the
      // depth -- and counting it would make the reading say the memory was the reason when the
      // panel's own number was.
      ++dropped;
    }
    persist_[i].store(nullptr, std::memory_order_relaxed);
  }
  persist_count_.store(kept, std::memory_order_relaxed);
  persist_bytes_.store(bytes, std::memory_order_relaxed);
  persist_dropped_.store(persist_dropped_.load(std::memory_order_relaxed) + dropped,
                         std::memory_order_relaxed);
}

void ScopeModel::set_persist_depth(int depth) {
  const int n = std::clamp(depth, 1, kMaxPersistDepth);
  std::lock_guard<std::mutex> lock(mutex_);
  trigger_.persist_depth = n;
  // Pruned here and not only at the next latch: lowering the depth is the user asking to stop
  // paying for the layers, and a trigger that has stopped firing never reaches another publish.
  prune_persist_locked(static_cast<std::size_t>(n - 1));
}

void ScopeModel::set_epoch_if_unset(f64 e) noexcept {
  if (epoch_ == 0.0) epoch_ = e;
}

}  // namespace udpscope