MARTe-Integrated-Components/Common/Client/go/wshub/ringbuf.go

package wshub

import (
	"log"
	"math"
	"sync"
)

// sigRing is a fixed-capacity circular buffer storing (time, value) pairs.
// Writes come from the Hub.Run() goroutine; reads come from HTTP handler goroutines.
// The embedded RWMutex protects concurrent access.
type sigRing struct {
	mu         sync.RWMutex
	t, v       []float64
	cap        int
	head, size int // next write position; current fill

	// bucket is how many source samples collapse into one min/max pair on the
	// way in. 1 stores the stream verbatim. Raising it trades resolution for
	// the timespan a fixed capacity covers, which is what lets a long display
	// window fit in a fixed per-signal memory budget.
	bucket int
	// In-progress bucket. accN counts source samples seen since the last pair
	// was emitted; the four acc fields are the extrema and when they occurred.
	accN             int
	accTMin, accVMin float64
	accTMax, accVMax float64

	// Source-sample accounting, kept because size and the stored timespan no
	// longer give the source rate once bucket > 1. Reset every
	// srcRateWindowSec so a producer restart or a rate change is not averaged
	// against the whole run.
	srcCount     int64
	srcT0, srcT1 float64
	haveSrc      bool
}

// srcRateWindowSec bounds how long a source-rate measurement accumulates before
// starting over. Long enough to average out per-frame jitter, short enough that
// a rate change is reflected within a few seconds.
const srcRateWindowSec = 10.0

func newSigRing(capacity int) *sigRing {
	return &sigRing{
		t:      make([]float64, capacity),
		v:      make([]float64, capacity),
		cap:    capacity,
		bucket: 1,
	}
}

// write appends (tArr[i], vArr[i]) pairs, overwriting oldest entries when full.
// With bucket > 1 each group of bucket samples contributes only its minimum and
// its maximum, in the order the two occurred.
func (rb *sigRing) write(tArr, vArr []float64) {
	rb.mu.Lock()
	defer rb.mu.Unlock()

	if n := len(tArr); n > 0 {
		if !rb.haveSrc || tArr[n-1]-rb.srcT0 > srcRateWindowSec || tArr[0] < rb.srcT0 {
			rb.srcT0, rb.srcCount, rb.haveSrc = tArr[0], 0, true
		}
		rb.srcT1 = tArr[n-1]
		rb.srcCount += int64(n)
	}

	if rb.bucket <= 1 {
		for i := 0; i < len(tArr); i++ {
			rb.pushLocked(tArr[i], vArr[i])
		}
		return
	}
	for i := 0; i < len(tArr); i++ {
		t, v := tArr[i], vArr[i]
		if rb.accN == 0 {
			rb.accTMin, rb.accVMin, rb.accTMax, rb.accVMax = t, v, t, v
		} else {
			if v < rb.accVMin {
				rb.accTMin, rb.accVMin = t, v
			}
			if v > rb.accVMax {
				rb.accTMax, rb.accVMax = t, v
			}
		}
		rb.accN++
		if rb.accN >= rb.bucket {
			rb.flushBucketLocked()
		}
	}
}

func (rb *sigRing) pushLocked(t, v float64) {
	rb.t[rb.head] = t
	rb.v[rb.head] = v
	rb.head = (rb.head + 1) % rb.cap
	if rb.size < rb.cap {
		rb.size++
	}
}

// flushBucketLocked emits the accumulated extrema oldest-first. Time order
// matters: every read binary-searches rb.t, so the stored timestamps must stay
// non-decreasing.
func (rb *sigRing) flushBucketLocked() {
	if rb.accN == 0 {
		return
	}
	if rb.accTMin <= rb.accTMax {
		rb.pushLocked(rb.accTMin, rb.accVMin)
		rb.pushLocked(rb.accTMax, rb.accVMax)
	} else {
		rb.pushLocked(rb.accTMax, rb.accVMax)
		rb.pushLocked(rb.accTMin, rb.accVMin)
	}
	rb.accN = 0
}

// setBucket changes the min/max reduction applied to incoming samples and
// reports whether it changed. Samples already stored keep the resolution they
// were written at; the ring converges on the new one as it rolls.
func (rb *sigRing) setBucket(n int) bool {
	if n < 1 {
		n = 1
	}
	rb.mu.Lock()
	defer rb.mu.Unlock()
	if n == rb.bucket {
		return false
	}
	// Emit what the old bucket had collected rather than dropping it.
	rb.flushBucketLocked()
	rb.bucket = n
	return true
}

func (rb *sigRing) bucketSize() int {
	rb.mu.RLock()
	defer rb.mu.RUnlock()
	return rb.bucket
}

// sourceRate is the measured rate of the incoming stream in samples per second,
// or 0 while there is too little to extrapolate from. Unlike stats() it counts
// source samples, so it is unaffected by bucketing.
func (rb *sigRing) sourceRate() float64 {
	rb.mu.RLock()
	defer rb.mu.RUnlock()
	if rb.srcCount < 2 || rb.srcT1 <= rb.srcT0 {
		return 0
	}
	return float64(rb.srcCount-1) / (rb.srcT1 - rb.srcT0)
}

// stats reports the current fill and the timespan it covers, so callers can
// estimate the stream's sample rate without copying the data out.
func (rb *sigRing) stats() (count int, span float64) {
	rb.mu.RLock()
	defer rb.mu.RUnlock()
	if rb.size < 2 {
		return rb.size, 0
	}
	start := 0
	if rb.size == rb.cap {
		start = rb.head
	}
	oldest := rb.t[start]
	newest := rb.t[(start+rb.size-1)%rb.cap]
	return rb.size, newest - oldest
}

func (rb *sigRing) capacity() int {
	rb.mu.RLock()
	defer rb.mu.RUnlock()
	return rb.cap
}

// ─── Ring tuning ─────────────────────────────────────────────────────────────

// ringHeadroom oversizes a reduced ring's span. It absorbs rate jitter and
// keeps the tail of a trigger window in the buffer long enough for the capture
// to read it. It applies only once the window no longer fits verbatim: at the
// boundary, spending a whole extra bucket step to buy 25 % more span would cost
// half the resolution.
const ringHeadroom = 1.25

// ringTuneIntervalSec throttles the retune sweep. The source rate only settles
// once data flows, so the sweep repeats rather than running once.
const ringTuneIntervalSec = 1.0

// defaultLiveWindowSec is the window assumed when no client has said what it is
// displaying — the native clients never do, and a browser has not yet at the
// moment the first samples land.
const defaultLiveWindowSec = 10.0

// ringBucketFor is how many source samples must collapse into one min/max pair
// for `window` seconds at `rate` samples/s to fit in `capacity` points.
//
//	rate*window <= capacity → 1, the buffer stays verbatim and reaches further
//	                          back than the window, which is free zoom headroom
//	rate*window >  capacity → >1, so the whole window fits at reduced resolution
//
// A bucket costs two points (its minimum and its maximum), hence the factor 2.
func ringBucketFor(rate, window float64, capacity int) int {
	if capacity <= 0 || rate <= 0 || window <= 0 {
		return 1
	}
	need := rate * window
	if need <= float64(capacity) {
		return 1
	}
	return int(math.Ceil(2 * need * ringHeadroom / float64(capacity)))
}

// ringCoverage is how many source samples a ring of `capacity` points holds at
// the given bucket. A bucket of 2 stores both of its samples, so it covers no
// more ground than a bucket of 1.
func ringCoverage(bucket, capacity int) int {
	if bucket <= 2 {
		return capacity
	}
	return capacity / 2 * bucket
}

// captureLagSec is how much further back than the window itself a ring has to
// reach to deliver a capture of it.
//
// A capture is not read out when its last sample arrives but captureMarginSec
// later, and then only on the next push tick — so by the time the window is
// extracted, its oldest sample is that much deeper in the ring. A ring holding
// exactly the window has already overwritten the front of its own capture, which
// is what made every shot at a short window come back missing its head. The
// pre/post split does not enter into it: the harvest is a post-window after the
// trigger and the read reaches a pre-window before it, so the two sum to the
// window whatever the split.
const captureLagSec = captureMarginSec + 1.0/30.0

// activeWindowSec is the timespan the buffers must cover. An armed trigger owns
// it: its pre-window has to already be in the ring when the trigger fires or
// there is nothing to back-fill the capture from. Otherwise it is the widest
// window any connected client is displaying.
func (h *Hub) activeWindowSec() float64 {
	if h.trigger != nil && h.trigger.Active() {
		if cfg := h.trigger.Config(); cfg.windowSec > 0 {
			return cfg.windowSec + captureLagSec
		}
	}
	widest := 0.0
	for c := range h.clients {
		if w := c.displayWindowSec(); w > widest {
			widest = w
		}
	}
	if widest <= 0 {
		return defaultLiveWindowSec
	}
	return widest
}

// retuneRings keeps every ring matched to the window being displayed: grown
// towards the per-signal budget, and bucketed so the window fits inside it.
//
// A fixed sample-count ring covers a fraction of a second at a megasample rate,
// which is why long windows used to come back with only their tail populated —
// in live mode as much as under a trigger. Spending the budget on min/max pairs
// rather than on a bigger allocation is what makes an arbitrarily long window
// work within a fixed memory bound.
//
// Called from Hub.Run() only, so reading h.clients here needs no lock.
func (h *Hub) retuneRings(nowSec float64) {
	if nowSec < h.ringTuneAt {
		return
	}
	h.ringTuneAt = nowSec + ringTuneIntervalSec

	window := h.activeWindowSec()
	if window <= 0 {
		return
	}
	// The archive answers for the same window as the rings — it is what a zoom
	// or a capture falls back on once they have rolled past it — so it is sized
	// from the same number.
	if h.hist.setWindow(window) {
		if msg := h.buildHistoryInfoMsg(); msg != nil {
			h.broadcast(msg)
		}
	}
	budget := h.ringBudget()

	h.ringsMu.RLock()
	keys := make([]string, 0, len(h.rings))
	rings := make([]*sigRing, 0, len(h.rings))
	for k, rb := range h.rings {
		keys = append(keys, k)
		rings = append(rings, rb)
	}
	h.ringsMu.RUnlock()

	for i, rb := range rings {
		rate := rb.sourceRate()
		if rate <= 0 {
			continue
		}
		// Claim the whole budget before deciding on a bucket: memory is what
		// buys resolution, so it is spent first and reduced from only if the
		// window still does not fit.
		if rb.capacity() < budget && rate*window > float64(rb.capacity()) {
			rb.grow(budget)
		}
		cur := rb.bucketSize()
		need := rate * window
		covered := float64(ringCoverage(cur, rb.capacity()))
		// Hysteresis. Retuning up and retuning down must not share a threshold:
		// a bucket step doubles or halves the span, so a rate jittering across
		// the boundary would otherwise flip the resolution every second. Hold
		// the current bucket while it covers the window without covering more
		// than twice it.
		if covered >= need && covered <= 2*need {
			continue
		}
		want := ringBucketFor(rate, window, rb.capacity())
		if !rb.setBucket(want) {
			continue
		}
		if want > 1 {
			log.Printf("hub: ring %s stores min/max over %d samples: %.0f s at %.0f kSps does not fit in %d points",
				keys[i], want, window, rate/1e3, rb.capacity())
		} else {
			log.Printf("hub: ring %s back to full resolution: %.0f s at %.0f kSps fits in %d points",
				keys[i], window, rate/1e3, rb.capacity())
		}
	}
}

// grow enlarges the buffer to newCap, keeping every sample it currently holds.
// Shrinking is refused: it would discard history a pending capture may need.
func (rb *sigRing) grow(newCap int) bool {
	rb.mu.Lock()
	defer rb.mu.Unlock()
	if newCap <= rb.cap {
		return false
	}
	nt := make([]float64, newCap)
	nv := make([]float64, newCap)
	start := 0
	if rb.size == rb.cap {
		start = rb.head
	}
	for i := 0; i < rb.size; i++ {
		p := (start + i) % rb.cap
		nt[i], nv[i] = rb.t[p], rb.v[p]
	}
	rb.t, rb.v = nt, nv
	rb.cap = newCap
	rb.head = rb.size // size < newCap, so no wrap
	return true
}

// slice returns copies of all (t, v) pairs whose timestamp falls in [t0, t1].
// The returned slices are safe to use after the call without holding any lock.
func (rb *sigRing) slice(t0, t1 float64) ([]float64, []float64) {
	rb.mu.RLock()
	defer rb.mu.RUnlock()

	if rb.size == 0 {
		return nil, nil
	}
	start := 0
	if rb.size == rb.cap {
		start = rb.head
	}
	physAt := func(k int) int { return (start + k) % rb.cap }

	// Binary search for t0
	lo, hi := 0, rb.size
	for lo < hi {
		m := (lo + hi) >> 1
		if rb.t[physAt(m)] < t0 {
			lo = m + 1
		} else {
			hi = m
		}
	}
	kStart := lo

	// Binary search for t1
	lo, hi = kStart, rb.size
	for lo < hi {
		m := (lo + hi) >> 1
		if rb.t[physAt(m)] <= t1 {
			lo = m + 1
		} else {
			hi = m
		}
	}
	kEnd := lo

	n := kEnd - kStart
	if n <= 0 {
		return nil, nil
	}
	outT := make([]float64, n)
	outV := make([]float64, n)
	for i := 0; i < n; i++ {
		p := physAt(kStart + i)
		outT[i] = rb.t[p]
		outV[i] = rb.v[p]
	}
	return outT, outV
}