theater/cabletest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Show median bucket rates, slide the error window, and poll sysfs off the draw loop

Browse Source
This commit is contained in:
flamingcow
2026-07-31 15:45:03 -07:00
parent cf204f5a23
commit c94b7ae33d
2 changed files with 186 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.go
+133 -212
View File
@@ -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 {