#pragma once #include #include #include #include #include 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& out_xy) const override; protected: void copy_raw(u64 begin, u64 end, f32* out) const override; private: RingBuffer 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