udpscope/src/acquisition.hpp

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

#pragma once

#include "capture.hpp"
#include "config.hpp"
#include "model.hpp"
#include "trigger.hpp"

#include "source/data_source.hpp"

#include <atomic>
#include <chrono>
#include <cstdint>
#include <limits>
#include <memory>
#include <thread>
#include <vector>

namespace udpscope {

/// Owns the data source and the producer thread that feeds the model's
/// acquisition buffer. Everything it touches from the thread is serialised on
/// `ScopeModel::mutex()`, so a UI-side structural change cannot race a frame.
class Acquisition : private SourceSink {
 public:
  Acquisition(ScopeModel& model, const Config& cfg);
  ~Acquisition();

  Acquisition(const Acquisition&) = delete;
  Acquisition& operator=(const Acquisition&) = delete;

  /// Opens the source (no socket until the first poll) and starts the thread.
  /// Throws std::runtime_error when the source cannot be created.
  void start();
  /// Stops the thread and destroys the client. Idempotent.
  void stop();

  void set_paused(bool paused) noexcept { paused_.store(paused, std::memory_order_relaxed); }
  bool paused() const noexcept { return paused_.load(std::memory_order_relaxed); }

  /// Turns the packet capture on or off -- the whole GTK side of the feature. The
  /// path is built here, once, *before* the flag is published, so the producer reads
  /// a complete string whenever it reads a true flag. `slots` is the ring's size;
  /// zero means the 1000-slot default. The producer does the open or the close at the
  /// top of the next datagram, so a toggle costs no lock and no second thread.
  void set_capture_want(bool want, std::uint32_t slots = 0);
  bool capture_want() const noexcept {
    return capture_want_.load(std::memory_order_acquire);
  }

  /// Asks for one **forced trigger acquisition** -- the trigger panel's Force button. The producer
  /// consumes the request on its next packet and opens a capture at that packet's newest stamp,
  /// which the engine's ordinary latch then publishes with the configured pre/post window. The
  /// request is dropped rather than retried when there is nothing to force: trigger mode off, no
  /// channel resolved, the engine disarmed, or a capture already filling (a capture in flight *is*
  /// the acquisition being asked for).
  ///
  /// **Write it under `ScopeModel::mutex()`, together with the `armed` flag the click also sets.**
  /// The producer holds that lock for a whole packet -- it reads the panel's `armed` at the top of
  /// one and consumes this request at the bottom -- so a request written between the two would be
  /// read by a packet whose engine was still disarmed, and then dropped.
  void request_force() noexcept { force_want_.store(true, std::memory_order_release); }

  /// The source this acquisition drives, for the UI's capability panels: it may only call
  /// `post()` on it. Null until `start()`.
  DataSource* source() noexcept { return source_.get(); }

  /// The capture, for the popover's row and the probe. Counters and errors are read
  /// cross-thread through PacketCapture's own published fields.
  const PacketCapture& capture() const noexcept { return capture_; }

 private:
  void run();
  void on_config(const std::vector<SignalInfo>& signals, std::uint8_t publish_mode) override;
  void on_data(const Frame& frame) override;
  void on_event(SourceEvent event, const char* detail) override;
  /// Drops the producer state that is per *connection*: the counter mark, the time
  /// anchors and the clock choice. A reconnect re-sends the CONFIG and the stream may
  /// have restarted its numbering; carrying the old session's marks across makes the
  /// first new frames read as late, as glitches or as an outage, which is recovery
  /// failing exactly when it matters. Called on the producer thread, from on_event().
  void reset_session_state() noexcept;
  /// The producer side of the toggle: open on a rising edge, close on a falling one,
  /// then store if enabled. Every file operation happens here, on this thread.
  void on_datagram(const std::uint8_t* bytes, std::size_t len, double recv_time);

  /// Fills `stamps_` with this frame's base-tick timestamps, in stream seconds,
  /// and returns how many were written (`frame.num_samples` unless the frame is
  /// malformed).
  std::uint32_t build_stamps(const Frame& frame, int time_signal,
                             bool counter_gap_refused);

  /// Pushes NaN for the `missing` packets' worth of samples that were never delivered,
  /// advances the decimation accumulators and tick_, records the hole in the model, and
  /// bumps the hole counters. Called before the frame's own samples so the gap sits
  /// *ahead* of the data that revealed it.
  void reserve_hole(std::uint32_t missing, const Partition& part,
                    const std::vector<ChannelInfo>& channels, BufferStore& store, u64 g);

  /// Writes a late frame's values into the NaN-filled holes their counter addresses,
  /// amending the value ring directly (producer path) and queuing pyramid amends for
  /// `ChannelBuffer::acquire()` to apply on the consumer thread.
  void backfill_from(const Frame& frame, const Partition& part,
                     const std::vector<ChannelInfo>& channels, BufferStore& store, u64 g);

  /// Keeps the wire's own cadence up to date from the arrival times, in the modes where
  /// `build_stamps()`'s arrival branch does not (see the definition).
  void note_frame_arrival(double recv_time) noexcept;

  /// Whether the counter's claim of `missing` lost packets can be believed *and* booked: the
  /// arrival clock has to corroborate it and the ring has to be able to hold it. See the
  /// definition for both tests and for what believing a claim that fails them looks like.
  bool gap_plausible(std::uint32_t missing, double elapsed, const Partition& part, u64 g) const;

  ScopeModel& model_;
  Config cfg_;
  std::unique_ptr<DataSource> source_;
  std::thread thread_;
  std::atomic<bool> running_{false};
  std::atomic<bool> paused_{false};
  PacketCapture capture_;
  std::atomic<bool> capture_want_{false};
  /// The pending Force request (see `request_force()`): a flag and not a queue, because two clicks
  /// inside one packet are one acquisition, which is what a momentary button means.
  std::atomic<bool> force_want_{false};
  std::uint32_t capture_slots_ = 1000;
  /// Written once, by the GTK thread, before `capture_want_` is ever published true;
  /// the producer reads it only after the flag, so no lock is needed.
  std::string capture_path_;
  TriggerEngine engine_;

  std::uint64_t tick_ = 0;        ///< base ticks consumed since the epoch
  std::uint64_t last_tick_ = ~std::uint64_t{0};
  double last_stamp_ = 0.0;       ///< last relative time written (monotonic guard)
  /// Last accepted stream-time stamp: a "time signal" whose values do not advance
  /// across packets (a producer that restarts them per packet, or a mislabelled
  /// signal) would freeze the whole time base -- the trigger's hold-off clock
  /// included -- so the stream source is dropped in favour of arrival time.
  double last_stream_anchor_ = 0.0;
  bool have_stream_anchor_ = false;
  /// Consecutive packets whose stream-time stamps failed those checks. A single bad
  /// packet is a glitch in the producer's clock and is bridged at the declared cadence;
  /// only a run of them means the declared signal is not a time base, and only then is
  /// the arrival clock taken. Any accepted packet resets it.
  unsigned stamp_failures_ = 0;
  /// Arrival-clock origin: the first packet's receive time, so arrival stamps are
  /// small seconds (the time ring stores f32, which cannot hold Unix epochs).
  double arrival_epoch_ = 0.0;
  bool have_arrival_epoch_ = false;
  /// Smoothed interval between two frames' arrivals, and the previous arrival: the
  /// spine of the arrival-anchored clock (see `build_stamps`). In relative mode no frame
  /// reaches that branch, so `on_data` keeps the same pair up to date itself -- the
  /// reservation's plausibility test below needs a measured cadence whatever clock the
  /// user chose, and the two writers are mutually exclusive by mode.
  double arrival_interval_ = 0.0;
  double arrival_prev_ = 0.0;
  bool have_arrival_interval_ = false;
  /// The previous *accepted* frame's receive time, for the one measurement the reservation
  /// test needs: how much wall time actually passed. Kept apart from `arrival_prev_` on
  /// purpose -- that one is the arrival clock's own bookkeeping and is re-anchored by the
  /// clock-choice code, while this is a bare witness of the wire and must never be moved by
  /// a mode change.
  double prev_frame_recv_ = 0.0;
  bool have_prev_frame_recv_ = false;
  /// The stream's time source, chosen once and sticky: 0 undecided, 1 the declared
  /// time signal, 2 arrival. Re-evaluated only on a CONFIG. A per-frame choice
  /// interleaved two timelines in one ring -- some frames stamped from a declared
  /// signal carrying volt-like values, others from the arrival clock -- which is a
  /// time base no reader can make sense of (see `build_stamps`).
  int time_source_ = 0;
  /// Relative-time mode's clock: seconds since the stream began, accumulated at the interval
  /// the plan declares. Advanced by `lost * batch_` samples' worth for every packet the counter
  /// reports missing (before the frame's own samples) and then by the frame's own `n`, so it is
  /// exactly `(counter * batch + sample_index) * dt` for a constant batch and degrades
  /// gracefully when a frame's length disagrees.
  ///
  /// It is a **time**, not a position multiplied by the current plan's interval, and the
  /// difference is the whole of one defect: a plan is rebuilt whenever the channel list changes
  /// (a CONFIG, an enable, a session's math channel) and `Partition::base_dt` -- which is the
  /// 1 kHz fallback until a *streamed* channel declares a rate -- can therefore change under a
  /// running stream. With `position * dt` a plan that shortened the interval made every later
  /// stamp *behind* the one already written, which the frame loop's monotonic clamp then held
  /// for the rest of the run: the timeline stood still, the trigger's `history` never reached
  /// its pre-window, and nothing was drawn (see `on_data`). Accumulated, a plan change is what
  /// it is -- a slope change -- and the clock cannot go backwards.
  double rel_base_s_ = 0.0;
  /// Samples per DATA packet, fixed by the first frame. Only ever needed to size what is
  /// *absent*: a delivered frame carries its own length. 0 until the first frame.
  std::uint32_t batch_ = 0;
  /// Latches the one warning about a frame whose length disagrees with the stream's first.
  bool warned_batch_ = false;
  /// The newest counter seen, and the hole's address. Tracked here rather than inferred from
  /// `frame->lost` because a *late* frame's counter is behind the sequence and must not move
  /// the mark forward.
  std::uint32_t last_counter_ = 0;
  bool have_counter_ = false;
  /// Last absolute stamp written, epoch-relative, so a reserved hole can continue the
  /// absolute clock at the declared cadence. The packets that would have carried the real
  /// stamps never arrived, so a cadence continuation is the only honest value available.
  double last_abs_stamp_ = 0.0;
  /// Warn-once latch for a loss past the reservation cap. A member, not a function-local
  /// `static`, matching the latch Task 2 added for the same reason.
  bool warned_outage_ = false;
  /// Warn-once latch for a reservation the arrival clock did not corroborate, or that the ring
  /// could not hold (see `gap_plausible()`). Once per connection, like the two above.
  bool warned_phantom_gap_ = false;
  /// Warn-once latch for the one bridged packet whose time signal glitched.
  bool warned_bridge_ = false;
  /// Warn-once latch for the declared time signal failing for good, so the arrival clock is
  /// taken instead.
  bool warned_time_source_ = false;
  /// Warn-once latch for a frame whose every stamp was held back by the frame loop's monotonic
  /// clamp. The clamp exists for a stream clock that glitches; a *whole frame* behind the clock is
  /// the timeline standing still, and while it lasts nothing is drawn and no trigger can latch.
  bool warned_clock_stall_ = false;
  /// Whether this connection has adopted its clock origin (`ScopeModel::set_epoch_if_unset()`).
  /// A state and not `epoch() == 0.0`, which re-fires in relative mode -- where the first frame's
  /// stamp *is* zero -- and moves the origin under the timeline. Per connection, like every other
  /// piece of session state here: a new session's clock is a new clock.
  bool epoch_adopted_ = false;
  /// The re-plan generation this object last reacted to. The buffer is re-planned whenever the
  /// model rebuilds it, and the state tied to the old store (`tick_` and the accumulators) is
  /// reset when this falls behind `ScopeModel::plan_generation()`.
  unsigned seen_plan_generation_ = 0;
  std::vector<double> stamps_;
  std::vector<double> abs_stamps_;
  /// The decimation filter's state: a running sum per channel (and one for the shared
  /// clock) over the group of base samples that becomes one stored sample. A boxcar
  /// average, which is what turns decimation into a *resampling* rather than a stride:
  /// its first null sits at the stored rate, so what would fold back is attenuated
  /// instead of reappearing as a false low frequency. Cleared whenever the partition
  /// changes, since the group length is `time_stride`.
  std::vector<double> value_acc_;
  /// The channels' values for the current *stored* slot, which the math channels are evaluated from
  /// once the slot's samples have all been pushed. One entry per channel, refilled every slot.
  std::vector<f32> math_sample_;
  std::vector<std::uint64_t> value_acc_n_;
  /// Per channel, `time_stride * stride[i] - 1`: the group boundary as a **mask**, so the
  /// innermost loop tests `(tick & mask) == mask` instead of `tick % period == period - 1`.
  /// The divisor is a runtime value, so that modulo compiled to a real 64-bit `div` once per
  /// channel per *wire* sample -- the filter reads every sample by design, so it was the
  /// producer's largest avoidable cost. Both factors are powers of two (`plan_partition()`
  /// doubles `time_stride` from 1, and every `stride[i]` is 1), which is what makes the mask
  /// equivalent; it is sized and filled with the accumulators below, so a re-plan cannot
  /// leave it addressing the previous partition's groups.
  std::vector<std::uint64_t> acc_mask_;
  double time_acc_ = 0.0;
  double abs_time_acc_ = 0.0;
  std::uint64_t time_acc_n_ = 0;
  /// The partition the accumulators were sized for, so a re-plan resets them rather than
  /// averaging across the boundary.
  std::uint64_t acc_stride_ = 0;
  std::size_t acc_channels_ = 0;
  std::chrono::steady_clock::time_point started_{};
};

}  // namespace udpscope