From 57d7c91df106b8de9ebd13c93d1dd3b869c9b727 Mon Sep 17 00:00:00 2001 From: flamingcow Date: Fri, 31 Jul 2026 16:15:16 -0700 Subject: [PATCH] Sample the rate buckets on their own clock instead of the display refresh --- main.go | 86 +++++++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 71 insertions(+), 15 deletions(-) diff --git a/main.go b/main.go index f5d6566..4424caa 100644 --- a/main.go +++ b/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 { - now := d.capture(t) - d.win.push(now) +// The one place the counters are read for the ring. Runs on its own ticker, so +// what the buckets measure does not move when the drawing does. +func (d *direction) sample(t time.Time) { + d.mu.Lock() + d.win.push(d.capture(t)) + d.mu.Unlock() +} - v := d.counters(now) - v.rxFrames = d.heldFrames.get(t, v.rxFrames) - v.rxGot = d.heldSent.get(t, v.rxGot) - if n := d.win.count(); n >= 2 { +// Draws whatever the sampler last put in the ring rather than reading the +// counters again, so the display is a consumer of the measurement and never a +// participant in it. +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 {