Sample the rate buckets on their own clock instead of the display refresh

main.go

@@ -50,14 +50,19 @@ type direction struct {
 	probeRxFD int
 	cable     *cableStats
 
+	// Guards everything the sampler touches. The counters are read on their own
+	// clock and drawn on another, and the two must not read them at once:
+	// sampleDrops consumes what it reads, so a second caller would see a gap.
+	mu          sync.Mutex
 	prevConsole counterSet
 	win         *rateWindow
-	heldFrames  heldValue
-	heldSent    heldValue
 	drops       uint64
 	base        counterSet
-	nic         atomic.Uint64
+
-	poller      *nicPoller
+	heldFrames heldValue
+	heldSent   heldValue
+	nic        atomic.Uint64
+	poller     *nicPoller
 }
 
 // Monotonic totals climb by tens of thousands per frame, which is unreadable
@@ -266,7 +271,10 @@ 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.mu.Lock()
 	d.base = d.capture(time.Now())
+	d.mu.Unlock()
+
 	d.heldFrames = heldValue{}
 	d.heldSent = heldValue{}
 	d.cable.reset()
@@ -364,6 +372,9 @@ func totalView(views []view) view {
 }
 
 func (d *direction) view(t time.Time) view {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
 	now := d.capture(t)
 	p := d.prevConsole
 	d.prevConsole = now
@@ -380,21 +391,36 @@ func (d *direction) view(t time.Time) view {
 	return v
 }
 
-func (d *direction) displayView(t time.Time) view {
+// The one place the counters are read for the ring. Runs on its own ticker, so
-	now := d.capture(t)
+// what the buckets measure does not move when the drawing does.
-	d.win.push(now)
+func (d *direction) sample(t time.Time) {
+	d.mu.Lock()
+	d.win.push(d.capture(t))
+	d.mu.Unlock()
+}
 
-	v := d.counters(now)
+// Draws whatever the sampler last put in the ring rather than reading the
-	v.rxFrames = d.heldFrames.get(t, v.rxFrames)
+// counters again, so the display is a consumer of the measurement and never a
-	v.rxGot = d.heldSent.get(t, v.rxGot)
+// participant in it.
-	if n := d.win.count(); n >= 2 {
+func (d *direction) displayView(t time.Time) view {
+	d.mu.Lock()
+	n := d.win.count()
+	if n == 0 {
+		d.mu.Unlock()
+		return view{cable: d.cable.view()}
+	}
+	v := d.counters(d.win.at(n - 1))
+	if n >= 2 {
 		v.window = errsBetween(d.win.at(0), d.win.at(n-1))
 	}
-
 	v.txPPS = d.win.median(txRatePPS)
 	v.rxPPS = d.win.median(rxRatePPS)
 	v.txGbps = d.win.median(txRateGbps)
 	v.rxGbps = d.win.median(rxRateGbps)
+	d.mu.Unlock()
+
+	v.rxFrames = d.heldFrames.get(t, v.rxFrames)
+	v.rxGot = d.heldSent.get(t, v.rxGot)
 	return v
 }
 
@@ -437,6 +463,7 @@ func buildDirection(label string, tx, rx endpoint, sizes []int, cfg config) (*di
 		cable:   newCableStats(),
 	}
 	d.poller = &nicPoller{txName: tx.name, rxName: rx.name, total: &d.nic}
+	d.win = newRateWindow(int(rateWindowSpan/sampleInterval) + 1)
 
 	for i := 0; i < cfg.streams; i++ {
 		et := uint16(etherBase + i)
@@ -578,12 +605,38 @@ 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
+	// What a bucket covers. Deliberately not displayInterval: how often the
+	// counters are read is a property of the measurement, and tying it to the
+	// refresh would let a slow or blocked draw stretch the window it reports.
+	// Short enough that a step lands promptly, long enough that a bucket holds
+	// tens of thousands of frames at line rate and is not itself noise.
+	sampleInterval = 16 * 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
 )
 
+// One sampler for both directions, so their buckets share an instant and the
+// cable length, which needs a figure from each, is never mixing two moments.
+type sampler struct {
+	dirs []*direction
+}
+
+func (s *sampler) run(done *atomic.Bool, startTx <-chan struct{}) {
+	<-startTx
+
+	tick := time.NewTicker(sampleInterval)
+	defer tick.Stop()
+
+	for !done.Load() {
+		now := <-tick.C
+		for _, d := range s.dirs {
+			d.sample(now)
+		}
+	}
+}
+
 func run(aName, bName, sizesArg string,
 	nStreams, batch int, nsPerM float64) error {
 
@@ -692,6 +745,12 @@ func run(aName, bName, sizesArg string,
 	for _, d := range dirs {
 		d.start(&wg, &doneTx, &doneRx, cfg, &rxReady, startTx)
 	}
+	samp := &sampler{dirs: dirs}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		samp.run(&doneRx, startTx)
+	}()
 	rxReady.Wait()
 
 	sig := make(chan os.Signal, 1)
@@ -724,9 +783,6 @@ func run(aName, bName, sizesArg string,
 
 	views := make([]view, len(dirs))
 	rows := make([]view, len(dirs))
-	for _, d := range dirs {
-		d.win = newRateWindow(int(rateWindowSpan/displayInterval) + 1)
-	}
 	stats := &streamTable{cols: intervalCols, headerEvery: 20}
 	for {
 		select {