Redraw at 60Hz with adaptive rate smoothing, console output at 1Hz

main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"flag"
 	"fmt"
+	"math"
 	"net"
 	"os"
 	"os/signal"
@@ -45,12 +46,138 @@ type direction struct {
 	rxFDs   []int
 	reports chan string
 
-	prev     sample
-	drops    uint64
-	errBase  sample
-	dropBase uint64
-	nicTX    nicCounters
-	nicRX    nicCounters
+	prevConsole sample
+	win         *rateWindow
+	est         *rateEstimators
+	heldFrames  heldValue
+	heldSent    heldValue
+	drops       uint64
+	errBase     sample
+	dropBase    uint64
+	nicTX       nicCounters
+	nicRX       nicCounters
+}
+
+// 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
+}
+
+func (e *rateEstimator) update(x float64) {
+	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},
+	}
+}
+
+// Monotonic totals climb by tens of thousands per frame, which is unreadable
+// churn at 60Hz, so the drawn value is held and refreshed a few times a second.
+type heldValue struct {
+	v  uint64
+	at time.Time
+}
+
+func (h *heldValue) get(now time.Time, cur uint64) uint64 {
+	if now.Sub(h.at) >= totalsHold {
+		h.v, h.at = cur, now
+	}
+	return h.v
+}
+
+type rateSample struct {
+	t                                    time.Time
+	txFrames, txBytes, rxFrames, rxBytes uint64
+}
+
+// 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.
+type rateWindow struct {
+	samples []rateSample
+	idx     int
+	filled  bool
+}
+
+func newRateWindow(n int) *rateWindow {
+	return &rateWindow{samples: make([]rateSample, n)}
+}
+
+func (w *rateWindow) push(s rateSample) {
+	w.samples[w.idx] = s
+	w.idx++
+	if w.idx == len(w.samples) {
+		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
+	if w.filled {
+		o = w.idx
+	}
+	return w.samples[o], w.samples[n], true
 }
 
 type sample struct {
@@ -168,20 +295,11 @@ type view struct {
 	kdrop, errors    uint64
 }
 
-func (d *direction) view(secs float64) view {
-	now := d.snapshot()
-	p := d.prev
-	d.prev = now
-	d.sampleDrops()
-
-	txF := now.txFrames - p.txFrames
-	rxF := now.rxFrames - p.rxFrames
+// 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
 	v := view{
-		txPPS:    float64(txF) / secs,
-		rxPPS:    float64(rxF) / secs,
-		txGbps:   gbps(now.txBytes-p.txBytes, txF, secs),
-		rxGbps:   gbps(now.rxBytes-p.rxBytes, rxF, secs),
 		txFrames: now.txFrames,
 		txSent:   now.txBytes,
 		rxFrames: now.rxFrames,
@@ -196,6 +314,53 @@ func (d *direction) view(secs float64) view {
 	return v
 }
 
+func (d *direction) view(prev *sample, secs float64) view {
+	now := d.snapshot()
+	p := *prev
+	*prev = now
+	d.sampleDrops()
+
+	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)
+	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})
+
+	v := d.counters(now)
+	v.txFrames = d.heldFrames.get(t, v.txFrames)
+	v.txSent = d.heldSent.get(t, v.txSent)
+
+	o, n, ok := d.win.span()
+	if !ok {
+		return v
+	}
+	secs := n.t.Sub(o.t).Seconds()
+	if secs <= 0 {
+		return v
+	}
+	txF := n.txFrames - o.txFrames
+	rxF := n.rxFrames - o.rxFrames
+	d.est.txPPS.update(float64(txF) / secs)
+	d.est.rxPPS.update(float64(rxF) / secs)
+	d.est.txGbps.update(gbps(n.txBytes-o.txBytes, txF, secs))
+	d.est.rxGbps.update(gbps(n.rxBytes-o.rxBytes, rxF, secs))
+
+	v.txPPS = d.est.txPPS.value()
+	v.rxPPS = d.est.rxPPS.value()
+	v.txGbps = d.est.txGbps.value()
+	v.rxGbps = d.est.rxGbps.value()
+	return v
+}
+
 func (d *direction) row(elapsed time.Duration, v view, target float64) []string {
 	return []string{
 		uptime(elapsed),
@@ -328,7 +493,16 @@ func main() {
 	}
 }
 
-const reportInterval = 500 * time.Millisecond
+const (
+	reportInterval = time.Second
+	// 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
+	totalsHold     = 50 * time.Millisecond
+)
 
 func run(aName, bName, sizesArg, patArg string,
 	nStreams, batch int, duplex bool) error {
@@ -474,8 +648,15 @@ func run(aName, bName, sizesArg, patArg string,
 	close(startTx)
 	tick := time.NewTicker(reportInterval)
 	defer tick.Stop()
+	frame := time.NewTicker(displayInterval)
+	defer frame.Stop()
 
 	last := time.Now()
+	views := 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 {
 		select {
@@ -490,14 +671,18 @@ func run(aName, bName, sizesArg, patArg string,
 			}
 			stats.sinceHeader = 0
 			fmt.Println(stats.rule("error counts reset"))
+		case now := <-frame.C:
+			for i, d := range dirs {
+				views[i] = d.displayView(now)
+			}
+			disp.render(dirs, views, now.Sub(start), target)
 		case now := <-tick.C:
 			secs := now.Sub(last).Seconds()
 			last = now
 			elapsed := now.Sub(start)
-			views := make([]view, len(dirs))
-			for i, d := range dirs {
-				views[i] = d.view(secs)
-				for _, line := range stats.emit(d.row(elapsed, views[i], target)) {
+			for _, d := range dirs {
+				v := d.view(&d.prevConsole, secs)
+				for _, line := range stats.emit(d.row(elapsed, v, target)) {
 					fmt.Println(line)
 				}
 				for _, line := range d.reportNIC() {
@@ -513,7 +698,6 @@ func run(aName, bName, sizesArg, patArg string,
 					break
 				}
 			}
-			disp.render(dirs, views, elapsed, target)
 		}
 	}
 }