udpscope/src/channel.cpp

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

#include "channel.hpp"

#include <algorithm>
#include <cmath>

namespace udpscope {
namespace {

/// Raw samples read per chunk on the gather path, so it never allocates.
constexpr u64 kChunk = 256;

/// The pyramid handed out for a slot with nothing bound to it. The renderer never
/// asks an empty series a bucket question, but the reference must be valid.
const plot::data::MinMaxPyramid& null_pyramid() {
  static const plot::data::MinMaxPyramid empty;
  return empty;
}

}  // namespace

// --- ChannelBuffer ----------------------------------------------------------

ChannelBuffer::ChannelBuffer(std::shared_ptr<const plot::data::RingBuffer<f32>> values,
                             std::shared_ptr<const plot::data::RingBuffer<f64>> times, u64 stride,
                             f64 epoch, f64 nominal_dt)
    : values_(std::move(values)), times_(std::move(times)), stride_(stride < 1 ? 1 : stride),
      epoch_(epoch), dt_(nominal_dt) {}

void ChannelBuffer::acquire() noexcept {
  const u64 produced = values_->produced();
  // Nothing new skips the append pass only: the backfill drain at the end of this function
  // runs either way, because a late packet can land in a frame interval that pushed nothing.
  if (produced > published_) {
    // A producer that lapped this ring -- a long trigger wait, or a store that was
    // not being published for a while -- makes the incremental append meaningless:
    // the samples between `published_` and `produced_` are no longer in the ring.
    // Rebuilding from the retained window is O(capacity) once, and it is the only
    // way the published window stays coherent (an empty or garbled window is what
    // "the trace disappeared" looks like from outside).
    if (produced - published_ > values_->capacity()) {
      published_ = 0;
      pyramid_.clear();
    }
    f32 chunk[kChunk];
    for (u64 i = published_; i < produced;) {
      const u64 take = std::min<u64>(kChunk, produced - i);
      values_->copy_from(i, take, chunk);
      for (u64 k = 0; k < take; ++k) pyramid_.append(chunk[k]);
      i += take;
    }
    published_ = produced;
    // The pyramid's window follows the ring's: sample index `base` is the oldest
    // one still retained. drop_before() is monotonic and cheap.
    pyramid_.drop_before(values_->base());
    // And the window's first stamp with it, here and nowhere else (see the member's note): the
    // window is pinned by the line above, so its origin is a fixed sample, and this is the one
    // place in the object's life where that sample's slot is safe to read.
    if (pyramid_.size() > 0) window_x0_ = time_of(absolute(0));
  }
  // Backfills, applied after the append pass above: an amend is only ever queued for a slot
  // the value ring already produced, so the append has reached it and the pyramid's bounds
  // test will accept it. Unconditional, outside the `produced > published_` guard -- a late
  // packet can land in a frame interval that pushed nothing new, and the plan's placement
  // behind the old early return would have left the pyramid stale until the next push.
  if (!pending_amends_.empty()) {
    for (const Amend& a : pending_amends_) pyramid_.amend(a.absolute, a.value);
    pending_amends_.clear();
  }
}

void ChannelBuffer::invalidate() noexcept {
  published_ = 0;
  window_x0_ = 0.0;
  pyramid_.clear();
  pending_amends_.clear();
}

void ChannelBuffer::republish_from(u64 absolute) noexcept {
  published_ = absolute;
  window_x0_ = 0.0;
  pending_amends_.clear();
  // `clear()` leaves the pyramid empty at absolute 0, and `drop_before()` past the end
  // restarts an empty window at the index it is given -- which is the only way to say
  // "nothing is published, and the next append is sample `absolute`". The no-op guard is
  // for `absolute == 0`, where `clear()` has already said exactly that.
  pyramid_.clear();
  if (absolute > 0) pyramid_.drop_before(absolute);
}

f64 ChannelBuffer::time_of(u64 absolute_index) const noexcept {
  // Channel sample `j` is produced on the tick that also pushes time slot
  // `j*stride + stride - 1` (see `Acquisition::on_data`): the producer stamps the
  // time ring every `time_stride` ticks and the channel every `stride` of those,
  // so the last tick of the group is the one that carries the channel's sample.
  const u64 index = absolute_index * stride_ + (stride_ - 1);
  // The time ring is circular, so an index stays readable while it is *retained*
  // and `copy_from()` masks it. Testing it against `capacity()` would reject every
  // index past the first wrap -- which is every sample once the ring has filled,
  // and would silently stamp them all with x = 0.
  //
  // An index the ring has already dropped is **clamped to the oldest readable slot,
  // never answered with a sentinel**. The published window is pinned by `acquire()`
  // to the value ring's base, and both rings keep sliding afterwards (their
  // capacities are equal, so there is no slack at all): once the ring saturates, the
  // few oldest samples of the window routinely point at slots the producer has
  // overwritten since. Returning 0 for those put a stamp of 0 into `x_bounds()`, and
  // from there into `dx()`/`x0()` and every consumer of the window's extent. So
  // *one* unreadable slot at the front inflated `dx()` by the ratio of the whole
  // elapsed stream time to the retained window: the axis in `x_bounds()` read the
  // stream's start rather than the window's, every measurement cut computed from it
  // came out too wide, and the readout disagreed with what the crosshair said
  // (`x_at()`/`neighbours()` go to the ring per sample and never saw the sentinel).
  // That is the "live trace is clipped, only the most recent part is visible, but
  // the crosshair still reads values there" defect.
  //
  // Clamping degrades the way the library's own `StreamSeries` documents for this
  // contract: a snapshot bounds the sample *count*, not the slots, so the oldest few
  // samples of a saturated window may read a newer value than the window names. One
  // frame's worth of samples at the very left edge is then drawn on a short plateau
  // of equal stamps -- monotone, so the binary searches still hold -- instead of the
  // whole trace's geometry collapsing.
  const u64 base = times_->base();
  const u64 produced = times_->produced();
  if (produced == 0) return 0.0;
  u64 slot = index < base ? base : index;
  if (slot >= produced) slot = produced - 1;
  f64 out = 0.0;
  times_->copy_from(slot, 1, &out);
  return out;
}

f64 ChannelBuffer::dx() const {
  const u64 n = pyramid_.size();
  if (n < 2) return dt_ > 0.0 ? dt_ : 1.0;
  const plot::Range b = x_bounds();
  const double span = b.max - b.min;
  if (!(span > 0.0) || !std::isfinite(span)) return dt_ > 0.0 ? dt_ : 1.0;
  return span / f64(n - 1);
}

f64 ChannelBuffer::x0() const {
  if (pyramid_.size() == 0) return epoch_;
  return epoch_ + window_x0_;
}

f64 ChannelBuffer::x_at(u64 i) const {
  if (i >= pyramid_.size()) return x0();
  // Sample 0 is served from the cache, never from the ring: it is the slot the producer is
  // overwriting, and `neighbours()` tests the asked-for x against this value, so a read one
  // ring-span too new made it refuse the whole visible range -- which left the renderer
  // gathering the entire ring at a coarse level and drawing it off the axis.
  if (i == 0) return x0();
  return epoch_ + time_of(absolute(i));
}

bool ChannelBuffer::neighbours(f64 x, u64& i0, u64& i1) const {
  const u64 n = pyramid_.size();
  if (n == 0) return false;
  if (x < x_at(0) || x > x_at(n - 1)) return false;
  // The time ring is monotonic, so the bracketing pair is a binary search rather
  // than a grid walk: after a lost packet the two samples either side of `x` are
  // not `x`/`dx` apart, and a grid would name the wrong pair.
  u64 lo = 0;
  u64 hi = n - 1;
  while (lo < hi) {
    const u64 mid = lo + (hi - lo) / 2;
    if (x_at(mid) < x) {
      lo = mid + 1;
    } else {
      hi = mid;
    }
  }
  i1 = lo;
  i0 = lo;
  if (x_at(lo) > x && lo > 0) i0 = lo - 1;
  return true;
}

u64 ChannelBuffer::snap_index(f64 x) const {
  u64 i0 = 0;
  u64 i1 = 0;
  if (!neighbours(x, i0, i1)) {
    if (pyramid_.size() == 0) return 0;
    return x < x_at(0) ? 0 : pyramid_.size() - 1;
  }
  if (i0 == i1) return i0;
  return (x - x_at(i0)) <= (x_at(i1) - x) ? i0 : i1;
}

plot::Range ChannelBuffer::x_bounds() const {
  const u64 n = pyramid_.size();
  if (n == 0) return plot::Range{epoch_, epoch_};
  return plot::Range{x0(), epoch_ + time_of(absolute(n - 1))};
}

plot::Range ChannelBuffer::y_bounds() const {
  const u64 n = pyramid_.size();
  if (n == 0) return plot::Range{0.0, 0.0};
  return reduce(0, n);
}

void ChannelBuffer::raw(u64 begin, u64 end, f32* out) const {
  const u64 n = pyramid_.size();
  const u64 b = std::min(begin, n);
  const u64 e = std::min(end, n);
  const u64 count = e > b ? e - b : 0;
  if (count > 0) values_->copy_from(absolute(b), count, out);
  for (u64 i = count; i < end - begin; ++i) out[i] = 0.0f;
}

void ChannelBuffer::gather(u32 level, u64 begin, u64 end, f64 x_origin, std::vector<f32>& out_xy,
                           plot::data::GatherMode mode) const {
  const u64 n = pyramid_.size();
  const u64 b = std::min(begin, n);
  const u64 e = std::min(std::max(end, b), n);
  if (b >= e) return;

  const u64 origin = pyramid_.window_base();
  const f64 base = x_origin;  // x is emitted relative to this (see the header)

  const auto emit_raw = [&](u64 from, u64 to) {
    for (u64 i = from; i < to; ++i) {
      f32 y = 0.0f;
      raw(i, i + 1, &y);
      out_xy.push_back(static_cast<f32>(x_at(i) - base));
      out_xy.push_back(y);
    }
  };
  const auto rel = [&](u64 i) { return static_cast<f32>(x_at(i) - base); };

  if (level == 0) {
    emit_raw(b, e);
    return;
  }
  level = std::min(level, plot::data::kMaxPyramidLevel);

  // One geometry element per *complete* bucket wholly inside the range. `first` is
  // a ceiling and `last` a floor, so no aggregate can reach outside [b, e) -- a
  // floor on `first` or a `last` that includes the bucket holding `e-1` would emit
  // aggregates covering samples the caller did not ask for (the library's B3 fix,
  // which this mirrors). The spans left at the two edges are drawn from raw
  // samples, so the geometry still covers the range exactly and a partial bucket is
  // drawn once, not twice.
  const u64 span = u64{plot::data::kBucket} << level;
  const u64 lbase = pyramid_.level_base(level);
  const u64 lend = lbase + pyramid_.level_size(level);
  const u64 ceil_div = (origin + b) / span + ((origin + b) % span == 0 ? u64{0} : u64{1});
  const u64 first = std::max(ceil_div, lbase);
  const u64 last = std::min((origin + e) / span, lend);
  if (first >= last) {
    emit_raw(b, e);
    return;
  }
  const u64 lead = first * span - origin;
  if (b < lead) emit_raw(b, std::min(lead, e));

  for (u64 bi = first; bi < last; ++bi) {
    const u32 count = pyramid_.bucket_count(level, bi);
    const u64 fs = bi * span - origin;  // >= 0: a stored bucket starts inside the window
    switch (mode) {
      case plot::data::GatherMode::Nearest: {
        const u64 centre = bi * span + span / 2 - origin;
        f32 y = 0.0f;
        raw(centre, centre + 1, &y);
        out_xy.push_back(rel(centre));
        out_xy.push_back(y);
        break;
      }
      case plot::data::GatherMode::Mean: {
        if (count == 0) break;  // no finite sample: no mean to draw
        const u64 centre = bi * span + span / 2 - origin;
        out_xy.push_back(rel(centre));
        out_xy.push_back(static_cast<f32>(pyramid_.bucket_sum(level, bi) / f64(count)));
        break;
      }
      default: {  // Envelope
        if (count == 0) break;
        const plot::data::MinMax m = pyramid_.bucket(level, bi);
        const u64 ls = std::min((bi + 1) * span, origin + e) - 1 - origin;
        out_xy.push_back(rel(fs));
        out_xy.push_back(m.min);
        out_xy.push_back(rel(ls));
        out_xy.push_back(m.max);
        break;
      }
    }
  }
  const u64 trail = last * span - origin;
  if (trail < e) emit_raw(trail, e);
}

// --- SlotSeries -------------------------------------------------------------

plot::Range SlotSeries::x_bounds() const {
  if (event_ != nullptr) return event_->x_bounds();
  return buffer_ != nullptr ? buffer_->x_bounds() : plot::Range{0.0, 1.0};
}

plot::Range SlotSeries::y_bounds() const {
  if (event_ != nullptr) return event_->y_bounds();
  return buffer_ != nullptr ? buffer_->y_bounds() : plot::Range{0.0, 0.0};
}

const plot::data::MinMaxPyramid& SlotSeries::pyramid() const {
  if (event_ != nullptr) return event_->pyramid();
  return buffer_ != nullptr ? buffer_->pyramid() : null_pyramid();
}

f64 SlotSeries::x_at(u64 i) const {
  if (event_ != nullptr) return event_->x_at(i);
  return buffer_ != nullptr ? buffer_->x_at(i) : 0.0;
}

bool SlotSeries::neighbours(f64 x, u64& i0, u64& i1) const {
  if (event_ != nullptr) return event_->neighbours(x, i0, i1);
  return buffer_ != nullptr && buffer_->neighbours(x, i0, i1);
}

u64 SlotSeries::snap_index(f64 x) const {
  if (event_ != nullptr) return event_->snap_index(x);
  return buffer_ != nullptr ? buffer_->snap_index(x) : 0;
}

void SlotSeries::gather(u32 level, u64 begin, u64 end, f64 x_origin, std::vector<f32>& out_xy,
                        plot::data::GatherMode mode) const {
  if (event_ != nullptr) {
    event_->gather(level, begin, end, x_origin, out_xy, mode);
    return;
  }
  if (buffer_ != nullptr) buffer_->gather(level, begin, end, x_origin, out_xy, mode);
}

void SlotSeries::copy_raw(u64 begin, u64 end, f32* out) const {
  if (event_ != nullptr) {
    event_->raw_values(begin, end, out);
    return;
  }
  if (buffer_ != nullptr) {
    buffer_->raw(begin, end, out);
    return;
  }
  for (u64 i = 0; i < end - begin; ++i) out[i] = 0.0f;
}

}  // namespace udpscope