From c94b7ae33d32e090e78c830e0fb95d0ad50e757d Mon Sep 17 00:00:00 2001
From: flamingcow <git@flamingcow.io>
Date: Fri, 31 Jul 2026 15:45:03 -0700
Subject: [PATCH] Show median bucket rates, slide the error window, and poll
 sysfs off the draw loop

---
 counters.go |  53 ++++++++
 main.go     | 345 ++++++++++++++++++++--------------------------------
 2 files changed, 186 insertions(+), 212 deletions(-)

diff --git a/counters.go b/counters.go
index b2c8b1f..daf3659 100644
--- a/counters.go
+++ b/counters.go
@@ -4,6 +4,8 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"sync/atomic"
+	"time"
 )
 
 // Split by direction of travel so each counter belongs to exactly one
@@ -58,3 +60,54 @@ func readNIC(ifname string) nicCounters {
 	c.carrierDown, _ = readUint(base + "/carrier_down_count")
 	return c
 }
+
+// A poll is thirty-odd sysfs attributes and each one costs microseconds of open
+// and read, so it runs here rather than on the goroutine that has a frame to
+// draw every sixteen milliseconds. Off that path the interval can be short
+// enough that an unplugged cable shows up as a link error while the hand is
+// still on it.
+const nicInterval = 200 * time.Millisecond
+
+type nicPoller struct {
+	txName, rxName string
+	total          *atomic.Uint64
+	raw            uint64
+}
+
+func (p *nicPoller) read() uint64 {
+	tx := readNIC(p.txName)
+	rx := readNIC(p.rxName)
+	return tx.tx + rx.rx + tx.carrierDown
+}
+
+// Sysfs nic counters run from boot and restart from zero whenever the driver
+// resets its statistics, so only their forward motion is accumulated. Taking
+// raw differences instead charges a boot's worth of errors to the first sample
+// and turns a reset into a near-2^64 underflow.
+func (p *nicPoller) poll() {
+	raw := p.read()
+	if raw > p.raw {
+		p.total.Add(raw - p.raw)
+	}
+	p.raw = raw
+}
+
+// Whatever the interfaces counted before now is not ours, so the first poll has
+// something to measure forward motion from and adds nothing.
+func (p *nicPoller) prime() {
+	p.raw = p.read()
+}
+
+// Held until the test starts, since polling before the baseline is primed would
+// charge everything the interfaces counted since boot to the first sample.
+func (p *nicPoller) run(done *atomic.Bool, startTx <-chan struct{}) {
+	<-startTx
+
+	tick := time.NewTicker(nicInterval)
+	defer tick.Stop()
+
+	for !done.Load() {
+		<-tick.C
+		p.poll()
+	}
+}
diff --git a/main.go b/main.go
index f1c2cc3..f5d6566 100644
--- a/main.go
+++ b/main.go
@@ -3,10 +3,10 @@ package main
 import (
 	"flag"
 	"fmt"
-	"math"
 	"net"
 	"os"
 	"os/signal"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -50,94 +50,14 @@ type direction struct {
 	probeRxFD int
 	cable     *cableStats
 
-	prevConsole sample
+	prevConsole counterSet
 	win         *rateWindow
-	est         *rateEstimators
 	heldFrames  heldValue
 	heldSent    heldValue
 	drops       uint64
-	errBase     sample
-	dropBase    uint64
-	nicRaw      uint64
-	nicNow      uint64
-	nicBase     uint64
-
-	recent      errs
-	recentBase  sample
-	recentDrops uint64
-	recentNic   uint64
-}
-
-// Smoothing has to be steady against high-frequency noise yet still chase a
-// real change quickly, with bounded state. So the gain is not fixed: the
-// innovation is compared against a running estimate of the noise itself (mean
-// absolute deviation, as in TCP's rtt/rttvar), and only an innovation that
-// stands out above that noise is chased hard.
-const (
-	estAlphaCalm = 0.015
-	estAlphaSnap = 0.45
-	estMADBeta   = 0.05
-	estNoiseK    = 3.0
-)
-
-type rateEstimator struct {
-	minStep float64
-	relStep float64
-
-	est   float64
-	mad   float64
-	shown float64
-	n     int
-}
-
-// While the rate window is still growing it already averages everything there
-// is, so smoothing it again would only average in the startup ramp twice.
-func (e *rateEstimator) update(x float64, windowFull bool) {
-	if !windowFull {
-		e.est, e.shown, e.mad, e.n = x, x, 0, 0
-		return
-	}
-	if e.n == 0 {
-		e.est, e.shown, e.n = x, x, 1
-		return
-	}
-	err := x - e.est
-	abs := math.Abs(err)
-	if e.n == 1 {
-		e.mad, e.n = abs, 2
-	} else {
-		e.mad += (abs - e.mad) * estMADBeta
-	}
-
-	a := estAlphaCalm
-	if e.mad > 0 {
-		if excess := abs/(estNoiseK*e.mad) - 1; excess > 0 {
-			a = estAlphaCalm + (estAlphaSnap-estAlphaCalm)*math.Min(excess, 1)
-		}
-	}
-	e.est += err * a
-
-	// A deadband on top, so the drawn text only changes when the estimate has
-	// actually moved rather than on every frame.
-	if math.Abs(e.est-e.shown) > math.Max(e.minStep, e.relStep*math.Abs(e.est)) {
-		e.shown = e.est
-	}
-}
-
-func (e *rateEstimator) value() float64 { return e.shown }
-
-type rateEstimators struct {
-	txGbps, rxGbps rateEstimator
-	txPPS, rxPPS   rateEstimator
-}
-
-func newRateEstimators() *rateEstimators {
-	return &rateEstimators{
-		txGbps: rateEstimator{minStep: 0.02, relStep: 0.001},
-		rxGbps: rateEstimator{minStep: 0.02, relStep: 0.001},
-		txPPS:  rateEstimator{minStep: 2000, relStep: 0.002},
-		rxPPS:  rateEstimator{minStep: 2000, relStep: 0.002},
-	}
+	base        counterSet
+	nic         atomic.Uint64
+	poller      *nicPoller
 }
 
 // Monotonic totals climb by tens of thousands per frame, which is unreadable
@@ -154,9 +74,19 @@ func (h *heldValue) get(now time.Time, cur uint64) uint64 {
 	return h.v
 }
 
-type rateSample struct {
-	t                                    time.Time
-	txFrames, txBytes, rxFrames, rxBytes uint64
+// Everything the display reads, taken at one instant: the worker counters plus
+// the two that are read rather than counted, so a pair of these describes both
+// the rates and the errors over the span between them.
+type counterSet struct {
+	t     time.Time
+	s     sample
+	drops uint64
+	nic   uint64
+}
+
+func (d *direction) capture(t time.Time) counterSet {
+	d.sampleDrops()
+	return counterSet{t: t, s: d.snapshot(), drops: d.drops, nic: d.nic.Load()}
 }
 
 // Late is counted but left out of the total, since a reordered frame arrived.
@@ -180,41 +110,84 @@ func (e errs) add(o errs) errs {
 	}
 }
 
-// A sliding window: the counters are sampled every frame and the rate is taken
-// across the whole window, so the figure moves every frame while still being
-// measured over a long enough span to be steady.
+// A ring of one bucket per drawn frame, spanning rateWindowSpan. Rates come
+// from the gap between adjacent buckets and errors from the ends of the ring,
+// so both slide forward every frame instead of stepping once a second.
 type rateWindow struct {
-	samples []rateSample
+	buf     []counterSet
 	idx     int
 	filled  bool
+	scratch []float64
 }
 
 func newRateWindow(n int) *rateWindow {
-	return &rateWindow{samples: make([]rateSample, n)}
+	return &rateWindow{buf: make([]counterSet, n)}
 }
 
-func (w *rateWindow) push(s rateSample) {
-	w.samples[w.idx] = s
+func (w *rateWindow) push(c counterSet) {
+	w.buf[w.idx] = c
 	w.idx++
-	if w.idx == len(w.samples) {
+	if w.idx == len(w.buf) {
 		w.idx = 0
 		w.filled = true
 	}
 }
 
-func (w *rateWindow) span() (oldest, newest rateSample, ok bool) {
-	if !w.filled && w.idx < 2 {
-		return oldest, newest, false
-	}
-	n := w.idx - 1
-	if n < 0 {
-		n = len(w.samples) - 1
-	}
-	o := 0
+func (w *rateWindow) count() int {
 	if w.filled {
-		o = w.idx
+		return len(w.buf)
 	}
-	return w.samples[o], w.samples[n], true
+	return w.idx
+}
+
+// Indexed oldest first, so a partly filled ring reads the same as a full one.
+func (w *rateWindow) at(i int) counterSet {
+	if w.filled {
+		i += w.idx
+	}
+	return w.buf[i%len(w.buf)]
+}
+
+// The median is steady against a bursty sender yet only ever a rate some bucket
+// 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()
+	if n < 2 {
+		return 0
+	}
+	w.scratch = w.scratch[:0]
+	prev := w.at(0)
+	for i := 1; i < n; i++ {
+		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
+	}
+	slices.Sort(w.scratch)
+	return w.scratch[len(w.scratch)/2]
+}
+
+func txRatePPS(p, c counterSet, secs float64) float64 {
+	return float64(c.s.txFrames-p.s.txFrames) / secs
+}
+
+func rxRatePPS(p, c counterSet, secs float64) float64 {
+	return float64(c.s.rxFrames-p.s.rxFrames) / secs
+}
+
+func txRateGbps(p, c counterSet, secs float64) float64 {
+	return gbps(c.s.txBytes-p.s.txBytes, c.s.txFrames-p.s.txFrames, secs)
+}
+
+func rxRateGbps(p, c counterSet, secs float64) float64 {
+	return gbps(c.s.rxBytes-p.s.rxBytes, c.s.rxFrames-p.s.rxFrames, secs)
 }
 
 type sample struct {
@@ -293,12 +266,9 @@ func (d *direction) snapshot() sample {
 // everything is measured from. Rates are deliberately left running, since they
 // are instantaneous and would only blink to zero and back.
 func (d *direction) reset() {
-	d.sampleDrops()
-	d.errBase = d.snapshot()
-	d.dropBase = d.drops
+	d.base = d.capture(time.Now())
 	d.heldFrames = heldValue{}
 	d.heldSent = heldValue{}
-	d.nicBase = d.nicNow
 	d.cable.reset()
 }
 
@@ -352,27 +322,29 @@ type view struct {
 	cable           cableView
 }
 
+func errsBetween(b, n counterSet) errs {
+	return errs{
+		lost:     n.s.lost - b.s.lost,
+		late:     n.s.late - b.s.late,
+		crc:      n.s.crcErr - b.s.crcErr,
+		badMagic: n.s.badMagic - b.s.badMagic,
+		badLen:   n.s.badLen - b.s.badLen,
+		kdrop:    n.drops - b.drops,
+		// A frame the stack refused and a frame the driver dropped are the same
+		// failure seen from either side of the ring, and never the same frame
+		// twice: a send that fails never reaches the driver to be dropped.
+		link: (n.nic - b.nic) + (n.s.txErrs - b.s.txErrs) + (n.s.rxErrs - b.s.rxErrs),
+	}
+}
+
 // Cumulative fields, which need no rate window and are identical for both the
 // console and the display.
-func (d *direction) counters(now sample) view {
-	b := d.errBase
+func (d *direction) counters(now counterSet) view {
 	return view{
-		rxFrames: now.rxFrames - b.rxFrames,
-		rxGot:    now.rxBytes - b.rxBytes,
+		rxFrames: now.s.rxFrames - d.base.s.rxFrames,
+		rxGot:    now.s.rxBytes - d.base.s.rxBytes,
 		cable:    d.cable.view(),
-		since: errs{
-			lost:     now.lost - b.lost,
-			late:     now.late - b.late,
-			crc:      now.crcErr - b.crcErr,
-			badMagic: now.badMagic - b.badMagic,
-			badLen:   now.badLen - b.badLen,
-			kdrop:    d.drops - d.dropBase,
-			// A frame the stack refused and a frame the driver dropped are the same
-			// failure seen from either side of the ring, and never the same frame
-			// twice: a send that fails never reaches the driver to be dropped.
-			link: d.nicNow - d.nicBase +
-				(now.txErrs - b.txErrs) + (now.rxErrs - b.rxErrs),
-		},
+		since:    errsBetween(d.base, now),
 	}
 }
 
@@ -391,52 +363,38 @@ func totalView(views []view) view {
 	return t
 }
 
-func (d *direction) view(prev *sample, secs float64) view {
-	now := d.snapshot()
-	p := *prev
-	*prev = now
-	d.sampleDrops()
+func (d *direction) view(t time.Time) view {
+	now := d.capture(t)
+	p := d.prevConsole
+	d.prevConsole = now
 
 	v := d.counters(now)
-	txF := now.txFrames - p.txFrames
-	rxF := now.rxFrames - p.rxFrames
-	v.txPPS = float64(txF) / secs
-	v.rxPPS = float64(rxF) / secs
-	v.txGbps = gbps(now.txBytes-p.txBytes, txF, secs)
-	v.rxGbps = gbps(now.rxBytes-p.rxBytes, rxF, secs)
+	secs := now.t.Sub(p.t).Seconds()
+	if secs <= 0 {
+		return v
+	}
+	v.txPPS = txRatePPS(p, now, secs)
+	v.rxPPS = rxRatePPS(p, now, secs)
+	v.txGbps = txRateGbps(p, now, secs)
+	v.rxGbps = rxRateGbps(p, now, secs)
 	return v
 }
 
 func (d *direction) displayView(t time.Time) view {
-	now := d.snapshot()
-	d.sampleDrops()
-	d.win.push(rateSample{t, now.txFrames, now.txBytes, now.rxFrames, now.rxBytes})
+	now := d.capture(t)
+	d.win.push(now)
 
 	v := d.counters(now)
-	v.window = d.recent
 	v.rxFrames = d.heldFrames.get(t, v.rxFrames)
 	v.rxGot = d.heldSent.get(t, v.rxGot)
-
-	o, n, ok := d.win.span()
-	if !ok {
-		return v
+	if n := d.win.count(); n >= 2 {
+		v.window = errsBetween(d.win.at(0), d.win.at(n-1))
 	}
-	secs := n.t.Sub(o.t).Seconds()
-	if secs <= 0 {
-		return v
-	}
-	txF := n.txFrames - o.txFrames
-	rxF := n.rxFrames - o.rxFrames
-	full := d.win.filled
-	d.est.txPPS.update(float64(txF)/secs, full)
-	d.est.rxPPS.update(float64(rxF)/secs, full)
-	d.est.txGbps.update(gbps(n.txBytes-o.txBytes, txF, secs), full)
-	d.est.rxGbps.update(gbps(n.rxBytes-o.rxBytes, rxF, secs), full)
 
-	v.txPPS = d.est.txPPS.value()
-	v.rxPPS = d.est.rxPPS.value()
-	v.txGbps = d.est.txGbps.value()
-	v.rxGbps = d.est.rxGbps.value()
+	v.txPPS = d.win.median(txRatePPS)
+	v.rxPPS = d.win.median(rxRatePPS)
+	v.txGbps = d.win.median(txRateGbps)
+	v.rxGbps = d.win.median(rxRateGbps)
 	return v
 }
 
@@ -460,52 +418,13 @@ func (d *direction) row(elapsed time.Duration, v view, target float64, length st
 	}
 }
 
-// Sampled once a second, since these are sysfs reads. The recent errors roll
-// here rather than over the rate window because the nic counters only move at
-// this rate, and a shorter span would alias them into a flicker.
-func (d *direction) sampleNIC() {
-	d.accumulateNIC()
-
-	now := d.snapshot()
-	b := d.recentBase
-	d.recent = errs{
-		lost:     now.lost - b.lost,
-		late:     now.late - b.late,
-		crc:      now.crcErr - b.crcErr,
-		badMagic: now.badMagic - b.badMagic,
-		badLen:   now.badLen - b.badLen,
-		kdrop:    d.drops - d.recentDrops,
-		link: d.nicNow - d.recentNic +
-			(now.txErrs - b.txErrs) + (now.rxErrs - b.rxErrs),
-	}
-	d.recentBase, d.recentDrops, d.recentNic = now, d.drops, d.nicNow
-}
-
-func (d *direction) readNICTotal() uint64 {
-	tx := readNIC(d.tx.name)
-	rx := readNIC(d.rx.name)
-	return tx.tx + rx.rx + tx.carrierDown
-}
-
-// Sysfs nic counters run from boot and restart from zero whenever the driver
-// resets its statistics, so only their forward motion is accumulated. Taking
-// raw differences instead charges a boot's worth of errors to the first sample
-// and turns a reset into a near-2^64 underflow.
-func (d *direction) accumulateNIC() {
-	raw := d.readNICTotal()
-	if raw > d.nicRaw {
-		d.nicNow += raw - d.nicRaw
-	}
-	d.nicRaw = raw
-}
-
 // Whatever the interfaces counted before now is not ours, and no interval has
 // elapsed yet, so every baseline starts here and nothing is reported until the
 // first one completes.
 func (d *direction) primeCounters() {
-	d.nicRaw = d.readNICTotal()
+	d.poller.prime()
 	d.reset()
-	d.recentBase, d.recentDrops, d.recentNic = d.snapshot(), d.drops, d.nicNow
+	d.prevConsole = d.base
 }
 
 func buildDirection(label string, tx, rx endpoint, sizes []int, cfg config) (*direction, error) {
@@ -517,6 +436,7 @@ func buildDirection(label string, tx, rx endpoint, sizes []int, cfg config) (*di
 		streams: newLossWindows(cfg.streams),
 		cable:   newCableStats(),
 	}
+	d.poller = &nicPoller{txName: tx.name, rxName: rx.name, total: &d.nic}
 
 	for i := 0; i < cfg.streams; i++ {
 		et := uint16(etherBase + i)
@@ -605,6 +525,12 @@ func (d *direction) start(wg *sync.WaitGroup, doneTx, doneRx *atomic.Bool, cfg c
 		defer wg.Done()
 		receiver.run(doneRx)
 	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		d.poller.run(doneRx, startTx)
+	}()
 }
 
 func (d *direction) close() {
@@ -652,9 +578,9 @@ const (
 	// Redraw fast so the panel feels live, but measure rates over a much longer
 	// window than a frame, since a frame's worth of a bursty sender is noise.
 	displayInterval = 16 * time.Millisecond
-	// Only long enough to take the edge off one frame's sample; the estimator
-	// does the real smoothing, so this stays small and bounded.
-	rateWindowSpan = 250 * time.Millisecond
+	// Both how far back the shown errors reach and how many buckets the median
+	// runs over, so a step in the rate lands half this late.
+	rateWindowSpan = time.Second
 	totalsHold     = 50 * time.Millisecond
 )
 
@@ -796,12 +722,10 @@ func run(aName, bName, sizesArg string,
 	frame := time.NewTicker(displayInterval)
 	defer frame.Stop()
 
-	last := time.Now()
 	views := make([]view, len(dirs))
 	rows := make([]view, len(dirs))
 	for _, d := range dirs {
 		d.win = newRateWindow(int(rateWindowSpan/displayInterval) + 1)
-		d.est = newRateEstimators()
 	}
 	stats := &streamTable{cols: intervalCols, headerEvery: 20}
 	for {
@@ -829,14 +753,11 @@ func run(aName, bName, sizesArg string,
 			disp.render(totalView(views), now.Sub(start),
 				target*float64(len(dirs)), cable)
 		case now := <-tick.C:
-			secs := now.Sub(last).Seconds()
-			last = now
 			elapsed := now.Sub(start)
 			// Length needs both directions, so every row is sampled before any of
 			// them is printed.
 			for i, d := range dirs {
-				d.sampleNIC()
-				rows[i] = d.view(&d.prevConsole, secs)
+				rows[i] = d.view(now)
 			}
 			length := "-"
 			if m, ok := cfg.cableMetres(rows); ok {