Martino Ferrari
d9909f2115
cover_buckets chose one level per pass, so an odd first whole bucket (half of all measurement ranges) never reached a level above 0 and folded every remaining bucket at level 0 -- 15624 entries for a 1e6-sample pyramid. Pick the largest aligned, in-range, materialised level at each cursor instead; the new cost test counts folded entries through a PLOT_PYRAMID_COUNT_COVERS-only hook (absent from normal builds) and fails on the old body. Ring slots are now relaxed std::atomic: overwrite-on-full means the producer reuses the slot of a snapshot's oldest sample while the consumer reads it, which was a data race, not just a stale value. copy_from loads explicitly instead of std::copy; push stays lock-free. A new producer/consumer case is the one TSan flags without this. Also extends the wrapped-window test with x0/bounds/pyramid/sum/sumsq/gather checks, and corrects the StreamSeries threading note to the sample-count bound.
86 lines
3.8 KiB
C++
86 lines
3.8 KiB
C++
#pragma once
|
|
#include <vector>
|
|
|
|
#include <plot/core/types.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. Once the producer has written more than
|
|
/// capacity() samples the oldest ones are gone, so acquire() notices that the
|
|
/// window's base moved and rebuilds the pyramid from the retained samples -- an
|
|
/// incremental append would splice a dropped prefix onto a retained suffix and
|
|
/// mis-align every bucket (bucket boundaries are absolute, not relative to the
|
|
/// window). Until the ring saturates, acquire() is O(new samples); once it wraps,
|
|
/// an acquire whose base moved costs O(capacity()).
|
|
///
|
|
/// x0() is therefore the x of the oldest retained sample, x_bounds().min == x0(),
|
|
/// and y_bounds() covers exactly the published samples.
|
|
///
|
|
/// 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 the one published count, and the next acquire() rebuilds from the
|
|
/// base it observes, 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:
|
|
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();
|
|
|
|
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, std::vector<f32>& out_xy) const override;
|
|
|
|
protected:
|
|
void copy_raw(u64 begin, u64 end, f32* out) const override;
|
|
|
|
private:
|
|
RingBuffer<f32> ring_;
|
|
f64 x0_ = 0.0;
|
|
f64 dx_ = 1.0;
|
|
u64 base_ = 0; // absolute index of retained index 0 in the published snapshot
|
|
u64 end_ = 0; // absolute index one past the published snapshot
|
|
MinMaxPyramid pyramid_;
|
|
Range y_bounds_{};
|
|
};
|
|
|
|
} // namespace plot::data
|