#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. /// /// 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. /// /// 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: 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; // 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