diff --git a/counters.go b/counters.go index 11afbed..ce29083 100644 --- a/counters.go +++ b/counters.go @@ -5,8 +5,6 @@ import ( "path/filepath" "strconv" "strings" - "sync/atomic" - "time" ) // Split by direction of travel so each counter belongs to exactly one @@ -79,72 +77,45 @@ func sumFields(base string, fields []string) uint64 { return total } -// A poll is an ioctl and a handful of sysfs attributes, each costing -// 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 it can -// match the sampler's cadence, so a link error lands on the panel as quickly -// as a lost frame rather than stepping five times a second. -const nicInterval = sampleInterval - -type nicPoller struct { - txBase string - rx *statReader - total *atomic.Uint64 - measuring *atomic.Bool - raw uint64 +// A read is an ioctl and a handful of sysfs attributes, each costing +// microseconds of open and read, so it rides the sampler's goroutine and +// clock rather than the one that has a frame to draw every sixteen +// milliseconds; a link error lands on the panel as quickly as a lost frame. +type nicCounters struct { + txBase string + rx *statReader + raw uint64 } -func newNICPoller(fd int, txName, rxName string, total *atomic.Uint64, measuring *atomic.Bool) (*nicPoller, error) { +func newNICCounters(fd int, txName, rxName string) (*nicCounters, error) { rx, err := newStatReader(fd, rxName, nicRxFields) if err != nil { return nil, err } - return &nicPoller{ - txBase: "/sys/class/net/" + txName, - rx: rx, - total: total, - measuring: measuring, + return &nicCounters{ + txBase: "/sys/class/net/" + txName, + rx: rx, }, nil } // Carrier is a property of the netdev rather than a counter of either side, and // is taken from the transmitting interface only: across both directions each // interface transmits exactly once, so every flap is counted once. -func (p *nicPoller) read() uint64 { - return sumFields(p.txBase, nicTxFields) + p.rx.sum() + - mustReadUint(p.txBase+"/carrier_down_count") +func (c *nicCounters) read() uint64 { + return sumFields(c.txBase, nicTxFields) + c.rx.sum() + + mustReadUint(c.txBase+"/carrier_down_count") } // 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. The hardware cannot pause its -// counters for a cable measure, so its blip is suppressed here at the read: -// the baseline tracks the raw values without accumulating until it is over. -func (p *nicPoller) poll() { - raw := p.read() - if raw > p.raw && !p.measuring.Load() { - 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() +// its statistics, so only forward motion is the delta: a raw difference would +// charge a boot's worth of errors to the first collection and turn a reset +// into a near-2^64 underflow. +func (c *nicCounters) delta() uint64 { + raw := c.read() + var d uint64 + if raw > c.raw { + d = raw - c.raw } + c.raw = raw + return d } diff --git a/direction.go b/direction.go index 9b60349..434eaf8 100644 --- a/direction.go +++ b/direction.go @@ -21,17 +21,23 @@ type direction struct { rxFDs []int statFD int - // While a cable measure runs, every failure counter in this direction is - // suppressed at its source rather than counted, hidden and reverted. - measuring *atomic.Bool + // The mask gate reads this: while a cable measure runs, every fault delta + // is dropped whole, so the measure's own link blip never reaches an + // aggregate. + masked *atomic.Bool - // 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 - win *rateWindow - drops uint64 - base counterSet + // Guards everything the sampler touches. Collection consumes self-clearing + // sources, so a second concurrent reader would see a gap. + mu sync.Mutex + prev sample + nic *nicCounters + + // Since the reset: plain sums of the deltas each tick lets through. Frames + // and bytes are traffic rather than faults and skip the mask gate. + frames uint64 + bytes uint64 + since errs + window *errsRing // The smeared receive bucket the display draws its rate from, refreshed by // the sampler because the buckets are keyed by the shared read clock and @@ -50,24 +56,6 @@ type direction struct { smBytes [smearWindow]uint64 smLen int smNext int64 - - nic atomic.Uint64 - poller *nicPoller -} - -// Everything the display reads, taken at one instant, 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() counterSet { - d.sampleDrops() - s := d.snapshot() - return counterSet{t: time.Now(), s: s, drops: d.drops, nic: d.nic.Load()} } // What someone testing a cable is asking, rather than how each failure happened @@ -86,41 +74,34 @@ func (e errs) add(o errs) errs { } } -// 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 { - buf []counterSet - idx int - filled bool -} - -func newRateWindow(n int) *rateWindow { - return &rateWindow{buf: make([]counterSet, n)} -} - -func (w *rateWindow) push(c counterSet) { - w.buf[w.idx] = c - w.idx++ - if w.idx == len(w.buf) { - w.idx = 0 - w.filled = true +func (e errs) sub(o errs) errs { + return errs{ + lost: e.lost - o.lost, corrupt: e.corrupt - o.corrupt, + link: e.link - o.link, internal: e.internal - o.internal, } } -func (w *rateWindow) count() int { - if w.filled { - return len(w.buf) - } - return w.idx +// The rolling short-term window: one slot of accepted fault deltas per sampler +// tick, spanning rateWindowSpan, with the sum kept current as slots enter and +// leave. A masked tick pushes zero, so the window slides on the clock +// regardless. +type errsRing struct { + slots []errs + idx int + sum errs } -// 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 +func newErrsRing(n int) *errsRing { + return &errsRing{slots: make([]errs, n)} +} + +func (r *errsRing) push(e errs) { + r.sum = r.sum.sub(r.slots[r.idx]).add(e) + r.slots[r.idx] = e + r.idx++ + if r.idx == len(r.slots) { + r.idx = 0 } - return w.buf[i%len(w.buf)] } // How long the frontier may sit still before the wire is taken to have gone @@ -207,28 +188,16 @@ func (d *direction) snapshot() sample { return s } -// Counters keep climbing in the workers, so resetting just moves the origin -// everything is measured from. Rates and the rolling error window are about now -// rather than since the reset, so they keep running; the origin goes into the -// ring so the newest bucket never sits behind it. +// Counters keep climbing in the workers; a reset zeroes the sums accumulated +// from them. Rates and the rolling window are about now rather than since the +// reset, so they keep running. func (d *direction) reset() { d.mu.Lock() - d.base = d.capture() - d.win.push(d.base) + d.frames, d.bytes = 0, 0 + d.since = errs{} d.mu.Unlock() } -// The socket statistic is read-and-clear, so it is always consumed; a drop -// during a cable measure is the blip's and is discarded at this source. -func (d *direction) sampleDrops() { - for _, fd := range d.rxFDs { - n := packetDrops(fd) - if !d.measuring.Load() { - d.drops += n - } - } -} - func gbps(bytes, frames uint64, secs float64) float64 { return float64((bytes+frames*wireOverhead)*8) / secs / 1e9 } @@ -243,31 +212,51 @@ type view struct { window errs } -func errsBetween(b, n counterSet) errs { +// One tick of every counter this direction owns, flattened to deltas: the +// worker counters grow forever, the socket drop statistic clears itself on +// read, and the nic counters re-base, none of which survives past here. +func (d *direction) collect() (frames, bytes uint64, e errs) { + cur := d.snapshot() + prev := d.prev + d.prev = cur + var drops uint64 + for _, fd := range d.rxFDs { + drops += packetDrops(fd) + } + return cur.rxFrames - prev.rxFrames, cur.rxBytes - prev.rxBytes, + classify(cur, prev, drops, d.nic.delta()) +} + +// Which delta feeds which bucket is the whole taxonomy the panel reports. +func classify(cur, prev sample, drops, nic uint64) errs { return errs{ - lost: n.s.lost - b.s.lost, + lost: cur.lost - prev.lost, // Four ways of noticing one thing: a payload that does not match its // checksum, a header that does not match its own, a header that is not // ours, and a length that cannot be. - corrupt: (n.s.crcErr - b.s.crcErr) + (n.s.badHdr - b.s.badHdr) + - (n.s.badMagic - b.s.badMagic) + (n.s.badLen - b.s.badLen), + corrupt: (cur.crcErr - prev.crcErr) + (cur.badHdr - prev.badHdr) + + (cur.badMagic - prev.badMagic) + (cur.badLen - prev.badLen), // What the hardware reported. Nothing the host declined to send is here, // so this one going red means the cable. - link: (n.nic - b.nic) + (n.s.rxErrs - b.s.rxErrs), + link: nic + (cur.rxErrs - prev.rxErrs), // Ours rather than the cable's. A late frame is unreachable while each // stream has a flow rule to its own queue, which is exactly why it is // worth counting. - internal: (n.drops - b.drops) + (n.s.late - b.s.late) + - (n.s.txErrs - b.s.txErrs), + internal: drops + (cur.late - prev.late) + (cur.txErrs - prev.txErrs), } } -func (d *direction) counters(now counterSet) view { - return view{ - rxFrames: now.s.rxFrames - d.base.s.rxFrames, - rxBytes: now.s.rxBytes - d.base.s.rxBytes, - since: errsBetween(d.base, now), +// The mask gate: a masked tick is a cable measure's own blip, and its faults +// stop dead here — only the clock reaches the aggregates. +func (d *direction) aggregate(frames, bytes uint64, e errs) errs { + d.frames += frames + d.bytes += bytes + if d.masked.Load() { + e = errs{} } + d.since = d.since.add(e) + d.window.push(e) + return e } func totalView(views []view) view { @@ -283,41 +272,43 @@ func totalView(views []view) view { return t } -func (d *direction) sample() { +// Returns the accepted deltas so the sampler can bucket them onward. The +// clock is read after the counters, per docs/measurement.md: stamps come from +// read time, never from the ticker. +func (d *direction) sample() errs { d.mu.Lock() - d.win.push(d.capture()) + defer d.mu.Unlock() + frames, bytes, e := d.collect() + e = d.aggregate(frames, bytes, e) d.readRateBucket(time.Now()) - d.mu.Unlock() + return e } -// Draws what the sampler last put in the ring rather than reading the counters +// Draws what the sampler last aggregated rather than reading the counters // again, so the display never participates in the measurement. func (d *direction) displayView() view { d.mu.Lock() - n := d.win.count() - if n == 0 { - d.mu.Unlock() - return view{} + v := view{ + rxFrames: d.frames, + rxBytes: d.bytes, + since: d.since, + window: d.window.sum, + rxPPS: float64(d.rateFrames) / rateBucketSecs, + rxGbps: gbps(d.rateBytes, d.rateFrames, rateBucketSecs), } - v := d.counters(d.win.at(n - 1)) - if n >= 2 { - v.window = errsBetween(d.win.at(0), d.win.at(n-1)) - } - v.rxPPS = float64(d.rateFrames) / rateBucketSecs - v.rxGbps = gbps(d.rateBytes, d.rateFrames, rateBucketSecs) d.mu.Unlock() return v } -// 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. +// Whatever was counted before now is not ours: the first collection primes +// every baseline and is discarded. func (d *direction) primeCounters() { - d.poller.prime() - d.reset() + d.mu.Lock() + d.collect() + d.mu.Unlock() } -func buildDirection(label string, tx, rx endpoint, measuring *atomic.Bool) (*direction, error) { +func buildDirection(label string, tx, rx endpoint, masked *atomic.Bool) (*direction, error) { // Built before the windows, since each window judges sequence numbers against // the frontier its own sender publishes. txs := make([]*txStats, numStreams) @@ -325,23 +316,23 @@ func buildDirection(label string, tx, rx endpoint, measuring *atomic.Bool) (*dir txs[i] = &txStats{} } d := &direction{ - txStats: txs, - streams: newLossWindows(txs), - measuring: measuring, + txStats: txs, + streams: newLossWindows(txs), + masked: masked, } - // Held open for the life of the run: the stats ioctl is issued five times a - // second and reopening a socket for each one is pure overhead. + // Held open for the life of the run: the stats ioctl rides every sampler + // tick and reopening a socket for each one is pure overhead. statFD, err := unix.Socket(unix.AF_INET, unix.SOCK_DGRAM, 0) if err != nil { return nil, fmt.Errorf("%s stats socket: %w", label, err) } d.statFD = statFD - d.poller, err = newNICPoller(statFD, tx.name, rx.name, &d.nic, measuring) + d.nic, err = newNICCounters(statFD, tx.name, rx.name) if err != nil { return nil, fmt.Errorf("%s: %w", label, err) } - d.win = newRateWindow(int(rateWindowSpan/sampleInterval) + 1) + d.window = newErrsRing(int(rateWindowSpan/sampleInterval) + 1) for i := 0; i < numStreams; i++ { et := uint16(etherBase + i) @@ -367,13 +358,12 @@ func buildDirection(label string, tx, rx endpoint, measuring *atomic.Bool) (*dir func (d *direction) start(wg *sync.WaitGroup, done *atomic.Bool, rxReady *sync.WaitGroup, startTx <-chan struct{}) { for i, fd := range d.txFDs { w := &txWorker{ - fd: fd, - stream: uint16(i), - spec: d.specs[i], - batch: batchSize, - stats: d.txStats[i], - measuring: d.measuring, - startTx: startTx, + fd: fd, + stream: uint16(i), + spec: d.specs[i], + batch: batchSize, + stats: d.txStats[i], + startTx: startTx, } wg.Add(1) go func() { @@ -384,14 +374,14 @@ func (d *direction) start(wg *sync.WaitGroup, done *atomic.Bool, rxReady *sync.W } for i, fd := range d.rxFDs { w := &rxWorker{ - fd: fd, - batch: batchSize, - stream: uint16(i), - spec: d.specs[i], - stats: d.rxStats[i], - loss: &d.streams[i], - measuring: d.measuring, - ready: rxReady, + fd: fd, + batch: batchSize, + stream: uint16(i), + spec: d.specs[i], + stats: d.rxStats[i], + loss: &d.streams[i], + masked: d.masked, + ready: rxReady, } wg.Add(1) go func() { @@ -400,13 +390,6 @@ func (d *direction) start(wg *sync.WaitGroup, done *atomic.Bool, rxReady *sync.W w.run(done) }() } - - wg.Add(1) - go func() { - defer wg.Done() - defer holdPanic() - d.poller.run(done, startTx) - }() } func (d *direction) close() { diff --git a/docs/measurement.md b/docs/measurement.md index b229c43..c594871 100644 --- a/docs/measurement.md +++ b/docs/measurement.md @@ -4,7 +4,7 @@ Hard-won rules about measuring correctly. Violating these produces numbers that ## Rate buckets: stamp at read time, never at ticker-fire time -- `direction.capture()` stamps each bucket with `time.Now()` taken immediately after reading the counters — never a timestamp passed in from the sampler ticker. +- `direction.sample()` reads the clock immediately after reading the counters — never a timestamp passed in from the sampler ticker. - Why: a rate is Δcounters/Δt. The sampler runs a variable delay after its tick, so a tick-time label pairs a right numerator with a wrong denominator — and because E[1/x] > 1/E[x], the error biases the rate **upward**; it doesn't cancel. - Measured: tick-time stamping read 27.55 G / 4.88 Mpps against a true 19.86 G — 35% high. Read-time stamping read within 2%. - A 1 s console interval hides it (jitter < 0.2%); a 16 ms panel window exposes it. @@ -22,7 +22,7 @@ This rule covers the **NIC-counter** buckets and the per-stream receive buckets - `/sys/class/net/*/statistics/*` run from boot, not process start, and reset to zero on driver stats resets. - In unsigned arithmetic that bites twice: a zero baseline charges the machine's whole lifetime to the run; a backward step underflows `now - base` to ~2^64. -- Rule: accumulate only forward motion into a process-local total, then take every baseline from that total — never from a raw reading. Establish all baselines in one place before traffic starts. +- Rule: only forward motion counts — the delta layer clamps a backward step to zero and re-bases, never handing a raw difference upward. Establish all baselines in one place before traffic starts. - Debugging tell: when a panel's background disagrees with rows drawn from the same value, suspect two vintages of one counter before a color bug. ## Only zero-baseline-loss runs count; small frames measure the host diff --git a/history.go b/history.go index 01aff08..2f82610 100644 --- a/history.go +++ b/history.go @@ -5,11 +5,12 @@ import ( "time" ) -// The panel's recent-history ring: 18 slots of 5 seconds, sampled every drawn -// frame from the same figures the panel paints, so a slot has data within one -// frame of being born. Slots are wall-time and survive a reset — recently is -// recently, whatever the counters were re-based to — while the since-reset -// aggregates and fault ages clear. +// The panel's recent-history ring: 18 slots of 5 seconds. Fault and corrected +// deltas arrive straight from the layers below the moment they clear the mask, +// and the per-frame judgments — measure state, rate, SNR — from the draw loop, +// so a slot has data within one tick of being born. Slots are wall-time and +// survive a reset — recently is recently, whatever the aggregates were +// re-based to — while the fault ages and since-reset extremes clear. const ( histSlots = 18 histSlotSpan = 5 * time.Second @@ -47,10 +48,6 @@ type history struct { start time.Time slots [histSlots]histSlot - prevValid bool - prevSince errs - prevCorrected uint64 - lastFault [4]time.Time sinceRateLow bool sinceHaveSNR bool @@ -69,15 +66,27 @@ func (h *history) slot(seq int64) *histSlot { return s } -// Counters only move forward between samples except across a reset, where -// they re-base to zero; a backward step is that re-base, not negative faults. -func delta(now, prev uint64) uint64 { - if now < prev { - return 0 +// Bucketing sits above the mask: only fault deltas that survived it arrive +// here, already flattened, so a slot is a plain sum. +func (h *history) addFaults(now time.Time, e errs) { + h.mu.Lock() + defer h.mu.Unlock() + s := h.slot(int64(now.Sub(h.start) / histSlotSpan)) + for i, c := range faultClasses { + if d := c.get(e); d > 0 { + s.faults[i] += d + h.lastFault[i] = now + } } - return now - prev } +func (h *history) addCorrected(now time.Time, d uint64) { + h.mu.Lock() + h.slot(int64(now.Sub(h.start)/histSlotSpan)).corrected += d + h.mu.Unlock() +} + +// The per-frame judgments, from the same figures the panel paints. func (h *history) sample(now time.Time, v view, phy phyDisplay, measuring bool, target float64) { h.mu.Lock() @@ -85,34 +94,20 @@ func (h *history) sample(now time.Time, v view, phy phyDisplay, s := h.slot(int64(now.Sub(h.start) / histSlotSpan)) - if h.prevValid { - for i, c := range faultClasses { - d := delta(c.get(v.since), c.get(h.prevSince)) - s.faults[i] += d - if d > 0 { - h.lastFault[i] = now - } - } - s.corrected += delta(phy.corrected, h.prevCorrected) - } - h.prevValid = true - h.prevSince = v.since - h.prevCorrected = phy.corrected - if measuring { s.measuring = true - } else { - if v.rxGbps < rateOKFrac*target { - s.rateLow = true - h.sinceRateLow = true + return + } + if v.rxGbps < rateOKFrac*target { + s.rateLow = true + h.sinceRateLow = true + } + if phy.haveSNR { + if !s.haveSNR || phy.worstMargin < s.snrMin { + s.haveSNR, s.snrMin = true, phy.worstMargin } - if phy.haveSNR { - if !s.haveSNR || phy.worstMargin < s.snrMin { - s.haveSNR, s.snrMin = true, phy.worstMargin - } - if !h.sinceHaveSNR || phy.worstMargin < h.sinceSNRMin { - h.sinceHaveSNR, h.sinceSNRMin = true, phy.worstMargin - } + if !h.sinceHaveSNR || phy.worstMargin < h.sinceSNRMin { + h.sinceHaveSNR, h.sinceSNRMin = true, phy.worstMargin } } } @@ -121,7 +116,6 @@ func (h *history) sample(now time.Time, v view, phy phyDisplay, // the reset origin clears. func (h *history) reset() { h.mu.Lock() - h.prevValid = false h.lastFault = [4]time.Time{} h.sinceRateLow = false h.sinceHaveSNR = false diff --git a/loss.go b/loss.go index f675fb4..967b211 100644 --- a/loss.go +++ b/loss.go @@ -12,13 +12,13 @@ const ( ) type lossWindow struct { - mu sync.Mutex - base uint64 - inited bool - quietEra bool - bits []uint64 - lost atomic.Uint64 - late atomic.Uint64 + mu sync.Mutex + base uint64 + inited bool + maskedEra bool + bits []uint64 + lost atomic.Uint64 + late atomic.Uint64 // The sending half of this stream. Both halves are in this process and the // receiving socket ignores its own outgoing frames, so this window is fed by @@ -48,24 +48,25 @@ func newLossWindows(tx []*txStats) []lossWindow { // the run. Nothing that arrives later is evidence enough to undo that, which is // why the sender's own frontier is the guard rather than a limit on how far a // single step may move. -// quiet slides the window without charging: gaps written off during a cable -// measure are its own link blip, suppressed at the source while the state -// machine stays in sync with the wire. -func (w *lossWindow) observe(seq uint64, quiet bool) bool { +// masked slides the window without charging. A gap's charge lands at eviction, +// which can fall after the mask lifts, so this source judges each charge +// against the mask itself rather than leaving a deferred judgment to the +// sampler's gate. +func (w *lossWindow) observe(seq uint64, masked bool) bool { if seq >= w.sent.Load() { return false } w.mu.Lock() - if w.quietEra && !quiet { + if w.maskedEra && !masked { // Leaving a measure: the window was full when it began, so every hole - // still inside went missing while failures were suppressed. Marking - // them delivered keeps the write-off complete however the eviction - // timing falls, instead of charging the gap's in-window tail later. + // still inside went missing while the mask was up. Marking them + // delivered keeps the write-off complete however the eviction timing + // falls, instead of charging the gap's in-window tail later. for i := range w.bits { w.bits[i] = ^uint64(0) } } - w.quietEra = quiet + w.maskedEra = masked if !w.inited { // Start half a window below the first sequence seen, so anything the // sender put on the wire before it lands inside the window rather than @@ -77,13 +78,13 @@ func (w *lossWindow) observe(seq uint64, quiet bool) bool { } if seq < w.base { w.mu.Unlock() - if !quiet { + if !masked { w.late.Add(1) } return true } if seq >= w.base+lossSlots { - w.evict(seq-lossSlots+1, quiet) + w.evict(seq-lossSlots+1, masked) } idx := seq & (lossSlots - 1) w.bits[idx>>6] |= 1 << (idx & 63) @@ -91,7 +92,7 @@ func (w *lossWindow) observe(seq uint64, quiet bool) bool { return true } -func (w *lossWindow) evict(newBase uint64, quiet bool) { +func (w *lossWindow) evict(newBase uint64, masked bool) { span := newBase - w.base if span >= lossSlots { var missing uint64 @@ -99,7 +100,7 @@ func (w *lossWindow) evict(newBase uint64, quiet bool) { missing += uint64(64 - bits.OnesCount64(w.bits[i])) w.bits[i] = 0 } - if !quiet { + if !masked { w.lost.Add(missing + span - lossSlots) } w.base = newBase @@ -115,7 +116,7 @@ func (w *lossWindow) evict(newBase uint64, quiet bool) { } w.bits[word] &^= bit } - if !quiet { + if !masked { w.lost.Add(missing) } w.base = newBase diff --git a/loss_test.go b/loss_test.go index 12291f0..3355aad 100644 --- a/loss_test.go +++ b/loss_test.go @@ -131,11 +131,11 @@ func TestLossWindowRefusesUnsentSeq(t *testing.T) { } } -// A gap written off while quiet — a cable measure's own link blip — charges +// A gap written off while masked — a cable measure's own link blip — charges // nothing, however the eviction timing falls: even the gap's tail still // inside the window when the measure ends is presumed delivered. Counting // resumes seamlessly and later real gaps are still caught. -func TestLossWindowQuietSlidesWithoutCharging(t *testing.T) { +func TestLossWindowMaskedSlidesWithoutCharging(t *testing.T) { w := newWindow(4 * lossSlots) for seq := uint64(0); seq < 100; seq++ { w.observe(seq, false) @@ -143,7 +143,7 @@ func TestLossWindowQuietSlidesWithoutCharging(t *testing.T) { // The blip: a huge jump arriving during the measure. w.observe(2*lossSlots, true) if got := w.lost.Load(); got != 0 { - t.Errorf("lost = %d after a quiet write-off, want 0", got) + t.Errorf("lost = %d after a masked write-off, want 0", got) } // The measure ends immediately — the worst case, with the gap's tail diff --git a/main.go b/main.go index f6938df..b32c3cd 100644 --- a/main.go +++ b/main.go @@ -152,9 +152,11 @@ const ( ) // One sampler for both directions, so they are read back to back on one clock -// rather than drifting apart on two. +// rather than drifting apart on two. Whatever each tick lets through the mask +// goes on into the history's buckets. type sampler struct { dirs []*direction + hist *history } func (s *sampler) run(done *atomic.Bool, startTx <-chan struct{}) { @@ -166,7 +168,7 @@ func (s *sampler) run(done *atomic.Bool, startTx <-chan struct{}) { for !done.Load() { <-tick.C for _, d := range s.dirs { - d.sample() + s.hist.addFaults(time.Now(), d.sample()) } } } @@ -241,7 +243,7 @@ func run(aName, bName string) (err error) { var dirs []*direction for _, p := range [][2]endpoint{{a, b}, {b, a}} { - d, err := buildDirection(p[0].name+"->"+p[1].name, p[0], p[1], &diag.measuring) + d, err := buildDirection(p[0].name+"->"+p[1].name, p[0], p[1], &diag.masked) if err != nil { return err } @@ -277,7 +279,8 @@ func run(aName, bName string) (err error) { for _, d := range dirs { d.start(&wg, &done, &rxReady, startTx) } - samp := &sampler{dirs: dirs} + hist := newHistory(time.Now()) + samp := &sampler{dirs: dirs, hist: hist} wg.Add(1) go func() { defer wg.Done() @@ -289,7 +292,7 @@ func run(aName, bName string) (err error) { go func() { defer wg.Done() defer holdPanic() - m.run(&done) + m.run(&done, hist) }() } // Every return from here on stops the workers before the deferred closes @@ -340,10 +343,9 @@ func run(aName, bName string) (err error) { } start := time.Now() - hist := newHistory(start) // A reset resets at the press, then re-measures; the measure's failures - // are suppressed at their sources while it runs, so there is nothing to - // hide or revert afterwards. + // are dropped at the mask gate while it runs, so there is nothing to hide + // or revert afterwards. kickMeasure := func() { if diag.kick(&done) { diag.resetLengths() diff --git a/main_test.go b/main_test.go index e8d139f..3777b22 100644 --- a/main_test.go +++ b/main_test.go @@ -1,6 +1,7 @@ package main import ( + "sync/atomic" "testing" "time" ) @@ -222,70 +223,76 @@ func TestSmearConservesFramesAndBytes(t *testing.T) { } } -// Indexed oldest first, so the error window still spans the whole ring once it -// has wrapped. -func TestRateWindowIndexesOldestFirst(t *testing.T) { - w := newRateWindow(3) +// The window is a rolling sum of one span of deltas: what leaves the ring +// leaves the sum, so after a wrap only the last len slots are counted. +func TestErrsRingRollingSum(t *testing.T) { + r := newErrsRing(3) for i := 1; i <= 5; i++ { - w.push(counterSet{s: sample{rxFrames: uint64(i)}}) + r.push(errs{lost: uint64(i), link: 1}) } - if got := w.count(); got != 3 { - t.Fatalf("count = %d, want 3", got) - } - if got := w.at(0).s.rxFrames; got != 3 { - t.Errorf("oldest = %d, want 3", got) - } - if got := w.at(2).s.rxFrames; got != 5 { - t.Errorf("newest = %d, want 5", got) + want := errs{lost: 3 + 4 + 5, link: 3} + if r.sum != want { + t.Errorf("sum = %+v, want %+v", r.sum, want) } } -// Which counter feeds which bucket is the whole taxonomy the panel reports, so -// it is pinned here rather than left to whoever reads errsBetween next. -func TestErrsBetweenBuckets(t *testing.T) { - n := counterSet{ - s: sample{ - lost: 1, late: 7, - crcErr: 2, badMagic: 3, badLen: 4, badHdr: 10, - txErrs: 6, rxErrs: 5, - }, - drops: 9, - nic: 8, +// Which delta feeds which bucket is the whole taxonomy the panel reports, so +// it is pinned here rather than left to whoever reads classify next. +func TestClassifyBuckets(t *testing.T) { + cur := sample{ + lost: 1, late: 7, + crcErr: 2, badMagic: 3, badLen: 4, badHdr: 10, + txErrs: 6, rxErrs: 5, } - got := errsBetween(counterSet{}, n) + got := classify(cur, sample{}, 9, 8) want := errs{lost: 1, corrupt: 2 + 3 + 4 + 10, link: 8 + 5, internal: 9 + 7 + 6} if got != want { - t.Errorf("errsBetween = %+v, want %+v", got, want) + t.Errorf("classify = %+v, want %+v", got, want) } } -// A reset re-bases from a fresh capture while the ring still holds buckets from -// just before it, so the newest bucket must not be left behind the new origin. -func TestResetDoesNotUnderflowTotals(t *testing.T) { - d := &direction{ - win: newRateWindow(8), - rxStats: []*rxStats{{}}, +// A masked tick's faults stop at the gate while traffic and the window clock +// keep moving; the tick after the mask lifts aggregates again. +func TestMaskedTickAggregatesNothing(t *testing.T) { + var masked atomic.Bool + d := &direction{masked: &masked, window: newErrsRing(8)} + + masked.Store(true) + if e := d.aggregate(10, 640, errs{lost: 5, link: 2}); e != (errs{}) { + t.Errorf("accepted %+v from a masked tick, want nothing", e) + } + v := d.displayView() + if v.since != (errs{}) || v.window != (errs{}) { + t.Errorf("since = %+v window = %+v after a masked tick, want both empty", v.since, v.window) + } + if v.rxFrames != 10 || v.rxBytes != 640 { + t.Errorf("traffic = %d frames %d bytes, want it counted through the mask", v.rxFrames, v.rxBytes) } - d.rxStats[0].frames.Store(100) - d.rxStats[0].bytes.Store(6400) - d.sample() + masked.Store(false) + d.aggregate(0, 0, errs{lost: 1}) + v = d.displayView() + if v.since.lost != 1 || v.window.lost != 1 { + t.Errorf("since.lost = %d window.lost = %d after the mask lifted, want 1 and 1", v.since.lost, v.window.lost) + } +} - d.rxStats[0].frames.Store(150) - d.rxStats[0].bytes.Store(9600) +// A reset zeroes the since-reset sums; the rolling window is about now rather +// than since the reset, so it keeps what it held. +func TestResetZeroesSinceAndKeepsWindow(t *testing.T) { + var masked atomic.Bool + d := &direction{masked: &masked, window: newErrsRing(8)} + d.aggregate(100, 6400, errs{lost: 3}) d.reset() v := d.displayView() - if v.rxFrames != 0 { - t.Errorf("rxFrames = %d, want 0", v.rxFrames) + if v.rxFrames != 0 || v.rxBytes != 0 { + t.Errorf("traffic = %d frames %d bytes after reset, want 0 and 0", v.rxFrames, v.rxBytes) } - if v.rxBytes != 0 { - t.Errorf("rxBytes = %d, want 0", v.rxBytes) + if v.since != (errs{}) { + t.Errorf("since = %+v after reset, want empty", v.since) } - - // The rolling window and the rates are about now rather than since the - // reset, so the buckets from before it have to survive. - if got := d.win.count(); got < 2 { - t.Errorf("ring holds %d buckets after reset, want the pre-reset history kept", got) + if v.window.lost != 3 { + t.Errorf("window.lost = %d after reset, want the pre-reset window kept", v.window.lost) } } diff --git a/phy.go b/phy.go index c6353e9..fae516d 100644 --- a/phy.go +++ b/phy.go @@ -112,9 +112,11 @@ const ( snrWarnMargin = 1.0 ) +// While masked a measure owns the module: every register it would poll is the +// measure's own disturbance, so nothing is read at all. type phyModule struct { - dev phyDev - busy atomic.Bool + dev phyDev + masked atomic.Bool mu sync.Mutex sampled bool @@ -131,9 +133,8 @@ type phyModule struct { notes []string } -// Silent while a measure owns the module. -func (m *phyModule) poll() error { - if m.busy.Load() { +func (m *phyModule) poll(hist *history) error { + if m.masked.Load() { return nil } link, linkRaw, err := devLinkUp(m.dev) @@ -156,11 +157,11 @@ func (m *phyModule) poll() error { return err } - // Setting busy is a measure's first act, so busy still clear here proves - // every read above finished before the measure disturbed anything; a - // measure that began mid-read makes the samples ambiguous, and forgive() + // Setting masked is a measure's first act, so masked still clear here + // proves every read above finished before the measure disturbed anything; + // a measure that began mid-read makes the samples ambiguous, and forgive() // re-baselines after it anyway. - if m.busy.Load() { + if m.masked.Load() { return nil } @@ -188,6 +189,7 @@ func (m *phyModule) poll() error { m.ber += ber m.retrains += rt m.recentDelta = delta + hist.addCorrected(time.Now(), delta) } else { m.recentDelta = 0 m.primed = true @@ -205,7 +207,7 @@ func (m *phyModule) takeNotes() []string { return n } -func (m *phyModule) run(done *atomic.Bool) { +func (m *phyModule) run(done *atomic.Bool, hist *history) { tick := time.NewTicker(phyInterval) defer tick.Stop() @@ -213,7 +215,7 @@ func (m *phyModule) run(done *atomic.Bool) { var lastErr error for !done.Load() { <-tick.C - if err := m.poll(); err != nil { + if err := m.poll(hist); err != nil { dark++ lastErr = err if dark >= phyMaxDark { @@ -437,12 +439,12 @@ func bcmEnd(mods []*phyModule) *bcm { // relink, so the MDI resolution is the fresh one. func measureCable(mods []*phyModule, done *atomic.Bool) (cableInfo, error) { for _, m := range mods { - m.busy.Store(true) + m.masked.Store(true) } defer func() { for _, m := range mods { m.forgive() - m.busy.Store(false) + m.masked.Store(false) } }() @@ -486,10 +488,10 @@ func measureCable(mods []*phyModule, done *atomic.Bool) (cableInfo, error) { type cableDiag struct { mods []*phyModule - // While set, every failure counter is suppressed at its source: the - // measure's own link blip is never counted anywhere, rather than counted, - // hidden and reverted. - measuring atomic.Bool + // The mask itself. While set, every fault delta is dropped at the mask + // gate before it can reach an aggregate: the measure's own link blip is + // never counted anywhere, rather than counted, hidden and reverted. + masked atomic.Bool mu sync.Mutex info cableInfo @@ -503,7 +505,7 @@ func newCableDiag(mods []*phyModule, info cableInfo) *cableDiag { func (c *cableDiag) snapshot() (cableInfo, cableLengths, bool) { c.mu.Lock() defer c.mu.Unlock() - return c.info, c.lengths, c.measuring.Load() + return c.info, c.lengths, c.masked.Load() } func (c *cableDiag) resetLengths() { @@ -516,7 +518,7 @@ func (c *cableDiag) resetLengths() { // tester that cannot run its diagnostics must not keep testing: the failure is // fatal, and the reboot's driver reload is also the wedged-µC recovery. func (c *cableDiag) kick(done *atomic.Bool) bool { - if !c.measuring.CompareAndSwap(false, true) { + if !c.masked.CompareAndSwap(false, true) { return false } go func() { @@ -533,7 +535,7 @@ func (c *cableDiag) kick(done *atomic.Bool) bool { } } c.mu.Unlock() - c.measuring.Store(false) + c.masked.Store(false) }() return true } @@ -575,9 +577,9 @@ func openModules(names [2]string) ([]*phyModule, [2]string, error) { time.Sleep(moduleUpPoll) } m := &phyModule{dev: dev} - // Born busy: the pollers stay silent through bringup's SETs and + // Born masked: the pollers stay silent through bringup's SETs and // retrains until the first measure completes and lifts the gate. - m.busy.Store(true) + m.masked.Store(true) mods = append(mods, m) } return mods, idents, nil diff --git a/rx.go b/rx.go index 5d11a78..dc26c45 100644 --- a/rx.go +++ b/rx.go @@ -110,14 +110,14 @@ func fillBack(frames, bytes []uint64) { } type rxWorker struct { - fd int - batch int - stream uint16 - spec *frameSpec - stats *rxStats - loss *lossWindow - measuring *atomic.Bool - ready *sync.WaitGroup + fd int + batch int + stream uint16 + spec *frameSpec + stats *rxStats + loss *lossWindow + masked *atomic.Bool + ready *sync.WaitGroup } func (w *rxWorker) run(done *atomic.Bool) { @@ -133,12 +133,13 @@ func (w *rxWorker) run(done *atomic.Bool) { w.ready.Done() for !done.Load() { - // One flag load covers the batch: failures seen while a cable measure - // runs are its own link blip and are never counted. - quiet := w.measuring.Load() + // The worker just counts; the mask gate above decides what a masked + // tick's counts are worth. One flag load covers the batch for the loss + // window, whose gap charges outlive the tick they happened in. + masked := w.masked.Load() n, err := recvmmsg(w.fd, hdrs, unix.MSG_WAITFORONE) if n <= 0 { - if err != nil && err != unix.EAGAIN && err != unix.EINTR && !quiet { + if err != nil && err != unix.EAGAIN && err != unix.EINTR { w.stats.rxErrs.Add(1) } continue @@ -149,9 +150,6 @@ func (w *rxWorker) run(done *atomic.Bool) { buf := bufs[i][:int(hdrs[i].len)] p, st := parseHeader(buf) if st != hdrOK { - if quiet { - continue - } if st == hdrForeign { w.stats.badMagic.Add(1) } else { @@ -165,22 +163,18 @@ func (w *rxWorker) run(done *atomic.Bool) { // The ethertype this socket is bound to already says which stream the // frame belongs to, so a header naming another one is damaged, as is a // sequence number the sender never reached. - if p.stream != w.stream || !w.loss.observe(p.seq, quiet) { - if !quiet { - w.stats.badHdr.Add(1) - } + if p.stream != w.stream || !w.loss.observe(p.seq, masked) { + w.stats.badHdr.Add(1) continue } want, ok := w.spec.expectedCRC(p.patIdx, p.payLen) if !ok { - if !quiet { - w.stats.badLen.Add(1) - } + w.stats.badLen.Add(1) continue } pay := buf[minFrame : minFrame+p.payLen] - if crc32.Checksum(pay, crcTable) != want && !quiet { + if crc32.Checksum(pay, crcTable) != want { w.stats.crcErr.Add(1) } } diff --git a/tx.go b/tx.go index a9670bc..59adc8f 100644 --- a/tx.go +++ b/tx.go @@ -17,13 +17,12 @@ type txStats struct { } type txWorker struct { - fd int - stream uint16 - spec *frameSpec - batch int - stats *txStats - measuring *atomic.Bool - startTx <-chan struct{} + fd int + stream uint16 + spec *frameSpec + batch int + stats *txStats + startTx <-chan struct{} } func (w *txWorker) run(done *atomic.Bool) { @@ -64,8 +63,7 @@ func (w *txWorker) run(done *atomic.Bool) { } // Taking fewer of the vector than offered is the ring's room, not a frame // lost: the rest go on the next pass. - if n < 0 && err != unix.EINTR && err != unix.EAGAIN && err != unix.ENOBUFS && - !w.measuring.Load() { + if n < 0 && err != unix.EINTR && err != unix.EAGAIN && err != unix.ENOBUFS { w.stats.errs.Add(1) } }