Rate from the newest bucket, stamped when the counters are actually read

This commit is contained in:
flamingcow
2026-08-04 13:45:16 -07:00
parent 2035d42734
commit 4e60757937
2 changed files with 26 additions and 40 deletions
+22 -36
View File
@@ -6,7 +6,6 @@ import (
"net" "net"
"os" "os"
"os/signal" "os/signal"
"slices"
"strings" "strings"
"sync" "sync"
"sync/atomic" "sync/atomic"
@@ -85,9 +84,10 @@ type counterSet struct {
nic uint64 nic uint64
} }
func (d *direction) capture(t time.Time) counterSet { func (d *direction) capture() counterSet {
d.sampleDrops() d.sampleDrops()
return counterSet{t: t, s: d.snapshot(), drops: d.drops, nic: d.nic.Load()} s := d.snapshot()
return counterSet{t: time.Now(), s: s, drops: d.drops, nic: d.nic.Load()}
} }
// What someone testing a cable is asking, rather than how each failure happened // What someone testing a cable is asking, rather than how each failure happened
@@ -117,7 +117,6 @@ type rateWindow struct {
buf []counterSet buf []counterSet
idx int idx int
filled bool filled bool
scratch []float64
} }
func newRateWindow(n int) *rateWindow { func newRateWindow(n int) *rateWindow {
@@ -148,30 +147,17 @@ func (w *rateWindow) at(i int) counterSet {
return w.buf[i%len(w.buf)] return w.buf[i%len(w.buf)]
} }
// The median is steady against a bursty sender yet only ever a rate some bucket func (w *rateWindow) latest(rate func(prev, cur counterSet, secs float64) float64) float64 {
// actually measured, so a step change is shown as a step: the old value holds
// until half the ring has turned over and then the new one takes it, passing
// through at most the one bucket the crossing lands on. Averaging the ring
// instead would spend the whole span sliding through rates that never happened.
func (w *rateWindow) median(rate func(prev, cur counterSet, secs float64) float64) float64 {
n := w.count() n := w.count()
if n < 2 { if n < 2 {
return 0 return 0
} }
w.scratch = w.scratch[:0] prev, cur := w.at(n-2), w.at(n-1)
prev := w.at(0) secs := cur.t.Sub(prev.t).Seconds()
for i := 1; i < n; i++ { if secs <= 0 {
cur := w.at(i)
if secs := cur.t.Sub(prev.t).Seconds(); secs > 0 {
w.scratch = append(w.scratch, rate(prev, cur, secs))
}
prev = cur
}
if len(w.scratch) == 0 {
return 0 return 0
} }
slices.Sort(w.scratch) return rate(prev, cur, secs)
return w.scratch[len(w.scratch)/2]
} }
func txRatePPS(p, c counterSet, secs float64) float64 { func txRatePPS(p, c counterSet, secs float64) float64 {
@@ -248,7 +234,7 @@ func (d *direction) snapshot() sample {
// ring so the newest bucket never sits behind it. // ring so the newest bucket never sits behind it.
func (d *direction) reset() { func (d *direction) reset() {
d.mu.Lock() d.mu.Lock()
d.base = d.capture(time.Now()) d.base = d.capture()
d.win.push(d.base) d.win.push(d.base)
d.mu.Unlock() d.mu.Unlock()
@@ -347,11 +333,11 @@ func totalView(views []view) view {
return t return t
} }
func (d *direction) view(t time.Time) view { func (d *direction) view() view {
d.mu.Lock() d.mu.Lock()
defer d.mu.Unlock() defer d.mu.Unlock()
now := d.capture(t) now := d.capture()
p := d.prevConsole p := d.prevConsole
d.prevConsole = now d.prevConsole = now
@@ -367,9 +353,9 @@ func (d *direction) view(t time.Time) view {
return v return v
} }
func (d *direction) sample(t time.Time) { func (d *direction) sample() {
d.mu.Lock() d.mu.Lock()
d.win.push(d.capture(t)) d.win.push(d.capture())
d.mu.Unlock() d.mu.Unlock()
} }
@@ -386,10 +372,10 @@ func (d *direction) displayView(t time.Time) view {
if n >= 2 { if n >= 2 {
v.window = errsBetween(d.win.at(0), d.win.at(n-1)) v.window = errsBetween(d.win.at(0), d.win.at(n-1))
} }
v.txPPS = d.win.median(txRatePPS) v.txPPS = d.win.latest(txRatePPS)
v.rxPPS = d.win.median(rxRatePPS) v.rxPPS = d.win.latest(rxRatePPS)
v.txGbps = d.win.median(txRateGbps) v.txGbps = d.win.latest(txRateGbps)
v.rxGbps = d.win.median(rxRateGbps) v.rxGbps = d.win.latest(rxRateGbps)
d.mu.Unlock() d.mu.Unlock()
v.rxFrames = d.heldFrames.get(t, v.rxFrames) v.rxFrames = d.heldFrames.get(t, v.rxFrames)
@@ -586,8 +572,8 @@ const (
totalsHold = 50 * time.Millisecond totalsHold = 50 * time.Millisecond
) )
// One sampler for both directions, so their buckets share an instant and the // One sampler for both directions, so they are read back to back on one clock
// cable length, which needs a figure from each, never mixes two moments. // rather than drifting apart on two.
type sampler struct { type sampler struct {
dirs []*direction dirs []*direction
} }
@@ -599,9 +585,9 @@ func (s *sampler) run(done *atomic.Bool, startTx <-chan struct{}) {
defer tick.Stop() defer tick.Stop()
for !done.Load() { for !done.Load() {
now := <-tick.C <-tick.C
for _, d := range s.dirs { for _, d := range s.dirs {
d.sample(now) d.sample()
} }
} }
} }
@@ -764,7 +750,7 @@ func run(aName, bName string, nsPerM float64) error {
// Length needs both directions, so every row is sampled before any of // Length needs both directions, so every row is sampled before any of
// them is printed. // them is printed.
for i, d := range dirs { for i, d := range dirs {
rows[i] = d.view(now) rows[i] = d.view()
} }
length := "-" length := "-"
if m, ok := cableMetres(rows, nsPerM); ok { if m, ok := cableMetres(rows, nsPerM); ok {
+1 -1
View File
@@ -16,7 +16,7 @@ func TestResetDoesNotUnderflowTotals(t *testing.T) {
d.rxStats[0].frames.Store(100) d.rxStats[0].frames.Store(100)
d.rxStats[0].bytes.Store(6400) d.rxStats[0].bytes.Store(6400)
d.sample(time.Now()) d.sample()
d.rxStats[0].frames.Store(150) d.rxStats[0].frames.Store(150)
d.rxStats[0].bytes.Store(9600) d.rxStats[0].bytes.Store(9600)