Martino Ferrari
c2c8c596f3
`Series::gather()` stages x as f32 relative to `x0()` -- the start of the *retained window* -- so the value's ULP was a fraction of data the view need not even hold: 1.9 us over a 20 s ring. A readout quotes `x_at()` in exact f64, so at a 10 us view across 700 px that put more than a hundred pixels between where a sample is drawn and the x reported for it. The origin is now a parameter, placed before `out_xy` so every call site is a type error rather than a silent change of behaviour. `Renderer::draw_traces` passes the visible range's minimum, which makes the error a fixed 2^-24 of the span on screen -- 4e-5 of a pixel over a 700 px axis -- at every zoom. `stage_xy()` loses its second origin as a consequence: the gather origin and the draw origin are now always the same value, so the linear case hands the gathered buffer straight to `add_polyline_xy()` with no per-point copy. `Plot::crosshair_point()` names the query itself, the one origin that cannot lose precision for a caller that might. Guarded by `a_gather_origin_at_the_view_keeps_a_deep_zoom_on_the_sample_it_quotes`, which measures the error in pixels both ways: under 0.05 px with the view's minimum, over 10 px with `x0()`.
116 lines
5.7 KiB
C++
116 lines
5.7 KiB
C++
// SPDX-License-Identifier: MIT
|
|
// SPDX-FileCopyrightText: 2026 Martino Ferrari
|
|
|
|
#pragma once
|
|
#include <vector>
|
|
|
|
#include <plot/core/types.hpp>
|
|
#include <plot/data/array_series.hpp>
|
|
#include <plot/data/minmax_pyramid.hpp>
|
|
#include <plot/data/ring_buffer.hpp>
|
|
#include <plot/data/series.hpp>
|
|
|
|
namespace plot::data {
|
|
|
|
/// Single-producer / single-consumer streaming series.
|
|
///
|
|
/// The producer thread only ever calls push()/push(ys, n) (and produced()): those
|
|
/// touch the ring buffer and nothing else -- no lock, no pyramid, no GL, so the
|
|
/// acquisition thread can never stall the renderer. The consumer thread calls
|
|
/// acquire() to publish a coherent snapshot, then queries it.
|
|
///
|
|
/// Indexing is the *retained window published by the last acquire()*: size(),
|
|
/// reduce(), sum(), sumsq(), gather() and pyramid() all use it, and index 0 is
|
|
/// the oldest sample still held.
|
|
///
|
|
/// The pyramid behind those queries is indexed in absolute sample space, so a
|
|
/// window that slides only makes it drop the leading buckets the ring has
|
|
/// overwritten and append the samples produced since the last acquire(): acquire()
|
|
/// costs O(samples produced), not O(capacity()), and the aggregates that stay in
|
|
/// the window are never recomputed. The one exception is a producer that laps the
|
|
/// consumer -- more than capacity() samples between two acquires -- where the
|
|
/// samples in between were never appended and the pyramid restarts at the new
|
|
/// base, so that acquire re-reads the whole window.
|
|
///
|
|
/// x0() is therefore the x of the oldest retained sample, x_bounds().min == x0(),
|
|
/// and y_bounds() covers exactly the published samples. `streaming()` says true:
|
|
/// this is the window that slides, the one `XFollow::Live` follows.
|
|
///
|
|
/// THREADING NOTE. The ring overwrites the oldest samples -- that is what keeps a
|
|
/// live stream showing the newest data, and what the drop test pins down -- so a
|
|
/// producer can overwrite the slots a consumer is reading even while it holds a
|
|
/// snapshot. Those slots are relaxed atomics (see RingBuffer), so this is not a
|
|
/// data race and no value tears; what it is not is a frozen window.
|
|
///
|
|
/// The bound is the sample count, not the geometry. A snapshot count fixes *which
|
|
/// absolute indices* a read walks, not that the slots still hold those samples:
|
|
/// every push after the snapshot overwrites exactly one slot, the one belonging to
|
|
/// the sample capacity() positions behind it. So if the producer pushes k samples
|
|
/// while a read is in flight, the first k positions of that read window can hold
|
|
/// one of those k newer samples (or a still-newer sample, once the producer has
|
|
/// lapped again) instead of the sample the snapshot names. Nothing else moves: the
|
|
/// count, the absolute index -> x mapping and the newest end of the window are
|
|
/// derived from one published count, and the next acquire() slides the pyramid's
|
|
/// window to the base it observes (dropping the leading buckets, appending what is
|
|
/// new), so the pyramid never mixes two windows. Ingest with no reuse
|
|
/// underneath a read at all needs the producer to stop before unread data
|
|
/// (drop-on-full) or a claim protocol, i.e. a different contract from the one this
|
|
/// class specifies.
|
|
class StreamSeries final : public Series {
|
|
public:
|
|
/// `capacity` is the sample count to retain; it is rounded up to a power of
|
|
/// two by the ring (see `RingBuffer`), and `size()` saturates at that rounded
|
|
/// `capacity()` -- a `StreamSeries(100)` retains up to 128 samples. The
|
|
/// pyramid's level table is reserved for the rounded capacity up front, so a
|
|
/// saturated stream never grows it.
|
|
explicit StreamSeries(u64 capacity, f64 x0, f64 dx);
|
|
|
|
/// Producer thread only.
|
|
void push(f32 y);
|
|
/// Producer thread only.
|
|
void push(const f32* ys, u64 n);
|
|
/// Producer thread only: number of samples ever pushed (monotonic).
|
|
u64 produced() const;
|
|
/// Consumer thread only: snapshots the produced count and brings the pyramid
|
|
/// and bounds up to it.
|
|
void acquire();
|
|
/// Consumer thread only: copies the window published by the *last acquire()*
|
|
/// into a fresh `ArraySeries` with this window's `x0()`/`dx()`. The frozen
|
|
/// series never changes again, no matter what the producer pushes afterwards
|
|
/// -- it answers from its own copy, not from the ring -- which is what a
|
|
/// consumer that needs a stable window for longer than one read wants. (The
|
|
/// ring itself keeps overwriting; only the copy is frozen, so this costs one
|
|
/// window's worth of samples. It reads through `copy_raw()`, so like every
|
|
/// read it is a snapshot of *which* absolute indices to walk, not a guarantee
|
|
/// about slots a producer is overwriting -- call it on the consumer thread,
|
|
/// interleaved with `acquire()`, per the class note.)
|
|
ArraySeries freeze() const;
|
|
|
|
u64 size() const override;
|
|
f64 x0() const override;
|
|
f64 dx() const override;
|
|
Range x_bounds() const override;
|
|
Range y_bounds() const override;
|
|
const MinMaxPyramid& pyramid() const override;
|
|
void gather(u32 level, u64 begin, u64 end, f64 x_origin, std::vector<f32>& out_xy,
|
|
GatherMode mode = GatherMode::Envelope) const override;
|
|
/// Always true: this series' `x_bounds()` is the retained ring window, which
|
|
/// slides with every `acquire()` -- the property `XFollow::Live` follows.
|
|
bool streaming() const noexcept override { return true; }
|
|
|
|
protected:
|
|
void copy_raw(u64 begin, u64 end, f32* out) const override;
|
|
|
|
private:
|
|
RingBuffer<f32> ring_;
|
|
f64 x0_ = 0.0;
|
|
f64 dx_ = 1.0;
|
|
// The pyramid owns the published window: its window_base() is the absolute
|
|
// index of retained index 0 and its size()/window_end() are the snapshot's
|
|
// extent, so there is no second copy of that state to keep in step.
|
|
MinMaxPyramid pyramid_;
|
|
Range y_bounds_{};
|
|
};
|
|
|
|
} // namespace plot::data
|